仅用标准库构建 MCP stdio 服务器

目标

只用 json、sys、sqlite3 写出一个 MCP 服务器。它要按顺序接收客户端发来的 initialize → notifications/initialized → tools/list → tools/call，最后再写一个把该服务器作为子进程启动的客户端。

为什么重要

用 SDK 的话，五行就能完成。然而事故恰恰出在 SDK 隐藏起来的地方——往 stdout 打印的一行调试信息会让客户端的 JSON 解析器崩溃；对通知发出了回复，产生了无法配对的响应；工具失败了却以协议错误返回，模型就会判断“服务器坏了”。分帧（一行 = 一条消息）、按 id 配对、忽略通知、区分两种错误——这四件事才是 MCP 的实质，亲手写一遍之后，就能读懂 SDK 的日志在说什么。

步骤

保存 /root/mcp/seed.sql，并加载到 /root/mcp/shop.db。customers 表要有 5 行，orders 表要有 8 行。
创建 /root/mcp/server.py。从 stdin 逐行读取 JSON-RPC 请求，并用包含 protocolVersion（值为 2025-06-18）、capabilities.tools 和 serverInfo.name 的 result 回答 initialize。
在同一个文件中，对通知（没有 id 的消息）不作回答，对未知方法回答 -32601，对损坏的 JSON 回答 -32700 错误（id 为 null）。服务器不能崩溃。
让 tools/list 返回两个工具——list_customers 和 count_orders（inputSchema.properties.status，字符串，必填）。每个工具都要有 description 和 type: object 的 inputSchema。
实现 tools/call。给 count_orders 传入 {"status":"paid"} 时，content[0].text 中要包含数据库中 paid 的记录数，list_customers 要包含全部五位客户的姓名。数据库路径：如果有环境变量 MCP_DB 就用它，没有就用 /root/mcp/shop.db。
把错误分成两种。不存在的工具名用 JSON-RPC 错误 -32602，给 count_orders 传入未知的状态值（例如 banana）时，要在 result 中返回 isError: true 和说明文本。
创建 /root/mcp/client.py。用 subprocess 启动服务器，依次发送 initialize → notifications/initialized → tools/list → tools/call（count_orders，paid），并把结果写入 /root/mcp/session.json，分别记为 tools（名称列表）和 paid_orders（响应文本）。如果有环境变量 MCP_SESSION_OUT，就写到该路径。
把服务器的所有日志都发送到 stderr。每处理一个请求，stderr 中就要打印一行方法名，而 stdout 中除 JSON 响应外不能出现任何东西。

参考

逐行读取的循环只要 for line in sys.stdin: 即可。写入响应要用 sys.stdout.write(json.dumps(...) + "\n")，之后必须 flush()——管道不是行缓冲，不 flush 的话，客户端会永远等下去。
是否为通知，要用 "id" in msg 来判断。如果只看 msg.get("id") 是否为 None，就无法与 id 实际为 null 的（错误）请求区分。
如果不捕获 json.loads 抛出的异常，一行损坏的数据就会让整个服务器崩溃。客户端会把它看成“服务器消失了”。
手动测试的方法：printf '%s\n' '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' | python3 /root/mcp/server.py
常见错误 1：用 print() 把调试语句打印到 stdout。第 8 步的评分会抓出这一点。
常见错误 2：把不存在的工具以 isError: true 返回。那不是工具在执行中失败，而是请求本身有误，所以属于协议错误。

创建商店数据库

保存 /root/mcp/seed.sql，并加载到 /root/mcp/shop.db。customers 表要有 5 行，orders 表要有 8 行。

sqlite3 可以用 sqlite3 shop.db < seed.sql 整个执行文件。用 Python 的话是 sqlite3.connect(...).executescript(open(...).read())。如果向已有的数据库再次加载，会报表已存在的错误，所以先把它删掉。

回答 initialize

创建 /root/mcp/server.py。从 stdin 逐行读取 JSON-RPC 请求，并用包含 protocolVersion（值为 2025-06-18）、capabilities.tools 和 serverInfo.name 的 result 回答 initialize。

在骨架示例的循环中，当 method == "initialize" 时构造 result 字典，用 reply(msg["id"], result=...) 发送即可。result 中包含 protocolVersion、capabilities（带有 tools 键的对象）、serverInfo（name、version）三项。

不回答通知

在同一个文件中，对通知（没有 id 的消息）不作回答，对未知方法回答 -32601，对损坏的 JSON 回答 -32700 错误（id 为 null）。服务器不能崩溃。

在 JSON-RPC 中，通知是没有 id 成员的请求，服务器不能回答。用 try 包住 json.loads，遇到 JSONDecodeError 就以 id 为 null 发送 -32700 并 continue。其余异常也捕获并以 -32603 回答，服务器就能存活下来。

给出工具列表

让 tools/list 返回两个工具——list_customers 和 count_orders（inputSchema.properties.status，字符串，必填）。每个工具都要有 description 和 type: object 的 inputSchema。

result 是 {"tools": [...]}，一个工具有 name、description、inputSchema 三个键。inputSchema 是 JSON Schema 对象，所以带有 "type": "object" 和 properties。没有参数的工具也要保留 properties: {}。

真正执行工具

实现 tools/call。给 count_orders 传入 {"status":"paid"} 时，content[0].text 中要包含数据库中 paid 的记录数，list_customers 要包含全部五位客户的姓名。数据库路径：如果有环境变量 MCP_DB 就用它，没有就用 /root/mcp/shop.db。

params 的形式是 {"name": 도구이름, "arguments": {...}}（占位符为工具名称）。结果必须是 {"content": [{"type": "text", "text": "..."}]} 的形式。用 os.environ.get("MCP_DB", "/root/mcp/shop.db") 取路径，并在每个请求里 sqlite3.connect，用完关闭。

错误有两种

把错误分成两种。不存在的工具名用 JSON-RPC 错误 -32602，给 count_orders 传入未知的状态值（例如 banana）时，要在 result 中返回 isError: true 和说明文本。

规范把“未知的工具、错误的参数”归为协议错误（error 成员，示例代码 -32602），把“工具在运行中失败”归为工具执行错误（result 中的 isError: true）。后者要用文本写明原因，让模型读到后可以再次尝试。

用客户端走完一次会话

创建 /root/mcp/client.py。用 subprocess 启动服务器，依次发送 initialize → notifications/initialized → tools/list → tools/call（count_orders，paid），并把结果写入 /root/mcp/session.json，分别记为 tools（名称列表）和 paid_orders（响应文本）。如果有环境变量 MCP_SESSION_OUT，就写到该路径。

用 subprocess.Popen([...], stdin=PIPE, stdout=PIPE, text=True) 启动，每次发送后都 stdin.flush()，接收时用 stdout.readline() 读一行并 json.loads。发送通知之后不要 readline——不会有答复，会永远等下去。结束时关闭 stdin 并 wait()。

stdout 只用于协议

把服务器的所有日志都发送到 stderr。每处理一个请求，stderr 中就要打印一行方法名，而 stdout 中除 JSON 响应外不能出现任何东西。

用 sys.stderr.write(...) 或 print(..., file=sys.stderr)。规范（stdio 传输）规定服务器不得向 stdout 写入有效 MCP 消息以外的内容，而 stderr 可以用作日志。评分器会把 stdout 的每一行都当作 JSON 来解析。