A real MCP server (stdio transport, mcp SDK)

Open in Colab Run this notebook in Colab

A real MCP server (stdio transport, mcp SDK)#

Track 08 - Production · Notebook 06 · Runtime: ~10s LIVE, <1s replay

Builds an MCP server with the official mcp Python package, drives it from a Python client over stdio, then prints the config snippet any MCP-aware editor needs to attach to it. Replaces the hand-rolled JSON-RPC in 04_agents/05 with the real protocol implementation.

from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root

import asyncio
import json
import os
from pathlib import Path

s = bootstrap("08_production_06_mcp_real_server")

try:
    import mcp  # noqa: F401
    HAS_MCP = True
except ImportError:
    HAS_MCP = False

LIVE = HAS_MCP and bool(os.environ.get("MCP_LIVE"))
FIXTURE = json.loads((Path(repo_root()) / "notebooks/08_production/_fixtures/06_mcp.json").read_text())
print(f"mode={'LIVE' if LIVE else 'REPLAY'}  mcp_installed={HAS_MCP}")

The server#

Server.list_tools() and Server.call_tool() decorators register handlers; mcp.server.stdio.stdio_server() is the JSON-RPC-over-stdio transport. The server below runs as python lab_mcp_server.py and speaks the MCP protocol on stdin/stdout — exactly what Cursor or any other MCP client expects.

SERVER_PATH = Path("/tmp") / "lab_mcp_server.py"
SERVER_PATH.write_text('"""A minimal MCP server with three tools, runnable via stdio.\n\n    python lab_mcp_server.py    # talks JSON-RPC on stdin/stdout\n\nCursor / Claude Code / any MCP-aware client can attach by pointing at this\nfile with `command: python` and `args: [path/to/lab_mcp_server.py]`.\n"""\nfrom __future__ import annotations\n\nimport asyncio\nimport ast\nimport operator as op\nfrom datetime import datetime, timedelta, timezone\n\nimport mcp.server.stdio\nimport mcp.types as types\nfrom mcp.server import Server, NotificationOptions\nfrom mcp.server.models import InitializationOptions\n\nserver = Server("lab-mcp-server")\n\n_OPS = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv,\n        ast.Pow: op.pow, ast.USub: op.neg, ast.UAdd: op.pos}\n\ndef _safe_eval(node):\n    if isinstance(node, ast.Constant): return node.value\n    if isinstance(node, ast.BinOp):    return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))\n    if isinstance(node, ast.UnaryOp):  return _OPS[type(node.op)](_safe_eval(node.operand))\n    raise ValueError(type(node).__name__)\n\n\nFACTS = {\n    "capital of france": "Paris", "capital of japan": "Tokyo",\n    "tallest mountain":  "Mount Everest",\n}\nNOW = datetime(2026, 4, 17, 12, 0, tzinfo=timezone.utc)\n\n\n@server.list_tools()\nasync def list_tools() -> list[types.Tool]:\n    return [\n        types.Tool(\n            name="calculator",\n            description="Evaluate a simple arithmetic expression like \\"2 + 3 * 4\\".",\n            inputSchema={"type": "object", "required": ["expression"],\n                         "properties": {"expression": {"type": "string"}}},\n        ),\n        types.Tool(\n            name="wiki_lookup",\n            description="Look up a short fact (capitals, authors, etc.).",\n            inputSchema={"type": "object", "required": ["query"],\n                         "properties": {"query": {"type": "string"}}},\n        ),\n        types.Tool(\n            name="get_date",\n            description="Return the date for a relative spec (today, tomorrow, year, ...).",\n            inputSchema={"type": "object", "required": ["spec"],\n                         "properties": {"spec": {"type": "string",\n                             "enum": ["today","tomorrow","yesterday","year","month","day"]}}},\n        ),\n    ]\n\n\n@server.call_tool()\nasync def call_tool(name: str, arguments: dict) -> list[types.TextContent]:\n    if name == "calculator":\n        v = _safe_eval(ast.parse(arguments["expression"], mode="eval").body)\n        out = str(int(v)) if isinstance(v, float) and v.is_integer() else str(v)\n    elif name == "wiki_lookup":\n        out = FACTS.get(arguments["query"].lower().strip(), "no match")\n    elif name == "get_date":\n        s = arguments["spec"].lower()\n        out = {\n            "year": str(NOW.year), "month": NOW.strftime("%B"), "day": NOW.strftime("%A"),\n            "today": NOW.date().isoformat(),\n            "tomorrow":  (NOW + timedelta(days=1)).date().isoformat(),\n            "yesterday": (NOW - timedelta(days=1)).date().isoformat(),\n        }.get(s, f"unknown {s!r}")\n    else:\n        raise ValueError(f"unknown tool {name!r}")\n    return [types.TextContent(type="text", text=out)]\n\n\nasync def main():\n    async with mcp.server.stdio.stdio_server() as (read, write):\n        await server.run(\n            read, write,\n            InitializationOptions(\n                server_name="lab-mcp-server", server_version="0.1.0",\n                capabilities=server.get_capabilities(\n                    notification_options=NotificationOptions(), experimental_capabilities={}),\n            ),\n        )\n\n\nif __name__ == "__main__":\n    asyncio.run(main())\n')
print(f"server saved to {SERVER_PATH}  ({SERVER_PATH.stat().st_size:,} bytes)")
print(SERVER_PATH.read_text().splitlines()[0])  # first line

The client#

stdio_client(StdioServerParameters) spawns the server as a subprocess and yields (read, write) streams. Wrap them in a ClientSession and you get initialize, list_tools, call_tool as awaitables.

async def drive_server() -> dict:
    """Spawn the server (LIVE) and run a small protocol session, or replay."""
    if not LIVE:
        return {
            "tools_list": FIXTURE["tools_list"],
            "calls":      [{"req": tc, "result": tc["result"]} for tc in FIXTURE["tool_calls"]],
            "server_info": {"name": FIXTURE["server_name"], "version": FIXTURE["server_version"]},
        }

    from mcp import ClientSession  # noqa: PLC0415
    from mcp.client.stdio import StdioServerParameters, stdio_client  # noqa: PLC0415

    params = StdioServerParameters(command="python", args=[str(SERVER_PATH)])
    out: dict = {"tools_list": [], "calls": [], "server_info": None}

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as sess:
            init = await sess.initialize()
            out["server_info"] = {"name": init.serverInfo.name,
                                  "version": init.serverInfo.version}
            tools = await sess.list_tools()
            out["tools_list"] = [
                {"name": t.name, "description": t.description, "inputSchema": t.inputSchema}
                for t in tools.tools
            ]
            for tc in FIXTURE["tool_calls"]:
                resp = await sess.call_tool(tc["name"], tc["arguments"])
                text = next((c.text for c in resp.content if c.type == "text"), "")
                out["calls"].append({"req": tc, "result": text})
    return out


# Jupyter has a running loop, so use top-level await rather than asyncio.run.
session = await drive_server()
print(f"server: {session['server_info']}")
print(f"tools: {[t['name'] for t in session['tools_list']]}")

Tool calls round-tripped#

for entry in session["calls"]:
    req = entry["req"]
    print(f"  {req['name']}({req['arguments']}) → {entry['result']}")

Attaching from an MCP client#

Drop this block into ~/.config/claude-desktop/claude_desktop_config.json, .cursor/mcp.json, or any MCP-compatible client. The same server, now exposed as tools to whatever model the client drives.

config = {
    "mcpServers": {
        "lab-mcp-server": {
            "command": "python",
            "args": [str(SERVER_PATH)]
        }
    }
}
print(json.dumps(config, indent=2))

Checks#

s.check(
    "server_advertised_three_tools",
    lambda: {t["name"] for t in session["tools_list"]} == {"calculator", "wiki_lookup", "get_date"},
    msg=f"got {[t['name'] for t in session['tools_list']]}",
)
s.check(
    "calculator_returns_1029",
    lambda: any(c["req"]["name"] == "calculator" and "1029" in c["result"] for c in session["calls"]),
    msg=f"calls = {session['calls']}",
)
s.check(
    "wiki_returns_paris",
    lambda: any(c["req"]["name"] == "wiki_lookup" and c["result"] == "Paris" for c in session["calls"]),
)
s.check(
    "every_tool_has_input_schema",
    lambda: all(
        isinstance(t["inputSchema"], dict) and t["inputSchema"].get("type") == "object"
        for t in session["tools_list"]
    ),
)
s.check(
    "server_self_identifies",
    lambda: session["server_info"]["name"] == "lab-mcp-server",
    msg=f"server_info={session['server_info']}",
)

Notes for production#

  • Stdio is the default for desktop integrations (Claude Code, Cursor, Cody, Zed). For network/multi-tenant deployments use the streamable HTTP transport (mcp.server.streamable_http), which replaced HTTP+SSE in the 2025-03 spec.

  • Resources and prompts, not just tools. The MCP spec covers three primitives: tools (callables), resources (readable URIs), and prompts (templates). Most servers only need tools, but resources are how you expose a docs corpus or a database read-only.

  • Auth: stdio inherits the spawning user’s credentials. For HTTP transports the spec uses OAuth 2.1 with PKCE and resource-indicators per RFC 8707.

  • Real servers worth reading: github-mcp-server (github.com/github/github-mcp-server), mcp-server-postgres, mcp-server-filesystem. All three are good templates for tools over a system you don’t control.

s.summary()
s.save()