execute_query
Execute SQL queries against Trino and Iceberg to retrieve data and return formatted results for analysis.
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 (handler)The actual implementation of execute_query that runs SQL queries against Trino using trino.dbapi.Cursor, fetches results, and returns them as JSON-formatted string.
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 (registration)MCP tool registration for execute_query using the @mcp.tool decorator. This registers the tool with the MCP server, defining the external interface including the query parameter with its description.
@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-129 (schema)Schema definition for the execute_query tool's input parameter using Pydantic Field to provide the description for the query parameter.
@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: