深入 MCP TypeScript SDK 的会话、状态与水平扩展：从无状态 HTTP 服务到跨节点事件总线

深入 MCP TypeScript SDK 的会话、状态与水平扩展：从无状态 HTTP 服务到跨节点事件总线

【免费下载链接】typescript-sdkThe official TypeScript SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/GitHub_Trending/ty/typescript-sdk

createMcpHandler默认按请求构建全新服务实例、请求之间不保留任何状态，因此 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.ts：

import { 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（源码第 831～838 行）；
• 之后的请求若未携带Mcp-Session-Id头，将被拒绝——会话校验与协议版本校验在非初始化请求上强制执行。

而在客户端侧，SDK 的StreamableHTTPClientTransport会在每个请求上自动回传该响应头，无需任何配置。

手工路由：一张 Map 管理“一个传输实例 = 一个会话”

一个传输实例就是一个会话，因此有状态部署的核心数据结构就是一张映射表：当initialize到来时构建传输实例并存入 Map，之后每个请求都根据其Mcp-Session-Id路由到对应传输实例。下面的 Express 路由覆盖全部三种动词——POST（RPC 请求）、GET（通知的 SSE 流）、DELETE（结束会话），应用本身的搭建见 Serve with Express：

const sessions = new Map<string, 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.ts：

interface EventStore { storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId>; getStreamIdForEventId?(eventId: EventId): Promise<StreamId | undefined>; replayEventsAfter( lastEventId: EventId, { send }: { send: (eventId: EventId, message: JSONRPCMessage) => Promise<void> } ): Promise<StreamId>; }

• storeEvent(streamId, message)：持久化一条消息并返回其事件 ID；
• replayEventsAfter(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 并返回生成的事件 ID；
• replayEventsAfter按事件 ID 字典序排序后，从lastEventId之后开始，仅重放同一流上的事件，通过send(eventId, message)逐条投递。

文档同时提醒：这是纯内存实现，仅适用于单进程场景；生产环境应换成数据库等持久化存储。

跨节点扩展：两种有状态路线与一条事件总线

无状态默认：负载均衡器直接搞定

如前所述，无状态模式是“扩展故事”本身：每个节点从同一工厂构建新实例、请求之间零共享，因此放到任何负载均衡器后面即可，无需任何额外配置。

有状态（2025 时代）节点的两条扩展路线

有状态节点把会话保存在进程内存中，因此扩展时有两条路：

1. 持久化存储路线：保留sessionIdGenerator，让所有节点指向同一个eventStore。这样任何节点都能从共享存储恢复断开的流，节点间无需共享会话本体。
2. 本地状态 + 消息路由路线：保持每个节点的本地会话，把每个会话的流量路由到拥有它的节点——可以用负载均衡器的会话粘滞，也可以在节点之间做 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)把事件转发给 broker；
• subscribe(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 总线）交给每个节点的createMcpHandler：

const 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，缺失头返回400；
• eventStore让断开的 SSE 流可恢复：客户端以Last-Event-ID重连，传输层重放错过的消息；参考实现见 examples/shared/src/inMemoryEventStore.ts；
• subscriptions/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