"""tiny_mcp.py — a hand-written MCP server and client, to SEE the protocol (lessons 12 and 14). Real servers use the official SDK (lesson 13): `from mcp.server import MCPServer`. This file exists so the course can show every JSON-RPC message on the wire, run offline, and need no extra packages. It follows the stateless 2026-07-28 revision of the spec, in miniature: * transport: stdio, one JSON-RPC message per line; logs go to stderr, never stdout * every request carries _meta["io.modelcontextprotocol/protocolVersion"] and ["…/clientCapabilities"] * methods: server/discover, tools/list, tools/call, resources/list, resources/read * every result carries resultType; list/read results carry ttlMs and cacheScope * tool failures are results with isError: true; protocol failures are JSON-RPC errors python tiny_mcp.py # run as a server on stdin/stdout (a client launches it) """ from __future__ import annotations import json import subprocess import sys PROTOCOL = "2026-07-28" META = "io.modelcontextprotocol/" SERVER_INFO = {"name": "tiny-docs", "version": "0.1.0"} # ------------------------------------------------------------------------------ server side _index = None def _docs(): """Load the lessons lazily, on the first call that needs them.""" global _index if _index is None: from mini_rag import BM25, doc_text from site_docs import load_sections docs = load_sections() _index = (docs, BM25([doc_text(d) for d in docs])) return _index def search_docs(query: str, k: int = 3) -> list[dict]: docs, bm25 = _docs() scores = bm25.scores(query) top = sorted(range(len(docs)), key=lambda i: -scores[i])[:k] return [{"url": docs[i]["url"], "title": docs[i]["title"], "section": docs[i]["section"], "score": round(float(scores[i]), 2)} for i in top if scores[i] > 0] def count_lessons(course: str) -> dict: docs, _ = _docs() courses = sorted({d["course"] for d in docs}) if course not in courses: raise ValueError(f"unknown course {course!r}; valid courses: {courses}") return {"course": course, "lessons": len({d["url"].split("#")[0] for d in docs if d["course"] == course})} TOOLS = { "search_docs": (search_docs, { "name": "search_docs", "title": "Search the course lessons", "description": "Keyword search over every lesson section. Returns url, title, section, score.", "inputSchema": {"type": "object", "additionalProperties": False, "required": ["query"], "properties": {"query": {"type": "string"}, "k": {"type": "integer", "minimum": 1, "maximum": 10}}}}), "count_lessons": (count_lessons, { "name": "count_lessons", "title": "Count lessons in a course", "description": "Number of lessons in one course folder, e.g. 'sql' or 'mysql'.", "inputSchema": {"type": "object", "additionalProperties": False, "required": ["course"], "properties": {"course": {"type": "string"}}}}), } def _result(**fields) -> dict: return {"resultType": "complete", **fields, "_meta": {META + "serverInfo": SERVER_INFO}} def handle(msg: dict) -> dict | None: """One JSON-RPC message in, one response out (None for notifications).""" if "id" not in msg: # a notification: never answered return None rid, method, params = msg["id"], msg.get("method"), msg.get("params") or {} meta = params.get("_meta") or {} def error(code, message, data=None): err = {"code": code, "message": message} if data is not None: err["data"] = data return {"jsonrpc": "2.0", "id": rid, "error": err} # stateless: every request must say which protocol version and client capabilities it uses if META + "protocolVersion" not in meta or META + "clientCapabilities" not in meta: return error(-32602, "missing required _meta fields (protocolVersion, clientCapabilities)") if meta[META + "protocolVersion"] != PROTOCOL: return error(-32022, "Unsupported protocol version", {"supported": [PROTOCOL]}) if method == "server/discover": result = _result(supportedVersions=[PROTOCOL], capabilities={"tools": {}, "resources": {}}, instructions="Search this learning site's lessons.", ttlMs=3_600_000, cacheScope="public") elif method == "tools/list": result = _result(tools=[spec for _, spec in TOOLS.values()], ttlMs=300_000, cacheScope="public") elif method == "tools/call": name = params.get("name") if name not in TOOLS: return error(-32602, f"Unknown tool: {name}") func, _ = TOOLS[name] try: value = func(**(params.get("arguments") or {})) result = _result(content=[{"type": "text", "text": json.dumps(value)}], structuredContent=value, isError=False) except Exception as e: # tool execution error: the model can fix it result = _result(content=[{"type": "text", "text": f"{type(e).__name__}: {e}"}], isError=True) elif method == "resources/list": docs, _ = _docs() lessons = sorted({d["url"].split("#")[0] for d in docs})[:5] result = _result(resources=[{"uri": f"lesson://{u}", "name": u, "mimeType": "text/plain"} for u in lessons], ttlMs=300_000, cacheScope="public") elif method == "resources/read": docs, _ = _docs() url = params.get("uri", "").removeprefix("lesson://") text = "\n\n".join(d["text"] for d in docs if d["url"].split("#")[0] == url) if not text: return error(-32602, f"Resource not found: {params.get('uri')}") result = _result(contents=[{"uri": params["uri"], "mimeType": "text/plain", "text": text[:2000]}], ttlMs=300_000, cacheScope="public") else: return error(-32601, f"Method not found: {method}") return {"jsonrpc": "2.0", "id": rid, "result": result} def serve() -> None: print("tiny-docs MCP server ready (logs go to stderr)", file=sys.stderr) for line in sys.stdin: if line.strip(): response = handle(json.loads(line)) if response is not None: sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() # ------------------------------------------------------------------------------ client side class StdioClient: """Launches a server subprocess and speaks newline-delimited JSON-RPC to it.""" def __init__(self, command: list[str], show: bool = False) -> None: self.proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding="utf-8") self.next_id, self.show = 1, show def request(self, method: str, params: dict | None = None, version: str = PROTOCOL) -> dict: params = dict(params or {}) params["_meta"] = {META + "protocolVersion": version, META + "clientCapabilities": {}, META + "clientInfo": {"name": "tiny-client", "version": "0.1.0"}} msg = {"jsonrpc": "2.0", "id": self.next_id, "method": method, "params": params} self.next_id += 1 self.proc.stdin.write(json.dumps(msg) + "\n") self.proc.stdin.flush() reply = json.loads(self.proc.stdout.readline()) if self.show: print("→", json.dumps({k: v for k, v in msg.items() if k != "jsonrpc"})[:150]) print("←", json.dumps({k: v for k, v in reply.items() if k != "jsonrpc"})[:150]) return reply def list_tools(self) -> list[dict]: return self.request("tools/list")["result"]["tools"] def call_tool(self, name: str, arguments: dict) -> dict: return self.request("tools/call", {"name": name, "arguments": arguments})["result"] def close(self) -> None: self.proc.stdin.close() # stdio shutdown: close the server's input, then wait self.proc.wait(timeout=10) def __enter__(self): return self def __exit__(self, *exc): self.close() if __name__ == "__main__": serve()