Bekzod Erkinov
Listen to Article
Loading...Table of contents · 9 sections
Build an MCP Server in Python (and Call It From Claude Code)
By the end you have a running read-only SQLite MCP server in Python (two tools and a resource), registered with claude mcp add, plus a CI script that fails the build if Claude Code cannot call it from claude -p. Every command and output below came from my own runs on Windows 11 with Python 3.12.10, mcp 2.3.0 and Claude Code 2.1.295 on 2026-10-10.
Build an MCP server in Python: install the SDK
python -m venv .venv
.venv\Scripts\activate # macOS/Linux: source .venv/bin/activate
pip install "mcp[cli]"
That installed mcp 2.3.0 for me (PyPI). If older tutorials say FastMCP: in 2.x that class was renamed. The migration guide says importing mcp.server.fastmcp now raises ModuleNotFoundError, and the new import is from mcp.server.mcpserver import MCPServer. The decorators (@mcp.tool(), @mcp.resource()) kept their signatures.
I ran both. In a second venv, pip install "mcp<2" gave 1.30.0, and from mcp.server.fastmcp import FastMCP worked. In the 2.3.0 venv the same import failed with a message pointing at the migration guide. The decision rule: new project, use 2.x and MCPServer; v1 code you don't want to touch yet, pin mcp<2.
Two more v2 changes from the same guide bite when you port code: transport settings (host, port) moved from the constructor to run(), and the constructor's positional order changed, so pass everything except name by keyword. The PyPI JSON API lists requires_python as >=3.10 for 2.3.0.
Create the SQLite database the MCP server will query
# seed.py
import sqlite3, pathlib
p = pathlib.Path("shop.db"); p.unlink(missing_ok=True)
c = sqlite3.connect(p)
c.executescript("""
CREATE TABLE customers(id INTEGER PRIMARY KEY, name TEXT NOT NULL, country TEXT);
CREATE TABLE orders(id INTEGER PRIMARY KEY, customer_id INTEGER REFERENCES customers(id), total REAL NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL);
INSERT INTO customers VALUES (1,'Ada','UK'),(2,'Linus','FI'),(3,'Grace','US'),(4,'Dennis','US');
INSERT INTO orders VALUES
(1,1,120.50,'paid','2026-09-01'),(2,1,80.00,'refunded','2026-09-03'),
(3,2,310.00,'paid','2026-09-05'),(4,3,45.25,'paid','2026-09-07'),
(5,3,99.99,'pending','2026-09-09'),(6,4,15.00,'paid','2026-09-10');
""")
c.commit()
c.close()
print('seeded')
Run it with python seed.py. It prints seeded and creates shop.db next to the script; the file is deleted and recreated on every run.
Write the MCP server in Python with MCPServer
# server.py
import logging
import sqlite3
import sys
from contextlib import closing
from pathlib import Path
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
# stderr only: stdout belongs to the JSON-RPC stream under stdio.
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("shopdb")
DB = Path(__file__).with_name("shop.db")
mcp = MCPServer("shopdb", instructions="Read-only access to a small shop SQLite database.")
# Only plain reads pass: no ATTACH, PRAGMA, writes or DDL.
ALLOWED = {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION}
def connect() -> sqlite3.Connection:
# mode=ro: SQLite itself refuses writes, whatever SQL the model sends.
con = sqlite3.connect(f"file:{DB.as_posix()}?mode=ro", uri=True)
con.row_factory = sqlite3.Row
con.set_authorizer(lambda action, *_: sqlite3.SQLITE_OK if action in ALLOWED else sqlite3.SQLITE_DENY)
return con
@mcp.tool()
def run_query(sql: str, limit: int = 20) -> list[dict]:
"""Run one read-only SELECT against the shop database and return rows as dicts."""
log.info("run_query: %s", sql)
try:
with closing(connect()) as con:
rows = con.execute(sql).fetchmany(limit)
except sqlite3.Error as e:
raise ToolError(f"SQLite rejected the query: {e}") from e
return [dict(r) for r in rows]
@mcp.tool()
def top_customers(min_total: float = 0.0, status: str = "paid", limit: int = 3) -> list[dict]:
"""Customers ranked by the sum of their orders with a given status.
Args:
min_total: only include customers whose summed total is at least this.
status: order status to count: paid, pending or refunded.
limit: maximum number of customers to return.
"""
with closing(connect()) as con:
rows = con.execute(
"""SELECT c.name, c.country, ROUND(SUM(o.total), 2) AS spent
FROM orders o JOIN customers c ON c.id = o.customer_id
WHERE o.status = ? GROUP BY c.id HAVING spent >= ?
ORDER BY spent DESC LIMIT ?""",
(status, min_total, limit),
).fetchall()
return [dict(r) for r in rows]
@mcp.resource("schema://tables")
def schema() -> str:
"""CREATE statements for every table in the database."""
with closing(connect()) as con:
return "\n".join(r[0] for r in con.execute("SELECT sql FROM sqlite_master WHERE type='table'"))
if __name__ == "__main__":
if "--http" in sys.argv:
mcp.run(transport="streamable-http", host="127.0.0.1", port=18911)
else:
mcp.run() # stdio by default
run_query accepts arbitrary SQL from a model, so the safety lives in the connection, not the prompt. Two layers: mode=ro makes SQLite refuse writes, and the authorizer callback allows only SELECT, column reads and functions, so PRAGMA and ATTACH are denied too (tested below). What this does not cover: the model can still read every table in the file, and it can still send an expensive query; limit only trims the rows returned, not the work SQLite does. If the database holds anything the model shouldn't see, point the server at a different file or a copy without it. top_customers uses bound parameters and a fixed query, the better tool for the common question.
Note closing(connect()) instead of with connect() as con: in sqlite3 the plain with form only commits or rolls back a transaction and never closes the connection (Python docs), so every tool call would leave a handle open. I re-ran inspect_stdio.py against the closing version and got the same output shown below.
Inspect the server over stdio
Before involving Claude, talk to the server with the SDK's own client. It shows exactly what a client sees.
# inspect_stdio.py
import asyncio, json, sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main(script):
params = StdioServerParameters(command=sys.executable, args=[script])
async with stdio_client(params) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
for t in (await s.list_tools()).tools:
print(t.name, json.dumps(t.input_schema, indent=1))
print([str(x.uri) for x in (await s.list_resources()).resources])
print((await s.read_resource("schema://tables")).contents[0].text[:120], "...")
print(await s.call_tool("top_customers", {"min_total": 100}))
for sql in ("DELETE FROM orders", "PRAGMA table_info(orders)", "ATTACH 'other.db' AS o"):
r = await s.call_tool("run_query", {"sql": sql})
print(r.is_error, r.content[0].text)
asyncio.run(main(sys.argv[1]))
Run python inspect_stdio.py server.py. A note on t.input_schema: in 2.x the model fields are snake_case, and the migration guide lists AttributeError: 'Tool' object has no attribute 'inputSchema' as the symptom of using the v1 spelling. On the wire it's still inputSchema, so if you copy this script into a v1 venv, change t.input_schema to t.inputSchema or it will fail; in 2.3.0 mcp.types.Tool.model_fields['input_schema'].alias printed inputSchema for me.
The schema for top_customers, generated from nothing but type hints and defaults (the script's raw output):
top_customers {
"properties": {
"min_total": {
"default": 0.0,
"title": "Min Total",
"type": "number"
},
"status": {
"default": "paid",
"title": "Status",
"type": "string"
},
"limit": {
"default": 3,
"title": "Limit",
"type": "integer"
}
},
"type": "object",
"title": "top_customersArguments"
}
float became number, int became integer, and parameters with defaults are optional. The run_query schema, printed by the same loop:
run_query {
"properties": {
"sql": {
"title": "Sql",
"type": "string"
},
"limit": {
"default": 20,
"title": "Limit",
"type": "integer"
}
},
"required": [
"sql"
],
"type": "object",
"title": "run_queryArguments"
}
Only sql is required. One thing I expected and didn't get: in my 2.3.0 test the Args: section of the top_customers docstring did not become per-parameter description fields. Printing each property's description gave [None, None, None], and the whole docstring, Args: block included, sat in the tool's top-level description. That is what I saw on this one version; I didn't check it against the SDK docs. If a parameter needs explaining to the model, like the allowed status values, say so in the first docstring lines.
The rest of the output, including three statements the guard should refuse:
['schema://tables']
CREATE TABLE customers(id INTEGER PRIMARY KEY, name TEXT NOT NULL, country TEXT)
CREATE TABLE orders(id INTEGER PRIMARY ...
meta=None content=[TextContent(type='text', text='{\n "name": "Linus",\n "country": "FI",\n "spent": 310.0\n}', annotations=None, meta=None), TextContent(type='text', text='{\n "name": "Ada",\n "country": "UK",\n "spent": 120.5\n}', annotations=None, meta=None)] structured_content={'result': [{'name': 'Linus', 'country': 'FI', 'spent': 310.0}, {'name': 'Ada', 'country': 'UK', 'spent': 120.5}]} is_error=False result_type='complete'
True Error executing tool run_query: SQLite rejected the query: not authorized
True Error executing tool run_query: SQLite rejected the query: not authorized
True Error executing tool run_query: SQLite rejected the query: not authorized
A list[dict] return comes back both as text content and as structured_content wrapped in {'result': ...}. The DELETE, PRAGMA and ATTACH calls each come back as an error result (is_error=True) rather than a crash. The resource read still works because the authorizer allows SELECT on sqlite_master.
Register the server with claude mcp add
claude mcp add --transport stdio shopdb -- C:/path/to/.venv/Scripts/python.exe C:/path/to/server.py
The -- matters: per the Claude Code MCP docs, everything after it goes to the server unchanged, and everything before it is Claude's own flags. Use the absolute path to the venv's Python; a bare python resolves to whatever is on PATH when Claude spawns it, which may not be the interpreter that has mcp installed. Forward slashes worked fine for me on Windows. Here is what I ran (paths shortened to C:/path/to) and what claude mcp get shopdb printed:
Added stdio MCP server shopdb with command: C:/path/to/.venv/Scripts/python.exe C:/path/to/server.py to local config
shopdb:
Scope: Local config (private to you in this project)
Status: ✔ Connected
Type: stdio
Command: C:/path/to/.venv/Scripts/python.exe
Args: C:/path/to/server.py
I used claude mcp get because claude mcp list health-checks every configured server, and with my other servers it didn't finish within two minutes; on a clean setup claude mcp list is the quick check. The default scope is local; use --scope project to write a shared .mcp.json (docs).
Call the MCP server from claude -p
My first headless run used my normal config with several other MCP servers and plugins, so it is an uncontrolled baseline, not a like-for-like comparison: its result record showed num_turns 14, total_cost_usd 0.383 and duration_ms 112599. For scripted use, isolate the run with a config file holding only this server. mcp.json, in full (path shortened):
{"mcpServers":{"shopdb":{"command":"C:/path/to/.venv/Scripts/python.exe","args":["C:/path/to/server.py"]}}}
claude -p "Using the shopdb MCP tools only: who are the top 2 customers by paid spend, and how many orders are pending? Answer in two short lines." \
--strict-mcp-config --mcp-config mcp.json \
--allowedTools "mcp__shopdb__top_customers" "mcp__shopdb__run_query" \
--output-format stream-json --verbose --model sonnet < /dev/null
--strict-mcp-config ignores every other server, --allowedTools pre-approves the two tools by their mcp__<server>__<tool> names so a non-interactive run doesn't stall on a permission prompt, and < /dev/null stops the "no stdin data" wait. My headless claude -p post covers running Claude Code non-interactively, and the AI Tutorials index lists related posts. The transcript, parsed from the stream-json output:
INIT [{'name': 'shopdb', 'status': 'connected', 'source': 'dynamic'}] ['mcp__shopdb__run_query', 'mcp__shopdb__top_customers']
CALL ToolSearch {"query": "select:mcp__shopdb__run_query,mcp__shopdb__top_customers", "max_results": 5}
RESULT [{'type': 'tool_reference', 'tool_name': 'mcp__shopdb__run_query'}, {'type': 'tool_reference', 'tool_name': 'mcp__shopdb__top_customers'}]
CALL mcp__shopdb__top_customers {"status": "paid", "limit": 2}
CALL mcp__shopdb__run_query {"sql": "SELECT COUNT(*) AS pending_orders FROM orders WHERE status = 'pending'"}
RESULT {"result":[{"name":"Linus","country":"FI","spent":310},{"name":"Ada","country":"UK","spent":120.5}]}
RESULT {"result":[{"pending_orders":1}]}
TEXT Top 2 by paid spend: Linus (FI) at 310, then Ada (UK) at 120.5.
Pending orders: 1.
META turns=4 cost_usd=0.1046514 ms=10650
The ToolSearch call is Claude Code's own tool. The INIT line lists only tool names, and the model fetched the two schemas through ToolSearch before calling them. The Claude Code MCP docs say tool search "is on by default". It appeared in every isolated run below. Claude picked the narrow tool for the ranking question, wrote its own SQL for the one-off count, and ran both in the same turn. The table below shows the num_turns, total_cost_usd and duration_ms fields from each run's result record on my machine; yours will differ.
| Run | num_turns |
total_cost_usd |
duration_ms |
|---|---|---|---|
| isolated: top 2 + pending count (above) | 4 | 0.1046514 | 10650 |
isolated: DELETE FROM orders |
3 | 0.1111452 | 10638 |
| isolated: HTTP config | 3 | 0.110742 | 8622 |
What the model sees when SQLite refuses
I asked Claude to run DELETE FROM orders. The server raised ToolError after the authorizer refused:
CALL mcp__shopdb__run_query {"sql": "DELETE FROM orders"}
RESULT Error executing tool run_query: SQLite rejected the query: not authorized is_error= True
TEXT The delete didn't run: the shopdb `run_query` tool is read-only, and SQLite rejected `DELETE FROM orders` with "not authorized", so the orders table is unchanged.
The tool result is flagged is_error and carries your message, so the model can explain or retry. The message is all it has to go on, so word it for a reader. In 2.x the migration guide separates two paths: ToolError (or a CallToolResult with is_error=True) for failures the model should see, and MCPError, which becomes a top-level JSON-RPC error. I only exercised ToolError.
Why a stray print breaks an MCP stdio server
The official build guide and the stdio transport spec say a stdio server must not write anything to stdout that isn't a valid MCP message, because stdout is the protocol channel. I cite the 2025-06-18 revision because it's the one my server negotiated. The newer 2026-07-28 stdio page, which I re-fetched while revising this post, still says: "The server MUST NOT write anything to its stdout that is not a valid MCP message," and allows stderr for logging. That's why server.py configures logging to stderr.
To see what happens, I replaced the log.info line with print("DEBUG running", sql) and re-ran both clients. The Python SDK client (inspect_stdio.py) printed this to stderr and carried on; the tool calls still returned results:
Failed to parse JSONRPC message from server
Traceback (most recent call last):
File "C:/path/to\.venv\Lib\site-packages\mcp\client\stdio.py", line 222, in _parse_line
message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
pydantic_core._pydantic_core.ValidationError: 1 validation error for union[JSONRPCRequest,JSONRPCNotification,JSONRPCResponse,JSONRPCError]
Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='DEBUG running DELETE FROM orders\r', input_type=str]
Claude Code 2.1.295, pointed at the same polluted server with a SELECT COUNT(*) FROM orders prompt, also completed the call and answered "The orders table has 6 rows." So: tolerated by these two clients, still forbidden by the spec. Another client may not be as forgiving, so keep stdout clean: log to stderr, and check any dependency that prints at import time.
stdio vs HTTP: which MCP transport to use
Start with stdio; move to HTTP only when a second person or machine needs the same server. The spec defines two standard transports. stdio: the client launches your server as a subprocess, one client per process. Streamable HTTP: your server runs independently on one endpoint (for example /mcp) and handles many clients.
| stdio | Streamable HTTP | |
|---|---|---|
| Clients | one, the process that launched it | many |
| Deploy | nothing; Claude Code starts and stops it | you run it and keep it alive |
| Auth | none needed, it's your local subprocess | none in the server.py --http I ran; you must add it |
| Use it for | a local tool for one developer | a server shared across people or machines |
The switch is the last block of server.py above: with --http it calls mcp.run(transport="streamable-http", host="127.0.0.1", port=18911). The migration guide shows the same form and says host and port belong on run() only. Start it with python server.py --http, then register it:
claude mcp add --transport http shopdb-http http://127.0.0.1:18911/mcp
claude mcp get shopdb-http
Added HTTP MCP server shopdb-http with URL: http://127.0.0.1:18911/mcp to local config
shopdb-http:
Scope: Local config (private to you in this project)
Status: ✔ Connected
Type: http
URL: http://127.0.0.1:18911/mcp
claude -p with an HTTP --mcp-config ({"mcpServers":{"shopdb":{"type":"http","url":"http://127.0.0.1:18911/mcp"}}}) made the same call, mcp__shopdb__top_customers {"limit": 2, "status": "paid"}, and got the same two customers back (last row of the table above).
As written, this HTTP server has no authentication: anyone who can reach the port can call run_query. Keep it on localhost, or put real auth in front of it before you expose it. The Origin check is not a substitute, though it does work: an initialize request sent with -H "Origin: http://evil.example" returned HTTP/1.1 403 Forbidden in my test. The migration guide says DNS rebinding protection is turned on automatically when host is 127.0.0.1, localhost or ::1, and the spec requires servers to validate Origin (2025-06-18).
Output limits apply either way: the Claude Code docs give 25,000 tokens as the default cap on MCP output, which is why run_query has a limit parameter.
Next step: fail CI when the tool call breaks
Pin your SDK on purpose (mcp>=2,<3 for new work, mcp<2 if you're keeping FastMCP imports), then add a check that runs claude -p against the isolated config and asserts that a shopdb tool was actually called, so a broken tool schema fails a build instead of a conversation:
# ci_check.py - fail the build if Claude can't call the tool
import json, subprocess, sys
PROMPT = "Using the shopdb MCP tools only: who are the top 2 customers by paid spend? One line."
out = subprocess.run(
["claude", "-p", PROMPT, "--strict-mcp-config", "--mcp-config", "mcp.json",
"--allowedTools", "mcp__shopdb__top_customers", "mcp__shopdb__run_query",
"--output-format", "stream-json", "--verbose", "--model", "sonnet"],
capture_output=True, text=True, stdin=subprocess.DEVNULL, encoding="utf8", timeout=180,
).stdout
events = [json.loads(line) for line in out.splitlines() if line.startswith("{")]
blocks = [b for e in events if e.get("type") == "assistant" for b in e["message"]["content"]]
calls = [b for b in blocks if b["type"] == "tool_use" and b["name"].startswith("mcp__shopdb__")]
assert calls, "Claude never called a shopdb tool"
assert "Linus" in out, "expected top customer Linus in the output"
print("ok:", [c["name"] for c in calls])
I ran it against the server above from the directory holding mcp.json:
ok: ['mcp__shopdb__top_customers']
Exit code 0. It calls a real model, so it costs money, and the model's wording can vary; the tool_use assertion is the stable part and "Linus" in out is the looser one. Drop the second assert if your data changes.
Keep reading
Bekzod Erkinov
AuthorFounder of NextGenBeing. Software engineer working with Laravel, Python, and cloud infrastructure. Writes about patterns that actually hold up in production. Based in Tashkent, Uzbekistan.
Get the AI-Assisted Developer's Field Guide
The workflow, prompts, and tools I use to ship faster with AI — free when you subscribe. Plus new deep-dives in your inbox. No spam, unsubscribe anytime.
Comments (0)
Please log in to leave a comment.
Log In