深入 MCP TypeScript SDK 的会话、状态与水平扩展从无状态 HTTP 服务到跨节点事件总线【免费下载链接】typescript-sdkThe official TypeScript SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/GitHub_Trending/ty/typescript-sdkcreateMcpHandler默认按请求构建全新服务实例、请求之间不保留任何状态因此 v2 服务天生无状态、可直接水平扩展。本文以 docs/serving/sessions-state-scaling.md 为核心结合仓库内 NodeStreamableHTTPServerTransport 与 serverEventBus.ts 等源码系统讲解如何用sessionIdGenerator把客户端固定到会话、用EventStore恢复断开的 SSE 流以及如何用共享ServerEventBus让subscriptions/listen跨节点扇出变更通知。读完你将对这套 SDK 的“无状态默认、按需有状态”扩展模型有完整的实战掌握。先理解默认模型无状态即水平扩展在 v2 中createMcpHandler接收一个工厂函数——每次 HTTP 请求都会调用它构建一个全新的McpServer实例请求结束后实例随之销毁handler 本身在请求之间不保留任何状态。正如 docs/serving/http.md 所描述的工厂接收era、authInfo与requestInfo请求上下文注册工具、资源、提示词都应在工厂内部完成而不是放在共享实例上。这套设计带来的直接红利就是水平扩展零成本每个节点都用同一个工厂构建新鲜实例、请求之间互不干扰因此你可以把任意数量的节点放到任何负载均衡器后面——不需要会话粘滞session affinity、不需要共享任何东西、也不需要额外配置。这正是 Serve over HTTP 的完整部署方案。那么为什么还需要“会话”这一章答案在于两类需求你需要支撑 2025 时代的有状态sessionful部署——例如协议要求跨请求保持同一传输实例的旧版客户端你需要在流断开后恢复resume错过的 SSE 消息你需要在多节点之间推送变更通知。下面分别展开这三种能力。把客户端固定到会话sessionIdGenerator 与 Mcp-Session-Id会话session的本质是把一个客户端固定到一个长生命周期传输实例上。需要特别说明的是会话属于手写直连hand-wired的 2025 时代传输——2026-07-28 修订版是每请求模型不存在Mcp-Session-Id参见 Protocol versions。开启会话只需一个生成器在NodeStreamableHTTPServerTransport上sessionIdGenerator选项负责开启会话保持其为undefined则是无状态模式。以下代码来自 sessions-state-scaling.examples.tsimport { NodeStreamableHTTPServerTransport } from modelcontextprotocol/node; import { randomUUID } from node:crypto; const transport new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () randomUUID() });从源码看WebStandardStreamableHTTPServerTransportOptions对该选项的约束是“生成的会话 ID 应当全局唯一且加密安全如安全生成的 UUID、JWT 或密码学哈希”见 packages/server/src/server/streamableHttp.ts 中sessionIdGenerator与配套回调onsessioninitialized的注释。开启后传输层的行为同样可以从 streamableHttp.ts 中handleRequest的实现确认收到initialize时调用sessionIdGenerator生成 ID 写入this.sessionId并通过Mcp-Session-Id响应头返回给客户端如果配置了onsessioninitialized在会话初始化后立即回调该 ID源码第 831838 行之后的请求若未携带Mcp-Session-Id头将被拒绝——会话校验与协议版本校验在非初始化请求上强制执行。而在客户端侧SDK 的StreamableHTTPClientTransport会在每个请求上自动回传该响应头无需任何配置。手工路由一张 Map 管理“一个传输实例 一个会话”一个传输实例就是一个会话因此有状态部署的核心数据结构就是一张映射表当initialize到来时构建传输实例并存入 Map之后每个请求都根据其Mcp-Session-Id路由到对应传输实例。下面的 Express 路由覆盖全部三种动词——POSTRPC 请求、GET通知的 SSE 流、DELETE结束会话应用本身的搭建见 Serve with Expressconst sessions new Mapstring, NodeStreamableHTTPServerTransport(); const route async (req: Request, res: Response) { const sessionId req.headers[mcp-session-id] as string | undefined; if (sessionId sessions.has(sessionId)) { await sessions.get(sessionId)!.handleRequest(req, res, req.body); return; } if (!sessionId isInitializeRequest(req.body)) { const transport new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () randomUUID(), onsessioninitialized: id { sessions.set(id, transport); } }); transport.onclose () { if (transport.sessionId) sessions.delete(transport.sessionId); }; await buildServer().connect(transport); await transport.handleRequest(req, res, req.body); return; } if (sessionId) { // Unknown session id: the client should start a new session. res.status(404).json({ jsonrpc: 2.0, error: { code: -32001, message: Session not found }, id: null }); return; } // No session header on a non-initialize request: the request is malformed. res.status(400).json({ jsonrpc: 2.0, error: { code: -32000, message: Bad Request: Session ID required }, id: null }); }; app.post(/mcp, route); app.get(/mcp, route); app.delete(/mcp, route);这段路由的自我维护机制值得细读创建无会话头且是initialize请求时构建新传输并用onsessioninitialized回调把新生成的 ID 写入 Map清理transport.onclose在会话结束时触发——无论是客户端发来DELETE还是你主动调用transport.close()——都会从 Map 中删除对应条目未知会话 ID → 404-32001 Session not found语义是“客户端应开启新会话”非 initialize 请求却无会话头 → 400-32000 Bad Request: Session ID required语义是“客户端应重新携带它已有的 ID而不是重新 initialize”。两个错误码把“重开会话”和“重发 ID”两种修复路径区分得清清楚楚客户端据此决定重试策略。提示优雅关停进程退出前务必关闭所有已存储的传输实例for (const [, transport] of sessions) await transport.close()。close()会结束会话的 SSE 流并拒绝其挂起的请求避免资源泄漏与悬挂连接。恢复断开的流EventStore 与 Last-Event-ID有状态客户端的GETSSE 流用于接收服务端通知而连接断开期间产生的任何消息都会丢失。EventStore 正是用来填补这个缺口配置了事件存储后传输层会在发送每条 SSE 消息前先从存储中取得一个事件 ID 并打在该消息上。EventStore 契约EventStore是一个两方法契约另有可选方法接口定义位于 packages/server/src/server/streamableHttp.tsinterface EventStore { storeEvent(streamId: StreamId, message: JSONRPCMessage): PromiseEventId; getStreamIdForEventId?(eventId: EventId): PromiseStreamId | undefined; replayEventsAfter( lastEventId: EventId, { send }: { send: (eventId: EventId, message: JSONRPCMessage) Promisevoid } ): PromiseStreamId; }storeEvent(streamId, message)持久化一条消息并返回其事件 IDreplayEventsAfter(lastEventId, { send })把该流上晚于lastEventId的每一条消息重新发送出去getStreamIdForEventId是可选的若不提供SDK 使用replayEventsAfter返回的streamId做流映射。实现应建立在所有节点都能访问的存储上比如基于数据库的databaseEventStore然后与sessionIdGenerator一起传给传输实例const transport new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () randomUUID(), eventStore: databaseEventStore });断线重连的完整闭环当连接断开时客户端会携带它收到的最后一个事件 ID放在Last-Event-ID请求头中重新连接传输层随即重放存储中该 ID 之后的所有事件。SDK 的StreamableHTTPClientTransport会自动完成重连并发送该头同样无需配置。传输层内部还维护了已重放事件 ID 集合见 streamableHttp.ts 中StreamMapping.replayedEventIds的注释避免重放期间流重新注册导致的重复写入。参考实现InMemoryEventStore仓库中的 examples/shared/src/inMemoryEventStore.ts 是一份完整的EventStore参考实现可读性极佳其关键逻辑包括事件 ID 生成${streamId}_${Date.now()}_${Math.random().toString(36).slice(2, 10)}以流 ID 为前缀保证可反解storeEvent将{ streamId, message }存入内部 Map 并返回生成的事件 IDreplayEventsAfter按事件 ID 字典序排序后从lastEventId之后开始仅重放同一流上的事件通过send(eventId, message)逐条投递。文档同时提醒这是纯内存实现仅适用于单进程场景生产环境应换成数据库等持久化存储。跨节点扩展两种有状态路线与一条事件总线无状态默认负载均衡器直接搞定如前所述无状态模式是“扩展故事”本身每个节点从同一工厂构建新实例、请求之间零共享因此放到任何负载均衡器后面即可无需任何额外配置。有状态2025 时代节点的两条扩展路线有状态节点把会话保存在进程内存中因此扩展时有两条路持久化存储路线保留sessionIdGenerator让所有节点指向同一个eventStore。这样任何节点都能从共享存储恢复断开的流节点间无需共享会话本体。本地状态 消息路由路线保持每个节点的本地会话把每个会话的流量路由到拥有它的节点——可以用负载均衡器的会话粘滞也可以在节点之间做 pub/sub 路由。唯一跨节点的东西subscriptions/listen 的 ServerEventBus即使是无状态部署仍有一样东西会跨节点subscriptions/listen的流。这些流投递的是发布在 handler 的ServerEventBus上的变更事件参见 Notifications而默认的 bus 是进程内的——节点 A 上调用handler.notify.toolsChanged()永远无法到达订阅流挂在节点 B 上的客户端。ServerEventBus接口定义在 packages/server/src/server/serverEventBus.ts只有两个方法interface ServerEventBus { publish(event: ServerEvent): void; subscribe(listener: (event: ServerEvent) void): () void; }publish(event)把事件转发给 brokersubscribe(listener)注册一个监听器返回幂等的取消订阅函数。事件本身是类型化的ServerEvent联合类型每种变体恰好对应一条线上通知tools_list_changed→notifications/tools/list_changed、prompts_list_changed→notifications/prompts/list_changed、resources_list_changed→notifications/resources/list_changed、resource_updated→notifications/resources/updated携带 URI。把基于 pub/sub 的实现如 Redis 总线交给每个节点的createMcpHandlerconst handler createMcpHandler(buildServer, { bus: redisBus });之后任意节点上调用handler.notify.resourceUpdated(uri)都会通过共享总线发布事件每个节点再把自己的变更通知投递到本节点持有的开放订阅流上。这样subscriptions/listen就获得了跨节点的扇出能力。注意InMemoryServerEventBus同文件内的默认实现的语义细节publish()同步投递给存活监听器集合某个监听器抛错不会阻断对其他监听器的投递同时它不得把事件回显给发布者自身——默认实现同步投递且监听器从不发布天然满足这一约束。多进程部署时用subscribe/publish挂接你自己的 broker 即可。回顾会话、状态与扩展要点createMcpHandler每请求构建新实例、请求间零状态无状态节点放在任意负载均衡器后面即可扩展无需会话粘滞会话属于手写直连的 2025 时代传输sessionIdGenerator开启会话响应携带Mcp-Session-Id有状态部署维护“每会话一个传输实例”按Mcp-Session-Id头路由每个请求未知 ID 返回404缺失头返回400eventStore让断开的 SSE 流可恢复客户端以Last-Event-ID重连传输层重放错过的消息参考实现见 examples/shared/src/inMemoryEventStore.tssubscriptions/listen跨节点扩展的答案是给每个节点的createMcpHandler传入同一个ServerEventBus让变更事件通过共享 pub/sub 总线扇出到各节点的开放订阅流。【免费下载链接】typescript-sdkThe official TypeScript SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/GitHub_Trending/ty/typescript-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考