DuckDB MCP Server

1.1.0 · active · verified Fri Apr 17

mcp-server-duckdb is a Python library that provides a Micro-Capability Platform (MCP) server exposing a DuckDB database via HTTP. It allows clients to execute SQL queries against a DuckDB instance, supporting both read and write operations, and a read-only mode. The current version is 1.1.0, with an active, feature-driven release cadence.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to start the `mcp-server-duckdb` server programmatically and then interact with it using HTTP requests. It creates a simple table, inserts data, and queries it. For persistent databases, specify a file path with `--db-path` when starting the server.

import os
import requests
from subprocess import Popen, PIPE, TimeoutExpired
import time

# Start the server in a separate process
# For simplicity, we use a temporary in-memory database here
# In production, specify a persistent .duckdb file, e.g., --db-path my_database.duckdb
print('Starting mcp-server-duckdb...')
server_process = Popen(['mcp-server-duckdb', '--port', '8001'], stdout=PIPE, stderr=PIPE)
time.sleep(2) # Give the server a moment to start

# Example: Send a query
try:
    # Create a table
    response = requests.post(
        'http://localhost:8001/queries',
        json={'query': "CREATE TABLE users (id INTEGER, name VARCHAR);"}
    )
    response.raise_for_status()
    print('CREATE TABLE response:', response.json())

    # Insert data
    response = requests.post(
        'http://localhost:8001/queries',
        json={'query': "INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob');"}
    )
    response.raise_for_status()
    print('INSERT INTO response:', response.json())

    # Select data
    response = requests.post(
        'http://localhost:8001/queries',
        json={'query': "SELECT * FROM users;"}
    )
    response.raise_for_status()
    print('SELECT * response:', response.json())

except requests.exceptions.RequestException as e:
    print(f"Error communicating with server: {e}")
finally:
    # Terminate the server process
    print('Stopping mcp-server-duckdb...')
    server_process.terminate()
    try:
        stdout, stderr = server_process.communicate(timeout=5)
        print('Server stdout:', stdout.decode())
        print('Server stderr:', stderr.decode())
    except TimeoutExpired:
        server_process.kill()
        stdout, stderr = server_process.communicate()
        print('Server (killed) stdout:', stdout.decode())
        print('Server (killed) stderr:', stderr.decode())

view raw JSON →