Build a safe MCP server in Python with FastMCP

Building an MCP server in Python takes a few minutes with FastMCP. Building one you would point at real data takes longer. The extra time goes into what happens when the model sends a query that never ends, asks for every row, or reads a file it shouldn't.

This post builds a small server: one read-only DuckDB query tool with a row cap, a timeout and a call log. I ran every line of it, and the output is below. One of my first assumptions failed the test: the framework's timeout did not stop the query. Then I show the production version, the MCP server in my own system, AEGIS, and the rules it adds once other people's data is involved.

What you need

uv venv && uv pip install fastmcp duckdb

That installed FastMCP 4.0.10, the MCP Python SDK 2.2.0 and DuckDB 1.5.6 on Python 3.13. FastMCP 4.0.0 (31 August 2026) was a rewrite on the MCP Python SDK v2 for the 2026-07-28 protocol revision. If you follow a FastMCP 2 or 3 tutorial, most code still works. I hit two deprecation warnings, covered at the end.

The server

Here is the whole file, exactly as I ran it:

"""A read-only DuckDB MCP server: row cap, timeout, call log."""
import json, os, threading, time

import duckdb
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

DB_PATH = os.environ.get("DUCKDB_PATH", "shop.duckdb")
MAX_ROWS = int(os.environ.get("MAX_ROWS", "200"))
TIMEOUT_S = float(os.environ.get("QUERY_TIMEOUT_S", "5"))
LOG_PATH = os.environ.get("CALL_LOG", "calls.jsonl")

# One read-only connection. Then shut the doors a read-only file leaves open:
# reading other files (read_csv('/etc/passwd')), installing extensions, and
# the model SETting either back.
db = duckdb.connect(DB_PATH, read_only=True)
db.execute("SET enable_external_access = false")
db.execute("SET autoinstall_known_extensions = false")
db.execute("SET autoload_known_extensions = false")
db.execute("SET lock_configuration = true")

mcp = FastMCP("shop-readonly", mask_error_details=True)


def log(**entry):
    entry["ts"] = time.strftime("%Y-%m-%dT%H:%M:%S")
    with open(LOG_PATH, "a") as f:
        f.write(json.dumps(entry) + "\n")


@mcp.tool(annotations={"readOnlyHint": True})
def list_tables() -> list[dict]:
    """List tables and their columns."""
    rows = db.cursor().execute(
        "SELECT table_name, column_name, data_type FROM information_schema.columns "
        "ORDER BY table_name, ordinal_position").fetchall()
    log(tool="list_tables", rows=len(rows), outcome="ok")
    return [{"table": t, "column": c, "type": d} for t, c, d in rows]


@mcp.tool(annotations={"readOnlyHint": True})
def query(sql: str) -> dict:
    """Run one read-only SQL query. Returns at most MAX_ROWS rows."""
    cur = db.cursor()
    timer = threading.Timer(TIMEOUT_S, cur.interrupt)  # stops the query itself
    started = time.perf_counter()
    timer.start()
    try:
        cur.execute(sql)
        rows = cur.fetchmany(MAX_ROWS + 1)
        cols = [d[0] for d in cur.description]
    except duckdb.InterruptException:
        log(tool="query", sql=sql, outcome="timeout", ms=round((time.perf_counter() - started) * 1000))
        raise ToolError(f"Query stopped after {TIMEOUT_S:g}s. Add a filter or an aggregate.")
    except duckdb.Error as e:
        log(tool="query", sql=sql, outcome="error", error=str(e)[:300])
        raise ToolError(str(e).splitlines()[0][:300])
    finally:
        timer.cancel()
    truncated = len(rows) > MAX_ROWS
    rows = rows[:MAX_ROWS]
    log(tool="query", sql=sql, outcome="ok", rows=len(rows), truncated=truncated,
        ms=round((time.perf_counter() - started) * 1000))
    return {"columns": cols, "rows": [list(map(str, r)) for r in rows], "truncated": truncated}


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

What each part is for:

• read_only=True stops writes to the database file. That is all it stops.
• enable_external_access = false stops SQL from reading other files. Without it, a read-only connection happily ran read_csv('/etc/passwd') in my test. If the server runs as you, that includes your cloud credentials.
• lock_configuration = true stops the model from running SET to undo the line above.
• fetchmany(MAX_ROWS + 1) caps the rows and tells the model when it hit the cap, so it can add a filter instead of guessing.
• cur.interrupt() on a timer is the timeout. More on why below.
• mask_error_details=True hides stack traces. Only the ToolError messages I wrote, cut to 300 characters, reach the model.
• calls.jsonl records every call: the SQL, the outcome, the row count and the time taken.

How to build an MCP server in Python: test it before Claude does

You don't need Claude to test this. FastMCP ships a client that launches the server over stdio, the same way Claude Code does:

import asyncio
from pathlib import Path
from fastmcp import Client

async def main():
    async with Client(Path("server.py")) as c:
        r = await c.call_tool("query", {"sql": "DELETE FROM orders"}, raise_on_error=False)
        print(r.is_error, r.content[0].text)

asyncio.run(main())

I created a 50,000-row orders table and sent seven calls:

Call	Result	Time
list_tables	4 columns of orders	112 ms (first call, includes startup)
Group-by on kind	3 rows, correct	7 ms
SELECT * FROM orders	200 rows, truncated: true	3 ms
DELETE FROM orders	Error: "attached in read-only mode"	3 ms
read_csv('/etc/passwd')	Error: "file system operations are disabled by configuration"	9 ms
SET enable_external_access = true	Error: "the configuration has been locked"	3 ms
Cross join of 100M × 100K rows	Error: "Query stopped after 5s"	5,002 ms

And the log has a line for each one:

{"tool": "query", "sql": "SELECT * FROM orders", "outcome": "ok", "rows": 200, "truncated": true, "ms": 1, "ts": "2026-10-01T23:50:17"}
{"tool": "query", "sql": "DELETE FROM orders", "outcome": "error", "error": "Invalid Input Error: Cannot execute statement of type \"DELETE\" on database \"shop\" which is attached in read-only mode!", "ts": "2026-10-01T23:50:17"}
{"tool": "query", "sql": "SELECT count(*) FROM range(100000000) a, range(100000) b WHERE a.range + b.range = 7", "outcome": "timeout", "ms": 5000, "ts": "2026-10-01T23:50:22"}

The timeout that didn't time out

FastMCP's @mcp.tool takes a timeout argument, so my first version used that. I tested it on its own: a sync tool with timeout=2 running a DuckDB query that takes about 10 seconds.

client got error=False after 10.2s: done
slow.log: query finished after 10.1s

No timeout. The query ran to the end and the client got a normal result. The FastMCP tools docs say why, in one line: "Inline sync calls have no cancellation checkpoints, so timeout cannot interrupt them." A database driver call is exactly that kind of call.

The lesson goes beyond FastMCP. A timeout in the framework can, at best, stop waiting for the result. It cannot stop the work. The query keeps holding a core and memory. So the limit belongs where the work runs. For DuckDB that is cursor.interrupt(), as above. For Postgres it is statement_timeout on the role, which the Postgres MCP server post sets.

Connect it to Claude Code

Per the Claude Code MCP docs, options go before the name and the server command after --:

claude mcp add --transport stdio shop \
  --env DUCKDB_PATH=/data/shop.duckdb --env MAX_ROWS=200 \
  -- /path/to/.venv/bin/python /path/to/server.py

Use the venv's own Python so the server gets the packages it was tested with. Configuration comes from environment variables, which is also where the MCP spec says a stdio server should get any credentials.

The production version: how AEGIS serves its tools

The server above is the right shape for one person and one database. My own system, AEGIS, serves its tools over MCP to headless Claude Code and Kimi CLI runs, and to Claude Desktop. That changes the threat. A run reads untrusted content, such as web pages. If it is talked into misbehaving, it will call whatever it can reach.

The AEGIS server is not FastMCP. It is a hand-written FastAPI endpoint that speaks JSON-RPC for the 2025-06-18 protocol revision. Writing it by hand meant every request passes through code I control: authentication, which tools exist, and what each call is allowed to do. Here is what it adds, from api/routes/mcp_server.py:

1. Off by default. Every method returns 403 until a setting turns the server on. It also returns 403 while the API's authentication is switched off, unless a separate setting accepts that risk by name.
2. One list for listing and calling. The served tools are derived once per request, and that one list answers both tools/list and tools/call. A tool that isn't listed can't be called.
3. Some tools are never served. Eleven tools are removed whatever the agent's tool set says. That includes every tool that starts another agent run, because a run that can start runs has no depth limit.
4. An allow-list, not a deny-list. On the gated endpoint, only 29 named tools run without a person approving them. A tool added tomorrow, or renamed, needs approval automatically. A CI test reads each listed tool's source and fails if one gains a write.
5. Approval fails closed. A write raises a card in Slack or the admin panel. Only an explicit approve runs it. A timeout, an error or a malformed answer counts as a deny. Each approval covers exactly one call with exactly the arguments shown on the card.
6. Caps on everything returned. Tool results are cut at 64 KB, errors at 300 characters. Calls time out after 30 seconds unless a tool has its own limit.
7. Every call recorded. Each call writes one row through record_tool_call, whatever the outcome: allowed, refused, blocked by the gate, timed out or failed. The log line records the argument names but not the values, because a value can hold personal text.
8. Stateless. No session ID is issued, and one sent by a client is ignored, so there is no session to hijack.

Authentication (the per-run token, and why the client's own permission prompts weren't enough) is its own subject. The Postgres MCP server post covers the named-tools-over-SQL part of the design.

Items 6 and 7 are the toy server's row cap and call log, grown up. The rest exist because more than one person, or more than one run, can reach the server.

Two FastMCP 4 gotchas

• Client("server.py") still works but warns: inferring stdio from a string is deprecated and will be removed in FastMCP 5. Pass Path("server.py") or a StdioTransport.
• Reading tool.annotations.readOnlyHint on the client warns that MCP SDK v2 renamed it read_only_hint. Passing {"readOnlyHint": True} to the decorator still worked; tools/list reported it as True.

So: what makes an MCP server safe?

Not the framework. FastMCP made the protocol part trivial: two decorators and mcp.run(). Every line that made the server safe was DuckDB settings, a timer and a log file. In production, it is the allow-list and the approval gate.

Before you connect any Python MCP server to real data, check these:

1. The database refuses writes on its own: a read-only connection or a read-only role.
2. SQL cannot read files or load extensions: enable_external_access = false and a locked configuration.
3. Every query has a timeout enforced by the database, not only by the framework.
4. Results have a row cap, and the model is told when it hit it.
5. Every call is logged with its arguments and its outcome.

The broader guardrails (views, PII, prompt injection through the data) are in how to connect an LLM to your database safely.

Local-First Analytics doesn't cover MCP. It covers the data a server like this exposes: DuckDB over Parquet as a warehouse (Chapter 5), and turning plain-English questions into DuckDB SQL with a local model (Chapter 12). Chapter 1 is free to read; the rest is on Amazon.

If you are building an MCP server over data your team depends on and want the design checked before an agent finds the gaps, that is work I do. If your server already passes the five checks above, you probably don't need me.

More in this series: DuckDB MCP server, MCP security and authentication and Postgres MCP server.