手写一个 MCP Server：从 JSON-RPC 到 Streamable HTTP 的底层全解析


2026 年 8 月，MCP（Model Context Protocol）官方 SDK 月下载量已突破 9700 万次，公开服务端超过 14000 个。OpenAI、Google、Microsoft 全部接入，连北京市 7 月发布的智能体政策都点名要求"推动自主可控互联协议"。但很多人只会 `npx mcp-server-xxx` 一把梭，真到排查线上问题、或者要把内部系统封装成 MCP Server 时，就抓瞎了。

>

这篇文章不聊概念，直接拆协议底层：JSON-RPC 2.0 怎么在 stdio 和 HTTP 两条传输层上跑起来，然后手写一个生产可用的 MCP Server 和一个最小 Client。

一、MCP 到底在底层干了什么

一句话：MCP 是把"模型 ↔ 工具/数据"之间的调用，标准化成一套 JSON-RPC 2.0 消息协议。

在 MCP 出现之前，每个 Agent 框架都自己造一套工具调用机制——OpenAI 有 Function Calling、LangChain 有 Tool 抽象层、HuggingFace 有 Transformers Agents，彼此完全不兼容。MCP 的野心就是做 AI 世界的"USB-C"：只要两端都实现同一套协议，任何 Host 都能接任何 Server。

协议定义了三个角色：

• **Host**：运行 LLM 的应用程序（Cursor、Claude Code、IDE 插件）

• **Client**：Host 内部负责与 Server 通信的协议组件，维护连接生命周期

• **Server**：暴露 Tools（工具）、Resources（资源）、Prompts（提示词）的轻量进程

底层所有消息都是 JSON-RPC 2.0，结构永远是 `jsonrpc / id / method / params` 四件套：

{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {"listChanged": true}}, "clientInfo": {"name": "my-host", "version": "1.0.0"} }}

二、两条传输层：stdio 与 Streamable HTTP

MCP 的消息格式统一，但"怎么把消息送过去"有两条路：

1. STDIO（标准输入输出）：Host 拉起 Server 子进程，通过 stdin/stdout 传 JSON 行。零网络开销、天然继承进程权限隔离，适合本地场景。注意：调试日志必须写 stderr，否则会污染协议流。

2. Streamable HTTP（2025 年规范更新后取代纯 SSE）：无状态 HTTP + 可选 SSE 流式响应，服务端可以挂负载均衡，适合云端部署。2026 年的规范重点就是优化这种无状态模式。

两条传输层共用同一套"会话语义"：连接后先 `initialize` 握手协商协议版本 → 双方交换 `notifications/initialized` → 然后就是 `tools/list`、`tools/call` 的常规调用。

三、手写一个 MCP Server（Python）

不依赖任何 MCP 框架，直接用标准库把协议跑通——这才能看到协议真身。这里实现 stdio 传输：

#!/usr/bin/env python3 """minimal_mcp_server.py — 纯标准库实现的 MCP Server（stdio 传输）""" import json import sys from typing import Any def send(msg: dict) -> None: """MCP 走 stdout，每行一个 JSON 对象""" sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n") sys.stdout.flush() def log(msg: str) -> None: """调试日志必须走 stderr，不能污染协议流""" sys.stderr.write(f"[server] {msg}\n") # 注册两个工具：一个加法、一个查询"今日特价" TOOLS = [ { "name": "add", "description": "计算两个整数之和", "inputSchema": { "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"}, }, "required": ["a", "b"], }, }, { "name": "get_discount", "description": "查询商品今日折扣", "inputSchema": { "type": "object", "properties": {"sku": {"type": "string"}}, "required": ["sku"], }, }, ] def call_tool(name: str, args: dict) -> Any: if name == "add": return {"result": args["a"] + args["b"]} if name == "get_discount": # 真实场景这里会查数据库/调内部 API return {"sku": args["sku"], "discount": 0.85} raise ValueError(f"unknown tool: {name}") def handle(msg: dict) -> None: method = msg.get("method") mid = msg.get("id") if method == "initialize": send({"jsonrpc": "2.0", "id": mid, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {}}, "serverInfo": {"name": "minimal-server", "version": "0.1.0"}, }}) elif method == "notifications/initialized": log("client initialized, ready") elif method == "tools/list": send({"jsonrpc": "2.0", "id": mid, "result": {"tools": TOOLS}}) elif method == "tools/call": params = msg.get("params", {}) try: r = call_tool(params["name"], params.get("arguments", {})) send({"jsonrpc": "2.0", "id": mid, "result": { "content": [{"type": "text", "text": json.dumps(r, ensure_ascii=False)}], "isError": False, }}) except Exception as e: send({"jsonrpc": "2.0", "id": mid, "result": { "content": [{"type": "text", "text": str(e)}], "isError": True, }}) elif method == "ping": send({"jsonrpc": "2.0", "id": mid, "result": {}}) else: log(f"unhandled method: {method}") if __name__ == "__main__": for line in sys.stdin: line = line.strip() if not line: continue handle(json.loads(line))

核心就三点：stdout 发消息、stderr 打日志、按 method 分发。一个能跑的 MCP Server 骨架就是这么薄。

四、手写最小 Client：理解 Host 侧的握手

再看 Host 侧怎么跟 Server 对话。一个最小 Client 需要经历：启动子进程 → `initialize` → 等 `initialized` 通知 → 调工具：

#!/usr/bin/env python3 """minimal_mcp_client.py — 最小 MCP Client，演示完整握手""" import json import subprocess import sys proc = subprocess.Popen( [sys.executable, "minimal_mcp_server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, ) def request(method: str, params: dict, mid: int) -> dict: proc.stdin.write(json.dumps( {"jsonrpc": "2.0", "id": mid, "method": method, "params": params} ) + "\n") proc.stdin.flush() return json.loads(proc.stdout.readline()) # 1. 握手：协商协议版本与能力 resp = request("initialize", { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "minimal-client", "version": "0.1.0"}, }, mid=1) assert resp["result"]["protocolVersion"] == "2025-06-18" print("✅ 握手成功:", resp["result"]["serverInfo"]) # 2. 通知 Server 初始化完成（通知没有 id） proc.stdin.write(json.dumps( {"jsonrpc": "2.0", "method": "notifications/initialized"} ) + "\n") proc.stdin.flush() # 3. 拉取工具清单 tools = request("tools/list", {}, mid=2)["result"]["tools"] print("🔧 工具:", [t["name"] for t in tools]) # 4. 调用工具 r = request("tools/call", {"name": "add", "arguments": {"a": 40, "b": 2}}, mid=3) print("🧮 add(40,2) =", r["result"]["content"][0]["text"])

跑起来输出：

✅ 握手成功: {'name': 'minimal-server', 'version': '0.1.0'} 🔧 工具: ['add', 'get_discount'] 🧮 add(40,2) = {"result": 42}

看到没？整个协议没有魔法，就是协商 → 通知 → 请求/响应三次交互，和普通 RPC 没有本质区别。

五、Streamable HTTP：无状态化的关键设计

本地用 stdio，云端就得上 HTTP。2025-06-18 规范把 SSE 升级为Streamable HTTP：普通请求走 POST 立即返回 JSON；需要流式时服务端用 `Content-Type: text/event-stream` 推事件，客户端拿到 `sessionId` 后在后续请求头里带上 `Mcp-Session-Id` 保持会话。

生产环境最关键的一条：POST 请求必须是幂等、无状态的，这样前面挂多少个 Nginx/LB 都不怕。典型请求长这样：

POST /mcp HTTP/1.1 Host: mcp.example.com Content-Type: application/json Accept: application/json, text/event-stream Mcp-Session-Id: a1b2c3d4e5 {"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_discount","arguments":{"sku":"SKU-001"}}}

服务端响应（流式）：

HTTP/1.1 200 OK Content-Type: text/event-stream event: message data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"{\"sku\": \"SKU-001\", \"discount\": 0.85}"}]}}

这也是为什么 2026 年 MCP 网关（Gateway）成为企业标配——它把散落的 stdio Server 统一转换成 HTTP 出口，还能顺手做鉴权、限流、审计。

六、底层容易被忽略的三个坑

1.协议污染：stdio 模式下任何多余输出（print 调试、第三方库的日志）都会让 Client 解析崩溃。所有日志走 stderr，这是线上事故第一高发点。

2.超时与泄漏：流式模式下 SSE 长连接要设置空闲超时，Client 用完必须释放，否则服务端连接数只会涨不会跌——典型的生产事故。

3.安全边界：MCP 统一了"接线"却不会自动装"保险丝"。2026 年安全研究界已定义出 stealth memory injection（隐形内存注入）等攻击手法。生产环境必须做到：最小权限账号、只读优先、root 目录限定、OAuth Token 短期化、高危操作默认禁止。

七、展望：MCP + A2A + ACP

2026 年的共识是三层协议各司其职：MCP 让 Agent 有了"手"（连工具），A2A 让 Agent 有了"嘴"（Agent 之间对话），ACP 让 Agent 有了"神经"（跨设备控制）。加上智能体操作系统（AOS）开始把编排、治理、资源管理标准化，底层协议栈正在变成下一代 AI 基础设施的地基。

对开发者来说，现在正是把 MCP 从"会用"升级到"懂底层"的最好时机——协议本身很薄，一小时就能读完规范，但它会成为未来五年最值钱的技能之一。

---

参考：MCP 官方规范（2025-06-18 版）| Anthropic MCP 生态数据（2026.08）| 北京市智能体政策（2026.07）