跳到主要内容

协议核心与传输:Protocol 与 Transport

本章讲底层骨架:Protocol 抽象类怎么在「不懂语义、只搬字节」的 Transport 之上,搭出请求/响应配对、超时、取消、进度通知;以及一条入站请求在 _onrequest 里经历的完整流水线。Client 和 Server 都继承自 Protocol。

1. 它要解决的小问题​

JSON-RPC 本身只是「请求带 id、响应回相同 id」的约定。但要做一个可靠的双向连接,你还需要:把异步响应跟原请求对上号、超时后报错、支持取消在途请求、转发进度通知、连接关闭时清理所有挂起的 promise。Protocol 把这些通用机制做一次,Client/Server 只填角色差异。

2. 思路/直觉:几张 Map 撑起整个状态机​

Protocol 的全部状态就是几张以 messageId 为键的 Map:

// packages/core-internal/src/shared/protocol.ts:519-525 —— 核心状态
private _requestMessageId = 0;                              // 自增 id 发号器
private _requestHandlers: Map<string, ...>;                 // 方法名 → 入站请求处理器
private _responseHandlers: Map<number, ...>;                // 我方请求 id → 等待响应的回调
private _progressHandlers: Map<number, ProgressCallback>;   // 请求 id → 进度回调
private _timeoutInfo: Map<number, TimeoutInfo>;             // 请求 id → 超时信息

直觉:「我方发出的请求」 在 _responseHandlers 里挂一个回调,等对方用相同 id 回响应时取出来 resolve;「对方发来的请求」 在 _requestHandlers 里按方法名查处理器。两个方向用同一套 messageId 空间但分开的表。

3. 主线一:连接与消息分流​

connect(transport)(protocol.ts:742)做一件事:把 transport 的三个回调接管过来。最关键的是 onmessage,它按消息形态分流:

// 示意,浓缩自 packages/core-internal/src/shared/protocol.ts:760-771
this._transport.onmessage = (message, extra) => {
  if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
    this._onresponse(message);        // 是响应 → 找等待的回调
  } else if (isJSONRPCRequest(message)) {
    this._onrequest(message, extra);  // 是请求 → 找处理器并分发
  } else if (isJSONRPCNotification(message)) {
    this._onnotification(message, extra);  // 是通知 → 找通知处理器
  } else {
    this._onerror(new Error(`Unknown message type: ...`));
  }
};

Transport 接口本身极简——只有 start/send/close 三个方法和 onmessage/onclose/onerror 三个回调,外加可选的 sessionId、setProtocolVersion(transport.ts:107)。它完全不懂 MCP 语义,只是个双向字节管道。这就是「可插拔传输」:stdio、Streamable HTTP、in-memory 都实现同一接口。

4. 主线二:发一个请求(request)​

request()(protocol.ts:1224)有两个重载:spec 方法(如 tools/list)自动从方法名推出结果类型;自定义方法要你传一个结果 schema。内部流程:

  request(req)
     │
     ▼
 ① _resolveOutboundCodec(method)   选出本次该用的 wire codec(见第 3 章)
     │
     ▼
 ② _assertOutboundRequestInEra      若方法不属于该纪元 → 本地抛 SdkError,不上线
     │
     ▼
 ③ 分配 messageId,挂 _responseHandlers / _setupTimeout
     │
     ▼
 ④ transport.send(jsonrpcRequest)
     │
     ▼
 ⑤ 等响应 → 用 codec.decodeResult 解码 → schema 校验 → resolve

超时与取消都挂在这条链上:

  • 超时: _setupTimeout(protocol.ts:694)起一个 setTimeout;maxTotalTimeout 给整个流程封顶。若开了 resetTimeoutOnProgress,每收到一条进度通知就 _resetTimeout(protocol.ts:711)——但仍受 maxTotalTimeout 硬上限约束。
  • 取消: 传入 options.signal,abort 时给对方发 notifications/cancelled(2026 纪元在 Streamable HTTP 上则改为「关闭该请求的 SSE 流」即取消,见 protocol.ts:1362 附近注释)。

5. 主线三:处理一个入站请求(_onrequest)​

这是全文件最密的一段(protocol.ts:886)。一条入站请求要过一条校验流水线才到你的处理器,顺序极讲究:

 入站请求
   │
   ▼
 ① liftWireOnlyMaterial   把 wire-only 材料(_meta 信封、重试字段)从 params 里「抬出」
   │                       —— 让处理器看到的就是干净的 2025 形态
   ▼
 ② 选 codec(_negotiatedWireCodec)
   │
   ▼
 ③ 纪元一致性检查          若传输已分类且分类纪元 ≠ 实例纪元 → 回 −32022 路由错误
   │
   ▼
 ④ 纪元门(era gate)       spec 方法不在本纪元 registry → 回 −32601(即使装了处理器!)
   │
   ▼
 ⑤ 查处理器               没有 → −32601 Method not found
   │
   ▼
 ⑥ checkInboundEnvelope    2026 纪元要求每请求带 _meta 信封,缺失 → −32602
   │                       (故意排在 ④⑤ 之后:方法不存在优先于参数非法)
   ▼
 ⑦ 跑处理器 → codec.encodeResult 编码结果 → transport.send

第 ① 步的「lift」很关键。看 liftWireOnlyMaterial(protocol.ts:210):它把两类东西从 params 里抽走,使处理器只看见 2025 纪元的纯净形态:

// 示意,概念来自 protocol.ts:210-250
// 1) 保留的 _meta 信封键(io.modelcontextprotocol/* 协议版本、客户端信息等)
// 2) 仅在「客户端发起的请求」上保留的多轮重试字段:inputResponses / requestState
// 抽出来后通过 ctx.mcpReq.envelope / .inputResponses / .requestState 暴露给协议层

顺序的精妙(protocol.ts:972-985): 信封校验(⑥)故意排在纪元门和处理器存在性检查之后。设计者的理由:一个未知方法即使同时缺信封,也应该回 −32601(方法不存在)而不是 −32602(参数非法)——「方法存在性」的优先级高于「参数有效性」。

第 ⑦ 步的编码用 try/catch 包住:编码本身可能抛(2026 codec 会强制删字段、stamp resultType),此时回 −32603 Internal error 而不是让请求一直挂到超时(protocol.ts:1070-1077)。

6. 主线四:收到响应(_onresponse)​

相对简单(protocol.ts:1152):用 id 从 _responseHandlers 取回调,清掉超时和进度处理器,然后:成功结果直接 handler(response);错误响应则用 ProtocolError.fromError 还原成带类型的错误再交给回调。

7. 巧妙之处​

  • Promise.resolve().then(...) 兜同步异常: 入站请求和通知分发都用 Promise.resolve().then(() => handler(...)) 起头(protocol.ts:1053、:881)。这样处理器里的同步抛错也会进入 promise 的 catch 链,不会逃逸成未捕获异常。

  • capturedTransport 防串台: _onrequest 一开始就把当前 transport 存进局部变量(protocol.ts:910),后续所有响应都发到这个被捕获的 transport——避免处理器执行期间 transport 被换掉导致响应发错客户端。

  • 连接关闭即结算: _onclose(protocol.ts:784)把所有挂起的 _responseHandlers 用一个 ConnectionClosed 错误统一 reject,并 abort 所有在途处理器的 AbortController——没有 promise 会永远悬着。

  • 通知防抖: debouncedNotificationMethods 选项让同一 tick 内的重复通知(如反复触发的 tools/list_changed)被合并(protocol.ts:526 的 _pendingDebouncedNotifications)。

8. 边界与局限​

  • Protocol 是抽象类,三个 assert*Capability 方法(protocol.ts:1189–1203)留给 Client/Server 实现——能力检查是角色相关的。
  • 严格能力检查默认关闭(enforceStrictCapabilities 默认 false),为兼容早期没正确声明能力的 SDK 版本;注释说未来会改默认为 true(protocol.ts:80)。
  • 进度通知靠把 messageId 当 progressToken:_onprogress(protocol.ts:1117)直接 Number(progressToken) 当 id 查回调,所以这套进度机制和请求 id 体系是绑定的。

9. 横向对比​

「可插拔传输 + 一个协议状态机类」是 JSON-RPC 库的经典分层。MCP SDK 的特色在于把纪元门也压进了这条入站流水线(④):同一个 Protocol 实例,根据协商出的版本,会物理性地拒绝不属于该纪元的方法——这把「协议演进」做成了运行时不变量,而非文档约定。详见 03-wire-codec-eras.md。

10. 代码地图​

主题文件符号
协议抽象类packages/core-internal/src/shared/protocol.tsProtocol
发请求packages/core-internal/src/shared/protocol.tsrequest、_requestWithSchemaViaCodec
收请求/响应/通知packages/core-internal/src/shared/protocol.ts_onrequest、_onresponse、_onnotification
wire-only 抬升packages/core-internal/src/shared/protocol.tsliftWireOnlyMaterial
超时packages/core-internal/src/shared/protocol.ts_setupTimeout、_resetTimeout
关闭结算packages/core-internal/src/shared/protocol.ts_onclose
传输接口packages/core-internal/src/shared/transport.tsTransport、TransportSendOptions
stdio 传输packages/server/src/server/stdio.tsStdioServerTransport