Dispatcher 抽象与 ServerRunner 内核
这是全书主线。v2 最大的架构改动是引入
Dispatcher这层「协议无关的收发通道」,并把 MCP 语义收进一个每连接的内核ServerRunner。读懂这一章,就读懂了 v2。
1. 为什么要有 Dispatcher 这层
它要解决的问题
v1 里,「JSON-RPC 编码」和「MCP 语义」是缠在一起的。想让 server 跑在别的编码上(比如内存直调做测试)、或想让传输和语义各自演进,都很难。
思路:切一刀,分成两个世界
在中间切一层,约定一个极简接口:
┌──────────────── 上面的世界:懂 MCP ────────────────┐
│ 请求/结果模型、能力协商、Context、类型校验 │
└───────────────────────┬───────────────────────────┘
│ Dispatcher 接口
│ · send_raw_request(method:str, params:dict) -> dict
│ · notify(method:str, params:dict)
│ · run(on_request, on_notify) 收循环
┌───────────────────────┴───────────────────────────┐
│ 下面的世界:懂线格 JSON-RPC / 内存直调 / (将来 gRPC)│
└────────────────────────────────────────────────────┘
关键约束(shared/dispatcher.py 文件头注释明确写着):Dispatcher 故意不懂 MCP——方法名就是字符串,参数和结果就是 dict[str, Any]。MCP 类型层坐在它上面,线格编码坐在它下面。
好处
- 同一套内核
ServerRunner,既能跑在JSONRPCDispatcher(真传输)上,也能跑在DirectDispatcher(内存直调,零序列化)上——测试和「进程内嵌 server」因此变得极简。 Dispatcher是typing.Protocol(结构类型),换实现不用改上层。
2. 三个协议(Protocol),先认清
shared/dispatcher.py 定义了三个结构类型,层层递进:
| Protocol | 是什么 | 关键方法 |
|---|---|---|
Outbound | 「能往对端发东西」的最小面。 | send_raw_request / notify |
DispatchContext | 处理一条入站消息时给 handler 的上下文,同时是 Outbound(可反向发)。 | 上面两个 + transport / can_send_request / request_id / progress |
Dispatcher | 完整的双工通道,同时是 Outbound。 | 上面 + run(on_request, on_notify) |
Outbound 是共用的「发」能力:顶层的 Dispatcher 是它(主动发),入站处理时的 DispatchContext 也是它(在处理请求的同时反向发,即 back-channel 反向通道)。
DispatchContext.can_send_request 为 False 时(传输无反向通道、或该入站请求已处理完),send_raw_request 抛 NoBackChannelError(shared/dispatcher.py DispatchContext.can_send_request 文档)。
3. run(on_request, on_notify):收循环的形状
整个「服务端处理请求」的入口就是 Dispatcher.run。它的契约(shared/dispatcher.py Dispatcher.run):
- 驱动收循环直到底层通道关闭。
- 每条入站请求在自己的任务里派发给
on_request;返回的 dict(或抛的MCPError)作为响应发回。 - 入站通知给
on_notify。 - 就绪后调
task_status.started(),让调用方能await tg.start(dispatcher.run, ...)。
on_request / on_notify 是什么?就是内核 ServerRunner 提供的两个回调(下一节)。所以整个架构是:
Dispatcher.run(收循环)
│ 每条入站消息
▼
ServerRunner.on_request / on_notify ← 内核在这里接管
4. JSONRPCDispatcher:生产实现
shared/jsonrpc_dispatcher.py:250 JSONRPCDispatcher 是走真传输的实现。它扛下四件脏活(文件头注释):请求 id 关联、收循环、每请求任务隔离、取消/进度接线,以及唯一的「异常→线格错误」边界。
请求 id 关联
出站请求要能等到对应的响应。做法是维护一张「待决表」:发请求时分配 id(_allocate_id,jsonrpc_dispatcher.py:749)、登记一个 _Pending(jsonrpc_dispatcher.py:119),收到同 id 的响应时唤醒。
有个跨 SDK 对齐的小陷阱:coerce_request_id(shared/dispatcher.py:coerce_request_id)把字符串化的整数 id 转回 int——"7" 和 7 在关联上算同一个 id(和 TS SDK 一致)。
每请求任务隔离
run(jsonrpc_dispatcher.py:465)收到入站请求后 _spawn(jsonrpc_dispatcher.py:654)一个新任务跑 _handle_request(jsonrpc_dispatcher.py:684)。这样慢请求不挡后来的、一个 handler 崩了不拖垮收循环。
异常→线格错误的单一边界
所有 handler 异常都在这里被归一化成线格 ErrorData。映射规则见 handler_exception_to_error_data(jsonrpc_dispatcher.py:83):
| handler 抛的 | 变成的线格错误 |
|---|---|
MCPError | 它自带的 ErrorData |
pydantic ValidationError | INVALID_PARAMS(不把 pydantic 文本泄到线上) |
| 其他 | 返回 None,由各调用方兜底(此处 pin code=0 兼容 v1) |
取消与进度
- 取消: 出站请求带
timeout,超时会抛并发notifications/cancelled;cancel_on_abandon控制放弃时是否礼貌通知对端(initialize设False,协议禁止取消它)。入站方向,收到notifications/cancelled默认"interrupt"取消 handler 作用域(PeerCancelMode,jsonrpc_dispatcher.py:77)。 - 进度: 请求的
_meta.progressToken若在,DispatchContext.progress(...)才真发notifications/progress,否则是 no-op(progress_token_from_params,jsonrpc_dispatcher.py:100)。
5. DirectDispatcher:内存直调实现
shared/direct_dispatcher.py 是最简可能的实现:一侧的请求直接调另一侧的 on_request,没有序列化、没有 JSON-RPC 框、没有流。存在意义(文件头注释):
- 证明
Dispatcher协议不依赖 JSON-RPC 也能实现; - 给上层(
ServerRunner/Context/Connection)提供快速测试底座; - 把 server 嵌进同进程,省掉 JSON-RPC 开销。
create_direct_dispatcher_pair(direct_dispatcher.py:294)造出对接好的一对。Client(mcp) 直连内存 server 对象走的就是这条路(第 4 章)。
6. ServerRunner:每连接的内核
server/runner.py:148 ServerRunner 是「一个客户端连接对应一个实例」的 handler 内核。它把 on_request / on_notify 交给 Dispatcher 的收循环。核心是 _on_request(runner.py:166)。
一次请求在内核里的五步
_on_request 里,真正的处理逻辑在内层 _inner,被中间件链包着。顺序(runner.py:176-224):
入站 (method, params)
│
① 造 Context: _make_context 把 dctx + connection 包成 ServerSession/ctx
│
② 中间件链(outermost-first),最外层默认是 OpenTelemetry span
│
③ _inner:
├ 若是 spec 方法:按【协议版本】校验 params(validate_client_request)
├ 若 method == "initialize":内核自己处理握手(不查 handler 表)
├ 查 handler 表:get_request_handler(method);没有 → METHOD_NOT_FOUND
├ 未初始化且非豁免方法 → INVALID_PARAMS
├ 用 entry.params_type 校验 params(缺 params 按 {} 校验)
├ await handler(ctx, typed_params)
└ 填缓存提示 → 序列化结果
│
④ 回一个 dict 给 Dispatcher,由它发回对端
几个诚实的细节:
- 按版本校验:
_methods.validate_client_request(method, version, params)用连接当前协商的protocol_version选对应 schema 纪元校验(runner.py:184)。方法在该版本不存在 →METHOD_NOT_FOUND。 initialize特殊:握手由内核自己管,注册它会被 lowlevelServer.add_request_handler拒绝(server/lowlevel/server.py:476,if method == "initialize": raise ValueError)。想观察/包裹握手用Server.middleware。- 缺 params 的契约:无
params的消息拿{}去校验;有必填字段的模型会拒(INVALID_PARAMS),全可选的模型带默认值进 handler——handler 永不收到None(add_request_handlerdocstring)。
通知(notification)的处理更宽容
_on_notify(runner.py:240):畸形通知丢弃 + 记日志而非报错;handler 崩了也只 logger.exception 不让它取消 Dispatcher 的任务组(runner.py _on_notify 末尾)。因为通知本就是「尽力而为、无响应」。
7. 中间件:唯一的横切点
Server.middleware 是一个列表,_compose_server_middleware(runner.py:289)把它们从外到内包住 _inner。签名是 (ctx, call_next)。默认自带 OpenTelemetryMiddleware(server/lowlevel/server.py:309 附近,self.middleware = [OpenTelemetryMiddleware()]),没装 OTel 导出器时是 no-op。
妙处:序列化和结果 dump 放在链内,所以最外层的 OTel span 也能观测到「handler 返回了非法结果」这种失败(runner.py _inner 里注释)。中间件能改 ctx(重写 method/params)供后续链使用——因为链是「拿运行时的 ctx」而非闭包死绑。
8. lowlevel Server:一张 handler 表
server/lowlevel/server.py:132 Server 本身很薄:两个字典 _request_handlers / _notification_handlers,每项是 HandlerEntry(params_type, handler)(server/lowlevel/server.py:86)。构造时把传入的 on_* 回调按 spec 方法名批量登记(server/lowlevel/server.py _spec_requests 列表)。
它也管能力协商(get_capabilities / create_initialization_options)和 run(进 lifespan → serve_dual_era_loop)。run 见下一章的传输部分。
9. 本章要点
Dispatcher= 协议无关的收发通道(字符串方法 + dict);MCP 语义全在它上面。- 两个实现:
JSONRPCDispatcher(真传输,管 id 关联/取消/进度/异常边界)和DirectDispatcher(内存直调)。 ServerRunner是每连接内核,一次请求走「造 Context → 中间件 → 按版本校验 → 查表 → 派发 → 序列化」。initialize握手由内核独占,不能当普通 handler 注册。
→ 下一章:这些 Dispatcher 底下真正搬字节的传输长什么样。