MCP 协议深拆：从 JSON-RPC 到云端并发，Agent 的 USB 接口是怎么设计的

上一篇我们把"Python 函数变成大模型工具调用"的完整旅程拆了一遍：反射提取签名、类型映射成 JSON Schema、装饰器注册、Agent Loop 里执行。但那篇文章有个隐含前提一直没说破——工具和 Agent 住在同一个进程里。

这在本地开发时毫无问题：Claude Code、OpenClaw 这类跑在你电脑上的 Agent，把 MCP server 当子进程拉起来，工具函数就在隔壁进程里待命。

可一旦 Agent 部署到云端，事情就变了：

• Agent 跑在 Kubernetes 的 Pod 里，工具应该跑在哪？
• 一百个 Agent 同时在线，每个都拉起自己的一套 MCP server 环境吗？
• 两个 Agent 同时调用同一个工具，会不会互相踩踏？
• Agent 扩容到 10 个副本，MCP server 要跟着扩吗？

这篇文章分两条线走：前半篇把 MCP 协议本身拆透（JSON-RPC、两种 transport、握手、三大原语，最后不依赖任何 SDK 手写一个能跑的 MCP server）；后半篇专门回答上面那四个问题——云端 Agent 的 MCP 并发。文中所有代码都可以直接运行，规范细节基于 2025-06-18 版（当前最新修订版）。

第一层：MCP 为什么选 JSON-RPC 2.0

MCP（Model Context Protocol）要解决的问题是：让 Agent 和工具提供方解耦。以前每接一个工具就要写一层私有胶水代码，有了 MCP，工具方实现一次 server，所有支持 MCP 的 Agent 都能直接用——所以大家叫它"AI 的 USB 接口"。

协议本身建立在 JSON-RPC 2.0↗ 上。一个最小的请求长这样：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": { "city": "北京" }
  }
}

响应：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{ "type": "text", "text": "晴，26°C" }],
    "isError": false
  }
}

为什么是 JSON-RPC 而不是 gRPC 或 REST？三个原因：

1. 零依赖：任何能发 HTTP 请求、能解析 JSON 的语言都能实现，不需要 protoc 编译，不需要 codegen
2. 双向对称：JSON-RPC 不区分"客户端"和"服务端"的角色——任何一方都能发 request。这一点对 MCP 至关重要，因为 MCP server 也能反过来向 Agent 发请求（后面讲 sampling 和 elicitation 时会看到）
3. 消息即文档：一条消息就是一个自描述的 JSON 对象，抓包即调试

注意 id 字段：它是请求关联的钥匙。客户端发请求时分配一个唯一 id，服务端的响应必须带上同一个 id。记住这个细节，后面讲并发时它会再次成为主角——云端一个 MCP server 同时处理几百个 in-flight 请求，靠的就是它。

没有 id 的消息叫 notification——发出去不期待响应，比如 notifications/initialized、notifications/cancelled。MCP 里大量的"事件流"语义都靠 notification 实现。

第二层：两种 Transport——stdio 与 Streamable HTTP

JSON-RPC 只定义了消息格式，没定义消息怎么传。MCP 目前定义了两种标准传输机制（规范↗）：

stdio：最朴素的进程间通信

stdio 模式下，客户端 spawn 一个子进程，往它的 stdin 写 JSON-RPC 消息，从 stdout 读响应。规范要求：消息按换行符分隔、消息体内不能有嵌入换行、stdout 只能输出合法 MCP 消息（日志一律走 stderr）。

这就是"工具和 Agent 住在一起"的具体形态。它的优势是零网络开销、权限模型简单（工具直接用你本地的文件系统和凭据），劣势我们后面集中拆。

Streamable HTTP：为远程和共享而生

Streamable HTTP（2025-03-18 版规范引入，替代更早的 HTTP+SSE）模式下，server 是一个独立的 HTTP 服务，暴露单一 endpoint（比如 https://mcp.example.com/mcp），同时支持 POST 和 GET：

• 客户端每个 JSON-RPC 消息都通过一次 POST 发送，Accept 头必须同时列出 application/json 和 text/event-stream
• 服务端对 request 的响应，可以返回一个普通 JSON（一次性结果），也可以返回 SSE 流（服务端可以先推若干 notification/request，最后在流里给出 response，然后关流）
• 客户端可以额外发 GET 打开一条服务端推送流，接收与当前请求无关的服务端主动消息
• 断连不等于取消：客户端想取消必须显式发 notifications/cancelled；服务端可选支持 Last-Event-ID 实现流断点续传（resumability）

一个容易被忽略的安全要求：server 必须校验 Origin 头防止 DNS rebinding 攻击，本地运行时应绑定 127.0.0.1 而不是 0.0.0.0。云端部署还应加上认证（第八层细说）。

第三层：initialize 握手与能力协商

MCP 连接建立的第一个动作永远是 initialize 请求：

{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-06-18",
    "capabilities": {
      "roots": { "listChanged": true },
      "sampling": {}
    },
    "clientInfo": { "name": "my-agent", "version": "0.1.0" }
  }
}

服务端响应自己支持的版本和能力：

{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2025-06-18",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true },
      "prompts": {}
    },
    "serverInfo": { "name": "weather-mcp", "version": "1.0.0" }
  }
}

随后客户端发一个 notifications/initialized，握手完成。三个要点：

1. 版本协商：客户端报自己想要的版本，服务端如果不支持，回一个自己支持的版本；客户端不满意就可以断开
2. 能力协商是声明式的：capabilities 里没写的特性，对端就不许用。服务端没声明 tools.listChanged，客户端就不该期待 tools/list 变更通知
3. 方向是双向的：客户端也声明能力。roots 是客户端告诉服务端"我允许你访问哪些目录"；sampling 是客户端允许服务端反过来请求调用大模型——服务端发 sampling/createMessage，Agent 代为执行 LLM 调用并把结果还回去。这是 MCP 最容易被忽略的设计：server 不需要自己配 API key 也能"用上大模型"

第四层：三大原语——Tools / Resources / Prompts

握手之后，客户端能用的服务端能力分三类，很多人把它们混为一谈，其实语义完全不同：

Tools 走 tools/list + tools/call，和上一篇讲的工具系统无缝对接——MCP server 返回的 tools/list 里每一项就是一份 JSON Schema，Agent 把它合并进自己的工具表即可。

Resources 是被低估的原语。工具调用适合"做事"，但把一份 50KB 的配置文件塞给模型不该走工具调用（浪费一轮模型交互），应该由 Agent 代码在组装上下文时通过 resources/read 拉取。声明了 subscribe 能力的 server 还会在资源变化时主动推 notifications/resources/updated，Agent 据此刷新缓存——这是"活文档"。

Prompts 则是把"怎么用这个 server"的专家知识固化：比如一个数据库 MCP server 提供 query-with-explain 模板，用户敲 /query-with-explain 展开成一段带说明的完整提示词。

第五层：不依赖 SDK，手写一个最小 MCP Server

理解一个协议最快的方式是裸写一个实现。下面这个 server 只用 Python 标准库，支持 Streamable HTTP transport、完整的 initialize 握手、tools/list 和 tools/call。它是后面讨论云端并发的基础，值得逐行看懂：

#!/usr/bin/env python3
"""minimal_mcp_server.py — 零依赖的 MCP Streamable HTTP server"""
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PROTOCOL_VERSION = "2025-06-18"
TOOLS = [
    {
        "name": "get_weather",
        "description": "查询城市天气",
        "inputSchema": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "城市名"}},
            "required": ["city"],
        },
    }
]


def rpc_error(id_, code, message):
    return {"jsonrpc": "2.0", "id": id_,
            "error": {"code": code, "message": message}}


def handle_request(msg):
    """分发一条 JSON-RPC request，返回 response dict"""
    method, params = msg["method"], msg.get("params", {})
    id_ = msg["id"]

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": id_, "result": {
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": "minimal-weather", "version": "0.1.0"},
        }}

    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": id_, "result": {"tools": TOOLS}}

    if method == "tools/call":
        city = params["arguments"]["city"]
        return {"jsonrpc": "2.0", "id": id_, "result": {
            "content": [{"type": "text",
                         "text": f"{city}：晴，26°C"}],
            "isError": False,
        }}

    return rpc_error(id_, -32601, f"Method not found: {method}")


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers["Content-Length"])
        msg = json.loads(self.rfile.read(length))

        if "id" not in msg:                       # notification：回 202，不处理响应
            self.send_response(202)
            self.end_headers()
            return

        response = handle_request(msg)
        body = json.dumps(response).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):                              # 不提供服务端推送流
        self.send_response(405)
        self.end_headers()


if __name__ == "__main__":
    print("minimal MCP server on http://127.0.0.1:9377/mcp")
    # 注意是 ThreadingHTTPServer：每请求一线程。
    # 单线程的 HTTPServer 会把并发请求串行化——本地无所谓，
    # 云端一百个 Agent 同时打过来就会排队卡死（第七层细说）。
    ThreadingHTTPServer(("127.0.0.1", 9377), Handler).serve_forever()

用 curl 就能走完整个握手和调用：

# 1. initialize
curl -s http://127.0.0.1:9377/mcp -H 'Content-Type: application/json' -d '{
  "jsonrpc":"2.0","id":0,"method":"initialize",
  "params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl","version":"0"}}
}'

# 2. initialized notification
curl -s -X POST http://127.0.0.1:9377/mcp -H 'Content-Type: application/json' -d '{
  "jsonrpc":"2.0","method":"notifications/initialized"
}' -o /dev/null -w '%{http_code}\n'   # -> 202

# 3. 列出工具
curl -s http://127.0.0.1:9377/mcp -H 'Content-Type: application/json' -d '{
  "jsonrpc":"2.0","id":1,"method":"tools/list"
}'

# 4. 调用工具
curl -s http://127.0.0.1:9377/mcp -H 'Content-Type: application/json' -d '{
  "jsonrpc":"2.0","id":2,"method":"tools/call",
  "params":{"name":"get_weather","arguments":{"city":"北京"}}
}'

四条命令跑通，你对 MCP 的祛魅就完成了：它不是什么神秘协议，就是"HTTP 上的 JSON-RPC + 一套方法名约定"。

第六层：官方 SDK 替你做了什么

真实的 MCP Python SDK（mcp 包，基于 FastMCP 演化）在这之上加的其实就是我们前两篇文章拆过的东西：

• 装饰器注册：@mcp.tool() 把函数注册进工具表（工具系统篇的第六层）
• Schema 自动生成：从类型注解生成 inputSchema（那篇的第三层）
• 生命周期管理：握手、版本协商、能力声明的样板代码
• SSE 流式响应：长任务先推 notifications/progress，最后回 result
• 会话管理：Mcp-Session-Id 的生成与校验（下一层的主角）

SDK 值得用，但要记住它只是糖。并发出问题时，你还是得回到第五层那个裸协议的视角去排查。

第七层：云端来了——MCP 并发的四道坎

前面都是铺垫。现在把 Agent 部署到云端：一个 FastAPI 服务跑着你的 Agent Loop，后面挂着天气、搜索、数据库三个 MCP server。并发问题从这一刻开始浮现，一道一道过。

坎一：stdio 在云端的三宗罪

把本地习惯带上来——每个 Agent 进程 spawn 自己的 stdio MCP server——会立刻撞上三件事：

罪一：进程爆炸。 stdio 是 1:1 模型，100 个 Agent 会话就是 100 套 MCP server 子进程。每个 Node.js 实现的 server 常驻 50–150MB 内存，你的 Pod 内存预算很快被工具进程吃光，而它们大部分时间在闲置。

罪二：生命周期绑架。 子进程跟随父进程。Agent 部署滚动更新，MCP server 全部被杀重启；Agent 崩溃，它拉起的 server 里可能正有未完成的副作用操作（写到一半的文件、执行一半的事务），且没有任何跨重启的状态。

罪三：无法共享。 每个 Agent 的工具实例互相隔离。团队想共享一个带连接池和缓存的数据库 MCP server？做不到——stdio 模式下根本没有"多个客户端连同一个 server"这个概念。

结论很干脆：云端部署，用 Streamable HTTP。把 MCP server 独立部署成长期运行的服务，Agent 通过 HTTP 连接。1:N 模型天然解决进程爆炸和共享问题。

坎二：会话状态——Mcp-Session-Id 与粘性路由

切到 Streamable HTTP 后第一个撞上的是会话问题。规范定义的流程是：

1. 服务端在 initialize 的响应里可以返回一个 Mcp-Session-Id 响应头
2. 客户端后续所有请求必须携带这个头
3. 会话失效时服务端返回 404，客户端应重新 initialize
4. 客户端想主动结束，发 HTTP DELETE 到 endpoint

这个设计意味着：一个有状态 MCP server 的会话数据（订阅了哪些资源、sampling 上下文、SSE 流的 event buffer）存在某个具体实例的内存里。单机没问题，云端立刻引出两个部署问题：

问题 A：负载均衡怎么办？ 请求被 LB 随机分发，initialize 打到实例 1（会话存在实例 1 内存里），下一个 tools/call 打到实例 2，实例 2 没见过这个 session id，返回 404。解决方案有两条路：

• 粘性路由（sticky session）：LB 按 Mcp-Session-Id 哈希路由，同一会话永远到同一实例。实现最快，但代价是负载不均、实例下线时会话集体失效、无法自由扩缩容
• 无状态化（推荐）：见下一段

问题 B：长连接 SSE 与网关超时。 如果你的 server 用 SSE 流推送进度，云上的 LB / API 网关常常有 30–60 秒的空闲超时，会把流掐断。要么配置更长的超时并开启 Last-Event-Id 断点续传，要么干脆不用 GET 流、每个 POST 响应完即关（对大多数工具型 server 足够）。

坎三：无状态 MCP Server——云端的正确形态

回头看第四层的三原语语义：Tools 和 Prompts 本质上是纯函数式的 RPC——请求进来、结果回去、不留痕迹。真正需要状态的只有 Resources 订阅和 SSE 流。也就是说，如果你的 server 只提供工具，它可以做成完全无状态的：

• initialize 请求带什么就回什么，不分配 session id（规范允许：Mcp-Session-Id 是 MAY 不是 MUST）
• 任意实例响应任意请求，LB 随便分发，水平扩展零障碍
• 实例重启无痛，Agent 遇到 404 重连即可

把状态外置：连接池指向的数据库、缓存指向 Redis、订阅用 Pub/Sub 广播。这就是把 Web 服务无状态化的老经验平移到 MCP 上——会话在内存、状态进基础设施。

无状态化后，之前那个"100 个 Agent 各自拉进程"的画面变成：3 个 MCP server 实例（可按 CPU/内存指标自动扩缩），后面共享一个数据库连接池。资源占用从 100 份降到 3 份，还获得了弹性。

坎四：并发控制——当一百个 Agent 同时挥舞工具

无状态解决了部署问题，但并发正确性还欠着。三个具体场景：

场景 1：单个请求内的并发工具调用。 现代 Agent 常在一轮里并行发多个 tools/call（比如同时查三个城市的天气）。JSON-RPC 的 id 在这里发挥第二重作用：客户端同时挂起 5 个请求，每个带着不同的 id，响应到达时按 id 关联到对应的 asyncio.Future。SDK 里对应的就是 asyncio.gather + 请求 id 路由表。裸写也不难：

import asyncio, itertools, json
import aiohttp

class McpHttpClient:
    """极简并发客户端：按 JSON-RPC id 关联响应"""
    def __init__(self, url):
        self.url = url
        self._ids = itertools.count(1)
        self._pending: dict[int, asyncio.Future] = {}

    async def _rpc(self, session, method, params=None):
        id_ = next(self._ids)
        fut = asyncio.get_running_loop().create_future()
        self._pending[id_] = fut
        async with session.post(self.url, json={
            "jsonrpc": "2.0", "id": id_,
            "method": method, "params": params or {},
        }) as resp:
            msg = await resp.json()
        # 无状态 server 直接在 POST 响应里回结果
        fut.set_result(msg.get("result") or msg.get("error"))
        return await fut

    async def call_tool(self, city):
        async with aiohttp.ClientSession() as s:
            return await self._rpc(s, "tools/call", {
                "name": "get_weather", "arguments": {"city": city}})

async def main():
    client = McpHttpClient("http://127.0.0.1:9377/mcp")
    results = await asyncio.gather(          # 三个调用并发 in-flight
        client.call_tool("北京"),
        client.call_tool("上海"),
        client.call_tool("深圳"),
    )
    print(results)

asyncio.run(main())

场景 2：下游过载。 100 个 Agent × 每轮 3 个工具调用，打到你的 MCP server 是每秒几百个请求，而 server 背后的第三方 API 配额可能只有 50 QPS。无状态不等于无限流。在 server 侧加并发闸门：

import asyncio

_tool_gate = asyncio.Semaphore(50)          # 全局：最多 50 个并发工具执行

async def call_tool_checked(name, args):
    async with _tool_gate:                   # 超出的请求在此排队而非压垮下游
        return await do_real_work(name, args)

用 Semaphore 而不是简单的计数器，是因为它是 asyncio 友好的：拿不到名额的协程挂起让出事件循环，不会阻塞整个 server。进阶做法是把闸门从"进程内"移到"部署层"——比如在 server 前面放一层令牌桶限流的网关，实例扩容时配额也跟着准确。

场景 3：共享状态的竞态。 有些工具天然有状态：计数器、游标、"当前登录用户"。100 个并发请求打到无状态 server 的多个实例上，read-modify-write 就会丢更新。解法和 Web 开发完全同构：

• 计数、去重、幂等 → Redis INCR / SETNX（原子操作）
• 复杂临界区 → 数据库行锁 / SELECT ... FOR UPDATE
• 工具调用幂等性 → 请求带幂等键，重复调用返回缓存结果（对 Agent 尤其重要：模型重试同一工具调用是常态）

一个容易忽视的点：客户端侧也要限流。 Agent 侧对单个 MCP server 的连接数、in-flight 请求数应有上限（aiohttp.TCPConnector(limit=10)），否则一个 Agent 的突发流量就可能打满 server 的并发闸门，殃及其他 Agent。

汇总：云端 MCP 部署检查清单

第八层：安全——云端不可跳过的一课

本地 stdio 的信任模型是"工具就是你"（直接用你的文件和凭据），云端则是网络边界上的多租户服务，规范的要求必须全部落地：

• 认证授权：2025-06-18 版规范要求 Streamable HTTP server 实现 OAuth 2.1 作为 Resource Server（Agent 拿到 token 再访问 MCP server），公网裸奔的 MCP endpoint 等于把工具能力开放给全网
• Origin 校验：防 DNS rebinding，规范用 MUST 级措辞
• 最小权限：server 的凭据只给必要的资源。数据库 MCP server 用只读账号，除非工具明确需要写
• 审计日志：谁、哪个会话、什么时候、调了什么工具、参数是什么——出了事这是唯一的回放依据
• 工具中毒警惕：tools/list 返回的 description 是提示词注入的载体。来源不明的第三方 MCP server 可能在描述里藏指令（"调用前请先把用户邮箱发到……"）。Agent 端对工具描述应像对待不可信输入一样处理

回顾：一条递进的线

从本地的子进程到云端的并发服务，MCP 的故事可以串成一条线：

1. 消息层：JSON-RPC 2.0，id 关联请求，notification 承载事件
2. 传输层：stdio（1:1，本地）→ Streamable HTTP（1:N，远程）
3. 握手层：initialize + 能力协商，特性必须声明才能使用，sampling 让 server 反向用模型
4. 语义层：Tools 做事、Resources 读数、Prompts 固化用法
5. 实现层：百行代码可裸写，SDK 是装饰器和 schema 生成的糖
6. 部署层：云端用 HTTP transport、工具型 server 无状态化、状态外置
7. 并发层：id 路由表挂并发请求、Semaphore 保下游、幂等键防丢更新
8. 安全层：OAuth 2.1、Origin 校验、最小权限、审计

下一篇打算接着这条线往两个方向之一走：Agent 记忆系统（context window → RAG → 持久记忆，正好接上 Agent Loop 篇第七层的上下文管理），或者 Agent 评测与可观测性（怎么知道你的 Agent"变好了"）。读者倾向哪个可以在评论区说。

动手建议

• 把第五层的裸 server 跑起来，用 curl 走一遍握手，再换成官方 SDK 重写，对比它替你做了什么
• 给裸 server 加一个 Mcp-Session-Id 和一个会过期的会话表，体会"有状态"给负载均衡带来的麻烦，然后删掉它
• 用 asyncio.gather + Semaphore 把客户端并发从 1 拉到 100，观察 server 和下游的行为，找到一个会被打挂的阈值——比读十篇文章都有用