Model Context Protocol (MCP) — a minimal server and client

Model Context Protocol (MCP) — a minimal server and client#

Track 04 - Agents · Notebook 05 · Runtime: ~5s on CPU

Prerequisites: 04_agents/01 (ReAct from scratch).

Reference: MCP specification.

MCP is a JSON-RPC 2.0 protocol for tool/resource servers an LLM client can attach to. We hand-roll both sides in pure stdlib so the wire format is fully visible — three RPC methods (initialize, tools/list, tools/call), three tools, in-process transport.

For the production version that uses the real mcp Python SDK with stdio transport (the form Cursor / Claude Code attach to), see 08_production/06_mcp_real_server.

from llm_systems_cookbook.nb import bootstrap

import ast
import json
import operator as op
from typing import Any

s = bootstrap("04_agents_05_mcp_server_client")

JSON-RPC 2.0 frame#

Every message is {"jsonrpc": "2.0", "id": ..., "method": ..., "params": ...} or a matching response with result or error. MCP layers its semantics on top. We don’t do streaming transport here - the client calls server.handle(request) synchronously so the notebook is deterministic.

def make_request(method: str, params: dict[str, Any] | None = None, req_id: int = 1) -> dict:
    return {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params or {}}


def make_response(result: Any, req_id: int) -> dict:
    return {"jsonrpc": "2.0", "id": req_id, "result": result}


def make_error(code: int, message: str, req_id: int | None = None, data: Any = None) -> dict:
    err: dict[str, Any] = {"code": code, "message": message}
    if data is not None:
        err["data"] = data
    return {"jsonrpc": "2.0", "id": req_id, "error": err}


# MCP reserves the JSON-RPC error range; these are the ones we use.
PARSE_ERROR = -32700
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
TOOL_NOT_FOUND = -32001

The server#

Three tools: a safe calculator (AST walk, never eval), a stub wiki_search, and a get_current_time. The server advertises them via tools/list (MCP schema shape) and dispatches via tools/call.

_ALLOWED_OPS = {
    ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv,
    ast.Pow: op.pow, ast.USub: op.neg, ast.UAdd: op.pos,
}


def safe_eval(expr: str) -> float:
    def _eval(node: ast.AST) -> float:
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        if isinstance(node, ast.BinOp):
            return _ALLOWED_OPS[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp):
            return _ALLOWED_OPS[type(node.op)](_eval(node.operand))
        raise ValueError(f"unsupported node {type(node).__name__}")
    return _eval(ast.parse(expr.strip(), mode="eval").body)


FACTS = {
    "capital of france": "Paris",
    "tallest mountain":  "Mount Everest",
    "speed of light":    "299,792,458 m/s",
}


class MCPServer:
    PROTOCOL_VERSION = "2024-11-05"
    TOOLS: dict[str, dict] = {
        "calculator": {
            "name": "calculator",
            "description": "Evaluate a simple arithmetic expression.",
            "inputSchema": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
        "wiki_search": {
            "name": "wiki_search",
            "description": "Look up a short fact.",
            "inputSchema": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    }

    def handle(self, req: dict) -> dict:
        if req.get("jsonrpc") != "2.0":
            return make_error(PARSE_ERROR, "invalid JSON-RPC version", req.get("id"))
        method = req.get("method")
        params = req.get("params", {}) or {}
        rid = req.get("id")

        if method == "initialize":
            return make_response(
                {"protocolVersion": self.PROTOCOL_VERSION, "serverInfo": {"name": "lab-mcp-server", "version": "0.1.0"}},
                rid,
            )

        if method == "tools/list":
            return make_response({"tools": list(self.TOOLS.values())}, rid)

        if method == "tools/call":
            name = params.get("name")
            args = params.get("arguments", {})
            if name not in self.TOOLS:
                return make_error(TOOL_NOT_FOUND, f"unknown tool {name!r}", rid)
            try:
                if name == "calculator":
                    value = safe_eval(args["expression"])
                    out = str(int(value)) if isinstance(value, float) and value.is_integer() else str(value)
                elif name == "wiki_search":
                    q = args["query"].lower().strip()
                    out = FACTS.get(q, "no match")
                else:
                    return make_error(METHOD_NOT_FOUND, "unhandled tool", rid)
            except KeyError as e:
                return make_error(INVALID_PARAMS, f"missing argument {e}", rid)
            except Exception as e:  # noqa: BLE001 - surface to client
                return make_error(INVALID_PARAMS, f"{type(e).__name__}: {e}", rid)
            return make_response({"content": [{"type": "text", "text": out}]}, rid)

        return make_error(METHOD_NOT_FOUND, f"unknown method {method!r}", rid)


server = MCPServer()
print("server ready")

The client#

Sends requests, parses results. In production the transport would be stdio or WebSocket; here we just call server.handle(req) directly. The structure of request and response is identical either way.

class MCPClient:
    def __init__(self, server: MCPServer) -> None:
        self.server = server
        self._rid = 0

    def _next_id(self) -> int:
        self._rid += 1
        return self._rid

    def initialize(self) -> dict:
        return self.server.handle(make_request("initialize", req_id=self._next_id()))

    def list_tools(self) -> list[dict]:
        r = self.server.handle(make_request("tools/list", req_id=self._next_id()))
        return r["result"]["tools"]

    def call(self, name: str, arguments: dict) -> dict:
        return self.server.handle(make_request(
            "tools/call",
            {"name": name, "arguments": arguments},
            req_id=self._next_id(),
        ))


client = MCPClient(server)
init = client.initialize()
print(f"initialize -> {init['result']}")

tools = client.list_tools()
print(f"tools/list -> {[t['name'] for t in tools]}")

calc = client.call("calculator", {"expression": "2 ** 10 + 5"})
print(f"tools/call calculator(2**10+5) -> {calc['result']['content'][0]['text']}")

wiki = client.call("wiki_search", {"query": "capital of france"})
print(f"tools/call wiki(capital of france) -> {wiki['result']['content'][0]['text']}")

missing = client.call("nonexistent_tool", {})
print(f"tools/call nonexistent -> {missing.get('error')}")

s.check(
    "initialize_returns_protocol_version",
    lambda: init["result"]["protocolVersion"] == MCPServer.PROTOCOL_VERSION,
    msg=f"got {init['result'].get('protocolVersion')}",
)
s.check(
    "tools_list_includes_both_tools",
    lambda: {"calculator", "wiki_search"} <= {t["name"] for t in tools},
    msg=f"{[t['name'] for t in tools]}",
)
s.check(
    "calculator_returns_correct_result",
    lambda: calc["result"]["content"][0]["text"] == "1029",
    msg=f"got {calc['result']['content'][0]['text']}",
)
s.check(
    "wiki_search_finds_paris",
    lambda: wiki["result"]["content"][0]["text"] == "Paris",
    msg=f"got {wiki['result']['content'][0]['text']}",
)
s.check(
    "unknown_tool_returns_json_rpc_error",
    lambda: missing.get("error", {}).get("code") == TOOL_NOT_FOUND,
    msg=f"error = {missing.get('error')}",
)
s.check(
    "every_response_has_matching_id",
    lambda: all(
        r.get("id") is not None
        for r in (init, missing)
    ),
    msg="initialize and missing-tool responses must echo their request id",
)

Per-method latency#

In-process server.handle(...) calls take microseconds - the JSON-RPC framing and dispatch dominate, since no real transport is involved. The left panel shows the median round-trip per method; the right panel shows the IQR box so you can spot outliers (the unknown-tool error path is slightly faster because it short-circuits before any tool code runs). Add a network transport and every bar shifts up by the same constant.

import time
import matplotlib.pyplot as plt

calls = [
    ("initialize",        lambda: client.initialize()),
    ("tools/list",        lambda: client.list_tools()),
    ("calculator",        lambda: client.call("calculator", {"expression": "2 ** 10 + 5"})),
    ("wiki_search",       lambda: client.call("wiki_search", {"query": "capital of france"})),
    ("unknown (err)",     lambda: client.call("nope", {})),
]
N = 200
latencies_us: dict[str, list[float]] = {}
for name, fn in calls:
    samples: list[float] = []
    for _ in range(N):
        t0 = time.perf_counter(); fn(); samples.append((time.perf_counter() - t0) * 1e6)
    latencies_us[name] = samples

medians = [sorted(latencies_us[n])[N // 2] for n, _ in calls]
labels = [n for n, _ in calls]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9, 3.2))
ax1.bar(labels, medians, color="tab:blue")
ax1.set_ylabel("median latency (µs)"); ax1.set_title(f"per-method round-trip (N={N})")
ax1.tick_params(axis="x", rotation=20)
for i, v in enumerate(medians):
    ax1.text(i, v, f"{v:.1f}", ha="center", va="bottom", fontsize=8)

# matplotlib 3.9 deprecated boxplot(labels=), removed in 3.11: use tick_labels=.
ax2.boxplot([latencies_us[n] for n, _ in calls], tick_labels=labels, showfliers=False)
ax2.set_ylabel("latency (µs)"); ax2.set_title("distribution per method")
ax2.tick_params(axis="x", rotation=20); ax2.grid(True, alpha=0.3, axis="y")
fig.tight_layout(); plt.show()

Exercises#

Real stdio transport. Wrap MCPServer in a loop that reads one JSON line from stdin and writes the response to stdout. Spawn it from a sibling Python process — the protocol is identical.
Streaming progress. MCP supports partial tool results; define a long-running tool that emits notifications/progress and consume it on the client.
Real client. pip install mcp, connect to a real server like github-mcp-server, call one of its tools.

References#

MCP spec (tools, resources, prompts).
08_production/06_mcp_real_server — the same machinery using the official mcp SDK with a real stdio transport.

s.summary()
s.save()