Claude Agent SDK in Production, Part 6: Custom Tools: Give the Analyst a Database

"314 rows removed." My analyst announced it in bold, cheerfully, four seconds after I asked. It had just run a DELETE against the company database with the same tool it uses to list files, and nothing in the system so much as cleared its throat. That run is real, it's later in this page, and it's the reason this part exists. By the end, the same request dies with SQL error: attempt to write a readonly database, because the analyst's database access will go through a tool you wrote, with the guardrail built into the tool itself. Welcome to Act II: four parts of taking the scissors away without taking the power away.

Act I ended with a complete product that has one data source: whatever CSVs land on the desk. Real analysts query systems: the sales database, the CRM, the warehouse. This part builds that bridge with the SDK's extension mechanism, the @tool decorator plus an in-process MCP server, and it teaches the single most underrated skill in agent building along the way: tool descriptions are prompts. If you're joining for Act II, the app so far is the part-05-sessions folder; clone it and you're caught up.

Beanline gets a real database

Six parts in, it's time to admit the CSVs were training wheels. A coffee chain's live numbers don't live in three files somebody uploads; they live in a database that's already there when the conversation starts. So Beanline's data moves into SQLite: same deterministic rows as the CSVs (planted duplicate and all), now with types, keys, and indexes. The builder script ships in the repo:

DDL = """
CREATE TABLE stores (
    store_id TEXT PRIMARY KEY,
    name     TEXT NOT NULL,
    city     TEXT NOT NULL,
    opened   TEXT NOT NULL
);
CREATE TABLE products (
    product_id TEXT PRIMARY KEY,
    name       TEXT NOT NULL,
    category   TEXT NOT NULL,
    unit_price REAL NOT NULL
);
CREATE TABLE sales (
    date       TEXT NOT NULL,
    store_id   TEXT NOT NULL REFERENCES stores(store_id),
    product_id TEXT NOT NULL REFERENCES products(product_id),
    units      INTEGER NOT NULL,
    revenue    REAL NOT NULL
);
CREATE INDEX idx_sales_date ON sales(date);
CREATE INDEX idx_sales_store ON sales(store_id);
"""

Build it once, from backend/:

uv run python data/build_beanline_db.py
# wrote data/beanline.db {'stores': 6, 'products': 12, 'sales': 11081}

Note where it lands: backend/data/, next to the server code, not in any workspace. The database is the company's, not the conversation's. That placement decision is about to do a lot of work.

The crowbar: watch Bash do it first

Before writing any new code, be honest about something: the analyst can already query this database. Bash is a universal crowbar. Upload beanline.db to a desk on the Part 5 app (it's a file like any other) and ask:

It listed the tables, checked the schema, wrote the SUM, and got it right: $206,815.80, nine seconds, $0.0290. If Bash can do this, why build anything? Ask one more question in the same conversation. This is the run from the first line of this page:

I verified afterward: zero March rows left for the Airport store in that desk copy. Three problems live in this pair of screenshots, and only one of them is the deletion. One: the desk copy is a copy; it goes stale the moment real sales continue, and every conversation would need its own upload. Two: the agent had to spelunk the schema before every query, on your token bill. Three: Bash access to a database file is read-write access, full stop. You can write "please be careful" in a prompt; rm and DELETE don't read prompts. In Part 1's terms: still running with scissors, now in a china shop.

The fix isn't to scold the agent. It's to give it a better door: a tool that reaches the live database, knows the schema by heart, and physically cannot write. That's a custom tool, and in this SDK custom tools are MCP servers.

Your first @tool

New file, backend/app/tools.py. Before the function, the part that matters most, and it isn't code:

DB_PATH = Path("data/beanline.db").resolve()
MAX_ROWS = 200

# The description is a prompt: the model READS it to decide when to reach
# for this tool and how to shape its SQL. Table names, column names, and
# the read-only warning all live here so no question has to guess them.
QUERY_DESCRIPTION = """Run one read-only SQL query against Beanline's live sales database (SQLite).

Tables:
  stores(store_id, name, city, opened)
  products(product_id, name, category, unit_price)
  sales(date, store_id, product_id, units, revenue)  -- one row per product per store per day

Prefer SQL aggregation (SUM, GROUP BY) over selecting raw rows. Results are
capped at 200 rows. The connection is read-only: INSERT, UPDATE, DELETE, and
schema changes fail with an error."""

Nobody executes that description. The model reads it, the same way it reads your system prompt, and decides from it when to reach for the tool and what to put in the sql argument. I can prove the sentence with receipts, because I ran the same March question against two versions of this tool in a terminal experiment. With a lazy description that named only the tables, the agent guessed at columns (column5, price, amount), ate three no such column errors, rescued itself with PRAGMA table_info, and billed $0.0217. With the columns spelled out: correct SQL, first try, $0.0055.

Now the tool itself. The decorator takes a name, that description, and an input schema; the function takes the arguments as a dict and returns MCP-shaped content:

@tool("query_database", QUERY_DESCRIPTION, {"sql": str})
async def query_database(args: dict) -> dict:
    try:
        con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
        try:
            cursor = con.execute(args["sql"])
            columns = [c[0] for c in cursor.description or []]
            rows = cursor.fetchmany(MAX_ROWS)
            clipped = cursor.fetchone() is not None
        finally:
            con.close()
    except sqlite3.Error as exc:
        return {
            "content": [{"type": "text", "text": f"SQL error: {exc}"}],
            "isError": True,
        }
    header = "| " + " | ".join(columns) + " |"
    divider = "| " + " | ".join("---" for _ in columns) + " |"
    body = ["| " + " | ".join(str(v) for v in row) + " |" for row in rows]
    note = [f"({MAX_ROWS}-row cap reached; aggregate instead)"] if clipped else []
    table = "\n".join([header, divider, *body, *note]) if columns else "(no rows)"
    return {"content": [{"type": "text", "text": table}]}

Four deliberate choices, top to bottom. mode=ro is the entire security story: the SQLite connection is opened read-only at the URI level, so writes fail inside SQLite itself, no matter how politely they're phrased. Errors return as content, not exceptions: a failed query becomes text the model reads and adapts to, exactly like a failed Bash command in Part 1. MAX_ROWS caps the blast radius of a careless SELECT *, and the cap message tells the model what to do instead (aggregate); even the error path is a prompt. Rows come back as a markdown table, because the model parses tables well and, two sections from now, the UI renders them for free.

A second tool, because one tool is a trick and two is a pattern. This one takes no arguments and exists for the questions the description can't anticipate:

@tool(
    "get_schema",
    "Return the Beanline database schema: every CREATE TABLE and CREATE INDEX "
    "statement, verbatim. Call this before writing SQL if you are unsure of "
    "a column name.",
    {},
)
async def get_schema(args: dict) -> dict:
    con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
    try:
        statements = [
            row[0]
            for row in con.execute(
                "SELECT sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY name"
            )
        ]
    finally:
        con.close()
    return {"content": [{"type": "text", "text": ";\n".join(statements) + ";"}]}


beanline_server = create_sdk_mcp_server(
    name="beanline", tools=[query_database, get_schema]
)

That last statement is the one that turns two functions into infrastructure.

An MCP server that never leaves home

MCP, the Model Context Protocol, is the standard for handing tools to models: a server describes its tools, a client (the agent) discovers and calls them. Most MCP servers are separate processes or remote services. create_sdk_mcp_server is the SDK's shortcut: it wraps your functions in the same protocol, but the "server" lives inside your process and a tool call is a function call. Same interface the whole ecosystem speaks, none of the operational overhead, which is why this part can stay this short.

Wiring it into the app is one line on the options, in build_options:

def build_options(workspace: Path, session_id: str | None) -> ClaudeAgentOptions:
    """Part 5's options plus one line: mcp_servers. The in-process server
    adds two mcp__beanline__* tools to the toolbox; tools= still curates
    only the built-ins."""
    return ClaudeAgentOptions(
        cwd=str(workspace),
        tools=["Read", "Glob", "Grep", "Bash", "Write"],
        mcp_servers={"beanline": beanline_server},
        permission_mode="bypassPermissions",
        model=MODEL,
        include_partial_messages=True,
        system_prompt={"type": "preset", "preset": "claude_code", "append": ANALYST_PROMPT},
        resume=session_id,
    )

One behavior here surprised me enough to verify twice. Since Part 1, tools=[...] has meant "these tools exist, nothing else". So does adding an MCP server require adding its tools to that list? No: the init handshake for this exact options object reports the toolbox as ['Bash', 'Glob', 'Grep', 'Read', 'Write', 'mcp__beanline__get_schema', 'mcp__beanline__query_database']. tools= curates the built-ins; MCP servers add their own entries alongside. I even tried listing an MCP name in tools=, misspelled and spelled right; both were ignored. The two dials don't touch.

The name is load-bearing

Look at those toolbox entries again: your function was called query_database, but the agent knows it as mcp__beanline__query_database. The rule: mcp__<server key>__<tool name>, where the server key is whatever string you used in the mcp_servers dict. Rename the key and every tool name changes with it.

Right now, under bypassPermissions, the name is only a label. The moment permissions matter, it becomes an address. You can prove that today with a 20-line probe (it's in the repo): run the app's options in default mode instead, with the tool auto-approved via allowed_tools, and misspell one letter:

    options = ClaudeAgentOptions(
        mcp_servers={"beanline": beanline_server},
        tools=["Read"],
        allowed_tools=allowed,
        permission_mode="default",  # no bypass: the allow list does the work
        model="claude-haiku-4-5",
    )

The refusal is Part 1's permission wall wearing a new name: the rule mcp__beanline__query_databse matches nothing, so the real tool was never granted. Two more spellings that do work, both verified: mcp__beanline__* (every tool on the server) and plain mcp__beanline (the whole server). File those away. In the next part, permission rules become the center of the story, and they match on exactly these names.

Ask, and watch the new badge

The backend is done. Boot it and ask the March question in a brand-new conversation with an empty desk, because that's the point now: no uploads, no sample data, and the analyst still knows the company's numbers.

Compare it with the crowbar run that opened this part: same question, same answer, but one tool call instead of three, no schema spelunking, and against the live database instead of a desk copy. The description did the navigating.

Two small pieces of polish made that screenshot legible, and both are worth their diff. First, MCP tool results arrive shaped differently from built-in ones: not a string but a list of content blocks ([{"type": "text", "text": ...}]). Left alone, every badge would display JSON packaging instead of the table inside it. The translator gains one function, and the tool_result branch wraps its content in it:

def flatten(content: object) -> object:
    """Custom (MCP) tool results arrive as a list of content blocks, e.g.
    [{"type": "text", "text": "..."}]. Unwrap the text so badges read like
    output instead of JSON. Built-in tools pass through untouched."""
    if isinstance(content, list):
        texts = [
            block["text"]
            for block in content
            if isinstance(block, dict) and block.get("type") == "text"
        ]
        if texts:
            return "\n".join(texts)
    return content

Second, the frontend. Part 3's toolLabel map was built with a default branch for exactly this day, and today it learns to read MCP names instead of shouting them raw:

    default: {
      // Custom tools arrive as mcp__<server>__<tool>. Give ours real
      // labels; any future MCP tool still gets a readable fallback.
      const mcp = block.name.match(/^mcp__(.+?)__(.+)$/);
      if (mcp?.[2] === "query_database") return `Querying the ${mcp[1]} database`;
      if (mcp?.[2] === "get_schema") return `Reading the ${mcp[1]} schema`;
      if (mcp) return `${mcp[1]}: ${mcp[2].replaceAll("_", " ")}`;
      return block.name;
    }

That's the entire frontend change for this part. Nine lines. The event vocabulary didn't change at all: a custom tool call is still a tool_use_start parcel with a stranger name, and Part 3's client was built to shrug at strangers.

Break it on purpose: ask it to write

The keycard says read-only. Time to rattle the handle. Ask the finished app, in plain user language, to add a sale:

Three details in that screenshot repay a close look. The agent's SQL is competent: it called get_schema first (shape contrast earning its keep), noticed sales wants ids rather than names, and wrote an INSERT ... SELECT to resolve them. Doesn't matter: SQLite's mode=ro refuses at the engine, and the error rides back as ordinary tool output. And then the reply: "The tool explicitly states 'The connection is read-only: INSERT, UPDATE, DELETE, and schema changes fail with an error.'" The model is quoting the description. Prompts, all the way down.

One honest wrinkle, so your mental model stays true: the badge wears a green check, not a red cross. The isError flag a custom tool returns doesn't currently propagate into the stream's ToolResultBlock.is_error, so the wire sees a successful call whose text is an error message. The model reads the text either way, which is what governs behavior; but if you want red badges for SQL errors someday, you'll key off the content, not the flag. Noted, filed, moving on.

Two new house rules, one of them earned the hard way

The system prompt does gain two lines this part. Here's the full set, additions at the bottom:

ANALYST_PROMPT = """You are the Beanline data analyst. House rules for every answer:

- When a chart would help, create it with matplotlib and save it as a PNG file
  in the working directory (plt.savefig(..., dpi=150), never plt.show()).
- Write your findings to report.md: a one-line headline, the key numbers as a
  markdown table, then a short interpretation. Create or overwrite it each turn.
- Keep the chat reply brief: the main numbers and the files you produced.
  Prefer tables over prose for numbers.
- Company-wide numbers live in the Beanline database: use the query_database
  tool for them. Files on your desk are user uploads, not the source of truth.
- Every number you state must come from a query result or script output.
  Never do arithmetic in your head, not even totals."""

The first addition is routing: tools tell the model how, the prompt tells it when. The second one I didn't plan. My first two takes of this part's finale produced per-store tables that were correct to the penny, and then volunteered a company-wide total the model had summed in its head: $584,502.90 one take, $654.7K the next. Both wrong (the real Q1 is $584,102.80). Same failure, twice, in the prose while the SQL-backed table sat there perfect. The fix is Part 4's acceptance-criteria rule applied to arithmetic: numbers come from query results, full stop. After the rule: every retake's totals exact, including the one recorded for this part's video. (An agent confidently decorating true tables with invented totals is also the trust problem in miniature; Part 11 hires a reviewer for it, and Part 13 builds the eval suite that catches regressions like this before your readers do.)

The dessert: four systems, one question

Now the run this part opened with. Fresh conversation, empty desk: "Compare each store's Q1 vs Q2 revenue from the database and chart the difference." Watch the badges in the hero screenshot at the top of this page tell the story in order: get_schema, a query_database that assumed 2025 and got zero rows, a check of the actual date range, the corrected query, then matplotlib via Bash, then report.md via Write. Custom tool, built-in tools, artifacts panel, and the house rules, all firing in a single $0.0544 turn, and every number in the report matched my ground-truth SQL exactly.

That mid-run stumble ("The database appears to have no sales data yet... Ah, it's 2026 data") is worth savoring rather than editing out. The tool returned an honest empty table, the model treated it as information, introspected the data, and corrected course. You built that resilience in Part 1 without knowing it; today it has better tools to be resilient with.

The cost ritual

Today's ledger, all real runs from building this part:

The row to remember is the cheapest one: $0.0062 bought an unsupervised DELETE. Guardrails aren't priced by what they cost you; they're priced by what their absence lets a four-second turn do. And the description experiment is the same lesson in the other direction: one paragraph of writing made every future query of this tool roughly four times cheaper. Words are infrastructure now. The billing mechanics live on the concepts page.

Commit it, from the project root:

git add backend frontend
git commit -m "part 6: custom tools - the beanline database, read-only by construction"

Your analyst now queries a live database through a keycard you cut for it. But look at the other tools on its belt: Bash still runs anything, Write still writes anywhere on the desk, and the crowbar demo needed no permission because under bypassPermissions nothing does. The keycard was Act II's first rung; the scissors are still in hand. In Part 7, risky tool calls stop mid-turn and wait, genuinely paused, for a human to click Approve or Deny, and Part 1's oldest warning finally starts getting repaid.

The complete, tested code for this part lives in part-06-custom-tools in the companion repo. Code blocks with a GitHub icon link straight to the exact file; "View full file" shows the whole file in place with this section's changes highlighted.