execute_query
Execute SQL queries on Trino and Iceberg data sources and receive results in a clear, readable format.
Instructions
Execute a SQL query and return results in a readable format
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to execute |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/trino_client.py:67-84 (helper)The actual implementation of execute_query that connects to Trino, executes the SQL, and returns results as JSON or a success message.
def execute_query(self, query: str) -> str: """Execute a SQL query against Trino and return results as a formatted string. Args: query (str): The SQL query to execute. params (Optional[dict]): Dictionary of query parameters with primitive types. Returns: str: JSON-formatted string containing query results or success message. """ cur: trino.dbapi.Cursor = self.client.cursor() cur.execute(query) if cur.description: return json.dumps( [dict(zip([col[0] for col in cur.description], row, strict=True)) for row in cur.fetchall()], default=str, ) return "Query executed successfully (no results to display)" - src/server.py:128-138 (handler)The MCP tool handler function that registers 'execute_query' as a tool, accepting a SQL query string parameter and delegating to trino_client.execute_query.
@mcp.tool(description="Execute a SQL query and return results in a readable format") def execute_query(query: str = Field(description="The SQL query to execute")) -> str: """Execute a SQL query and return formatted results. Args: query: The SQL query to execute Returns: str: Query results formatted as a JSON string """ return client.execute_query(query) - src/server.py:128-138 (registration)Registration of the 'execute_query' tool via the @mcp.tool decorator with description 'Execute a SQL query and return results in a readable format'.
@mcp.tool(description="Execute a SQL query and return results in a readable format") def execute_query(query: str = Field(description="The SQL query to execute")) -> str: """Execute a SQL query and return formatted results. Args: query: The SQL query to execute Returns: str: Query results formatted as a JSON string """ return client.execute_query(query)