query
Execute SQL queries on DuckDB databases to retrieve, analyze, or modify data stored locally, in memory, or in the cloud.
Instructions
Use this to execute a query on the DuckDB database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SQL query to execute that is a dialect of DuckDB SQL |
Implementation Reference
- src/mcp_server_duckdb/server.py:128-135 (handler)Handler logic within handle_tool_call that executes the 'query' tool by invoking DatabaseClient.query on the provided SQL query.
if name == "query": if arguments is None: return [ types.TextContent(type="text", text="Error: No query provided") ] tool_response = db_client.query(arguments["query"]) return [types.TextContent(type="text", text=str(tool_response))] - JSON Schema definition for the 'query' tool input, specifying the required 'query' string parameter.
types.Tool( name="query", description="Use this to execute a query on the DuckDB database", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL query to execute that is a dialect of DuckDB SQL", }, }, "required": ["query"], }, ), - src/mcp_server_duckdb/server.py:94-117 (registration)Registration of the 'query' tool via the @server.list_tools() decorator and handle_list_tools function that returns the tool definition.
@server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ logger.info("Listing tools") return [ types.Tool( name="query", description="Use this to execute a query on the DuckDB database", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL query to execute that is a dialect of DuckDB SQL", }, }, "required": ["query"], }, ), ] - DatabaseClient.query method, the core helper function that executes the SQL query and formats the results, called by the tool handler.
def query(self, query: str) -> str: try: return self._execute(query) except Exception as e: raise ValueError(f"❌ Error executing query: {e}")