Build an MCP Server in Python (and Call It From Claude Code) - NextGenBeing
Back to discoveries

Build an MCP Server in Python (and Call It From Claude Code)

Build a read-only SQLite MCP server in Python with the official SDK's MCPServer (formerly FastMCP), run it over stdio and HTTP, and call it from Claude Code with a real tool-call transcript.

AI Tutorials 11 min read
Bekzod Erkinov

Bekzod Erkinov

Oct 11, 2026 • 0 views
Size:
Height:
11 min read 4,465 words Focus mode: Eye care:

Listen to Article

Loading...
0:00 / 0:00
0:00 0:00
Low High
0% 100%
⏸ Paused ▶️ Now playing... Ready to play ✓ Finished
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.

Bekzod Erkinov

Bekzod Erkinov

Author

Founder of NextGenBeing. Software engineer working with Laravel, Python, and cloud infrastructure. Writes about patterns that actually hold up in production. Based in Tashkent, Uzbekistan.

Free guide

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

Related Articles

Don't miss the next deep dive

Get one well-researched tutorial in your inbox each week. No spam, unsubscribe anytime.