DuckDB MCP Server

Instructions

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))]

• src/mcp_server_duckdb/server.py:102-115 (schema)

  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"],
              },
          ),
      ]

• src/mcp_server_duckdb/database.py:221-226 (helper)

  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}")