Claude Code CLI
     ↓ calls tool
MCP Server (localhost)
     ↓ executes query
Database (SQLite / PostgreSQL / MongoDB)
     ↓ returns results
Claude Code CLI
     ↓ shows answer

#!/usr/bin/env python3
import sqlite3
import json
import sys
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("sqlite-database")

# Configure your database path
DB_PATH = "./project.db"

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="query_database",
            description="Execute a SELECT query on the SQLite database and return results",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL SELECT query to execute"},
                    "limit": {"type": "integer", "description": "Max rows to return (default 50)"}
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="get_schema",
            description="Get the complete database schema — all tables, columns, and types",
            inputSchema={"type": "object", "properties": {}}
        ),
        types.Tool(
            name="insert_record",
            description="Insert a new record into a table",
            inputSchema={
                "type": "object",
                "properties": {
                    "table": {"type": "string"},
                    "data": {"type": "object", "description": "Column:value pairs to insert"}
                },
                "required": ["table", "data"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    if name == "query_database":
        query = arguments["query"]
        limit = arguments.get("limit", 50)
        # Safety: only allow SELECT
        if not query.strip().upper().startswith("SELECT"):
            return [types.TextContent(type="text", text="Error: Only SELECT queries are allowed")]
        cursor.execute(query + f" LIMIT {limit}")
        rows = [dict(row) for row in cursor.fetchall()]
        return [types.TextContent(type="text", text=json.dumps(rows, indent=2))]

    elif name == "get_schema":
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = [row[0] for row in cursor.fetchall()]
        schema = {}
        for table in tables:
            cursor.execute(f"PRAGMA table_info({table})")
            columns = [{"name": col[1], "type": col[2]} for col in cursor.fetchall()]
            schema[table] = columns
        return [types.TextContent(type="text", text=json.dumps(schema, indent=2))]

    elif name == "insert_record":
        table = arguments["table"]
        data = arguments["data"]
        columns = ", ".join(data.keys())
        placeholders = ", ".join(["?" for _ in data])
        cursor.execute(f"INSERT INTO {table} ({columns}) VALUES ({placeholders})", list(data.values()))
        conn.commit()
        return [types.TextContent(type="text", text=f"Inserted record with ID {cursor.lastrowid}")]

    conn.close()

async def main():
    async with stdio_server() as (read, write):
        await app.run(read, write, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

import asyncpg

# Replace the sqlite3 connection with:
conn = await asyncpg.connect(
    host=os.getenv("PG_HOST", "localhost"),
    database=os.getenv("PG_DB", "myapp"),
    user=os.getenv("PG_USER", "postgres"),
    password=os.getenv("PG_PASSWORD")
)
rows = await conn.fetch(query)
results = [dict(row) for row in rows]

from pymongo import MongoClient

client = MongoClient(os.getenv("MONGO_URI", "mongodb://localhost:27017"))
db = client["myapp"]

# Tool: find_documents
def find_documents(collection_name, filter_query, limit=50):
    collection = db[collection_name]
    docs = list(collection.find(filter_query).limit(limit))
    # Convert ObjectId to string for JSON serialization
    for doc in docs:
        doc["_id"] = str(doc["_id"])
    return docs