Build a Claude agent with MCP¶

"""A minimal MCP server exposing a SQLite query tool over stdio."""

import sqlite3
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("sqlite-orders")

DB_PATH = "data/orders.db"


@mcp.tool()
def list_tables() -> str:
    """List all tables in the orders database."""
    with sqlite3.connect(DB_PATH) as conn:
        rows = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ).fetchall()
    return "\n".join(r[0] for r in rows)


@mcp.tool()
def describe_table(table: str) -> str:
    """Return the schema for a given table."""
    with sqlite3.connect(DB_PATH) as conn:
        rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
    return "\n".join(f"{r[1]} {r[2]}" for r in rows)


@mcp.tool()
def query(sql: str) -> str:
    """Run a read-only SQL query and return up to 50 rows as text.

    Reject anything that isn't a SELECT to keep the model from being
    able to mutate the database.
    """
    if not sql.strip().lower().startswith("select"):
        return "ERROR: only SELECT queries are allowed."
    with sqlite3.connect(DB_PATH) as conn:
        try:
            rows = conn.execute(sql).fetchmany(50)
        except sqlite3.Error as e:
            return f"ERROR: {e}"
        cols = [d[0] for d in conn.execute(sql).description]
    header = " | ".join(cols)
    body = "\n".join(" | ".join(str(c) for c in r) for r in rows)
    return f"{header}\n{body}"


if __name__ == "__main__":
    mcp.run()

import os
import asyncio
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


SYSTEM = """You are a helpful data assistant.

You have access to two tool servers:
1. sqlite-orders: query a SQLite database of customer orders.
2. filesystem: read files from a sandboxed data directory.

Strategy:
- For questions about orders, use the sqlite tools.
- For questions about files in data/, use the filesystem tools.
- If unsure, list tables / list files first.
- Always cite your evidence.
"""


client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])


async def collect_tools(session: ClientSession):
    tools = await session.list_tools()
    return [{
        "name": t.name,
        "description": t.description or "",
        "input_schema": t.inputSchema,
    } for t in tools.tools]


async def run(question: str, max_steps: int = 10):
    sqlite_params = StdioServerParameters(command="python", args=["sqlite_mcp.py"])
    fs_params = StdioServerParameters(
        command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "./data"]
    )

    async with stdio_client(sqlite_params) as (sqlite_r, sqlite_w), \
               ClientSession(sqlite_r, sqlite_w) as sqlite_sess, \
               stdio_client(fs_params) as (fs_r, fs_w), \
               ClientSession(fs_r, fs_w) as fs_sess:

        await sqlite_sess.initialize()
        await fs_sess.initialize()

        tools_sqlite = await collect_tools(sqlite_sess)
        tools_fs = await collect_tools(fs_sess)

        # Tag each tool with the server it lives on so we can dispatch later.
        all_tools = (
            [{**t, "_server": "sqlite"} for t in tools_sqlite] +
            [{**t, "_server": "fs"} for t in tools_fs]
        )
        servers = {"sqlite": sqlite_sess, "fs": fs_sess}

        # Strip _server before sending to Claude
        api_tools = [{k: v for k, v in t.items() if not k.startswith("_")} for t in all_tools]

        messages = [{"role": "user", "content": question}]

        for step in range(max_steps):
            resp = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                system=SYSTEM,
                tools=api_tools,
                messages=messages,
            )
            messages.append({"role": "assistant", "content": resp.content})

            if resp.stop_reason == "end_turn":
                # Final text answer
                for block in resp.content:
                    if hasattr(block, "text"):
                        print(block.text)
                return

            if resp.stop_reason == "tool_use":
                tool_results = []
                for block in resp.content:
                    if block.type == "tool_use":
                        # Find which server owns this tool
                        owner = next(t["_server"] for t in all_tools if t["name"] == block.name)
                        print(f"[tool] {owner}::{block.name}({block.input})")
                        result = await servers[owner].call_tool(block.name, block.input)
                        text = "\n".join(
                            c.text for c in result.content if hasattr(c, "text")
                        )
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": text,
                        })
                messages.append({"role": "user", "content": tool_results})
                continue

            print(f"Unexpected stop reason: {resp.stop_reason}")
            return

        print("Hit max_steps without finishing.")


if __name__ == "__main__":
    import sys
    question = " ".join(sys.argv[1:]) or "Who has spent the most across all orders, and what's their total?"
    asyncio.run(run(question))