# MCP 协议实战:不用 SDK,手写 JSON-RPC 走通 MCP,给大模型接上本地工具
## 痛点:大模型再强,也摸不到你电脑里的文件
大模型能聊天、能写代码,但它有个硬伤:它够不着你的本地文件、数据库和内部系统。问它“帮我统计一下 sample.txt 有多少字”,它只能瞎编一个数字。这不是模型笨,是它没有“手”。
MCP(Model Context Protocol,模型上下文协议)就是给大模型装“手”的标准协议:模型通过协议调用你提供的本地工具,工具执行完把结果还给模型,模型再组织成回答。2026 年 MCP 已经是 Agent 接入工具的事实标准,但很多教程一上来就让装 SDK、配一堆依赖,反而把最核心的东西盖住了:MCP 本质上就是 JSON-RPC 2.0 over stdio。
这篇文章不用任何第三方库,只用 Python 标准库,手写一个能跑通的 MCP 服务端和客户端。跑通之后你会发现,协议本身一点都不神秘。
## 协议极简版:一次会话只有 4 个来回
MCP over stdio 的一次完整会话,归纳起来就 4 步:
1. 客户端发 initialize,告诉服务端“我是谁、我支持哪个协议版本”;
2. 服务端回 capabilities,告诉客户端“我能提供什么(这里是 tools)”;
3. 客户端发 notifications/initialized(这是 notification,不需要回包);
4. 客户端发 tools/list 查工具清单,发 tools/call 真正调工具。
注意一个细节:第 3 步是 notification,没有 id,服务端不用回包。新手最容易在这里卡住:等一个永远不会来的回包,然后超时。
## 步骤 1:写服务端 mini_mcp_server.py
服务端从 stdin 一行一行读 JSON 请求,处理完把响应写到 stdout。两个本地工具:读文件、统计字数。读文件时我加了个限制——只允许当前目录下的文件,路径里带斜杠直接拒绝。这是真要上生产必须有的习惯:工具一旦暴露给模型,就等于暴露给所有能操纵模型输入的人。
```python
"""mini_mcp_server.py: 不用 SDK, 用标准库手写 MCP 协议服务端。"""
import json
import sys
def read_local_file(path: str) -> str:
"""工具1: 读本地文本文件(限制在当前目录, 防止越权)。"""
if "/" in path or "\\" in path:
return "错误: 只允许读取当前目录下的文件"
try:
with open(path, encoding="utf-8") as f:
return f.read()[:2000]
except FileNotFoundError:
return f"错误: 文件 {path} 不存在"
def count_words(path: str) -> str:
"""工具2: 统计文件字数。"""
text = read_local_file(path)
if text.startswith("错误"):
return text
return f"{path} 共 {len(text)} 字, {len(text.split())} 个词"
TOOLS = {
"read_local_file": (read_local_file, "读取当前目录下的文本文件", ["path"]),
"count_words": (count_words, "统计文本文件的字数和词数", ["path"]),
}
def handle(req: dict) -> dict | None:
method, params, rid = req.get("method"), req.get("params", {}), req.get("id")
if method == "initialize":
return {"jsonrpc": "2.0", "id": rid, "result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "mini-mcp-demo", "version": "0.1.0"}}}
if method == "notifications/initialized":
return None # notification 无需回包
if method == "tools/list":
return {"jsonrpc": "2.0", "id": rid, "result": {"tools": [
{"name": n, "description": d, "inputSchema": {
"type": "object", "properties": {a: {"type": "string"} for a in args},
"required": args}}
for n, (_, d, args) in TOOLS.items()]}}
if method == "tools/call":
name, args = params.get("name"), params.get("arguments", {})
if name not in TOOLS:
return {"jsonrpc": "2.0", "id": rid, "error": {
"code": -32602, "message": f"未知工具: {name}"}}
result = TOOLS[name][0](**args)
return {"jsonrpc": "2.0", "id": rid, "result": {
"content": [{"type": "text", "text": result}]}}
return {"jsonrpc": "2.0", "id": rid, "error": {
"code": -32601, "message": f"未知方法: {method}"}}
def main() -> None:
for line in sys.stdin:
line = line.strip()
if not line:
continue
resp = handle(json.loads(line))
if resp is not None:
print(json.dumps(resp, ensure_ascii=False), flush=True)
if __name__ == "__main__":
main()
```
这里有个必须注意的点:服务端 stdout 只能走协议 JSON,不能 print 调试信息。真实 MCP 服务端调试日志要走 stderr,一旦 stdout 混进一句 print,客户端的 JSON 解析立刻就炸。这是新手最常见的翻车,没有之一。
## 步骤 2:写客户端 mini_mcp_client.py
客户端用 subprocess 拉起服务端进程,通过 stdin/stdout 一问一答。注意两个细节:每次写完必须 flush,否则服务端在 stdin 缓冲里等不到数据;读响应用 readline,因为协议约定“一行一个 JSON”。
```python
"""mini_mcp_client.py: 用标准库 subprocess 驱动服务端, 走完一次完整 MCP 会话。"""
import json
import subprocess
import sys
def rpc(proc, method, params=None, rid=1):
req = {"jsonrpc": "2.0", "id": rid, "method": method}
if params is not None:
req["params"] = params
proc.stdin.write(json.dumps(req, ensure_ascii=False) + "\n")
proc.stdin.flush()
return json.loads(proc.stdout.readline())
def main() -> None:
proc = subprocess.Popen([sys.executable, "mini_mcp_server.py"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
text=True, encoding="utf-8")
init = rpc(proc, "initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {}, "clientInfo": {"name": "demo-client", "version": "0.1.0"}}, rid=1)
print("initialize ->", init["result"]["serverInfo"])
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n")
proc.stdin.flush()
tools = rpc(proc, "tools/list", rid=2)
print("tools/list ->", [t["name"] for t in tools["result"]["tools"]])
call = rpc(proc, "tools/call", {
"name": "count_words", "arguments": {"path": "sample.txt"}}, rid=3)
print("tools/call ->", call["result"]["content"][0]["text"])
proc.stdin.close()
proc.wait()
if __name__ == "__main__":
main()
```
## 步骤 3:实测,输出长这样
先准备一个测试文件 sample.txt(两行中文),然后跑客户端:
```
initialize -> {'name': 'mini-mcp-demo', 'version': '0.1.0'}
tools/list -> ['read_local_file', 'count_words']
tools/call -> sample.txt 共 27 字, 4 个词
```
三行输出对应三次往返:握手成功、工具清单拿到、工具真正执行并返回了结果。没有 SDK、没有依赖,协议就这么跑通了。完整代码和这次的运行记录我都留着,读者照抄就能复现。
## 避坑:4 个真踩过的点
1. stdout 混日志:服务端所有调试信息走 stderr,stdout 只留协议 JSON,否则客户端解析必炸。
2. notification 等回包:notifications/initialized 没有 id、不回包,客户端写完直接往下走,不要 readline。
3. 忘记 flush:Python 的 stdin/stdout 有缓冲,不 flush 对端就一直等,表现就是“程序卡住不动”,不是协议错了。
4. 工具不做参数校验:演示里的读文件限制了当前目录。真实场景里,模型传进来的路径、SQL、shell 命令都不可信,工具层必须自己设防,不能指望模型“自觉”。
## 下一步:从演示到能用
这个 Demo 离生产还差三件事,按顺序补:第一,把 mock 的客户端换成真正的大模型——模型输出 tools/call 请求,你的客户端原样转发给服务端,再把结果塞回模型上下文;第二,给工具加超时和异常隔离,单个工具挂了不能拖垮整个会话;第三,工具清单要配好 description,模型选工具全靠读这段描述,描述写得含糊,模型就会调错工具。
协议本身只用了 100 多行标准库代码就讲清楚了。真正难的从来不是协议,而是工具设计:给模型什么样的“手”,它才能干出靠谱的活。这也是下一篇 Agent 实战要接着聊的。
(本文代码全部用 Python 3.12 标准库实测跑通,无第三方依赖。)