第13章 Streamable HTTP：远程流式传输

在第12章中，我们分析了 stdio 传输——一种简单而高效的本地通信方式。但现实世界中的 AI 系统，从多人协作的开发环境到云端部署的 Agent 平台，都需要跨网络的远程通信能力。这就是 Streamable HTTP 传输层要解决的核心问题。

MCP 协议在 2024-11-05 版规范中使用的是 SSE+HTTP POST 的组合方案（HTTP+SSE 传输）：客户端通过一个长连接的 SSE 通道接收服务端消息，再通过独立的 POST 请求发送消息。这种设计虽然可行，但存在明显的架构缺陷——两个独立的 HTTP 连接需要通过额外的机制（如 URL 中嵌入的 endpoint 路径）进行关联，增加了实现复杂度，也限制了部署灵活性。2025-03-26 版规范引入 Streamable HTTP，正是为了取代这一方案。

本章讲的是本专栏锚定的 2025-11-25 版 Streamable HTTP。2026-07-28 版规范对它改动很大——去掉了协议级会话、GET 流和 SSE 断线续传——涉及的地方会逐一注明。

13.1 设计动机：为什么需要 Streamable HTTP

要理解 Streamable HTTP 的设计选择，我们需要先看清它面临的约束条件。

第一个约束是 HTTP 的半双工特性。 与 WebSocket 这样的全双工协议不同，HTTP 本质上是请求-响应模式——客户端发起请求，服务端返回响应。在一次 HTTP 事务中，通信方向是固定的。这意味着服务端无法在没有客户端请求的情况下主动推送消息。

第二个约束是基础设施兼容性。 现实中的网络基础设施——代理服务器、负载均衡器、CDN、WAF——都是为 HTTP 设计的。WebSocket 虽然是全双工的，但并非所有中间件都能正确处理它。Streamable HTTP 选择在标准 HTTP 语义之上构建，最大程度地利用现有基础设施。

第三个约束是流式传输的需求。 MCP 的许多操作天然是流式的——工具调用可能需要数分钟才能完成，期间需要向客户端推送进度更新。普通的 HTTP 请求-响应无法满足这一需求。

Streamable HTTP 的核心设计思路可以用一句话概括：用 POST 请求承载客户端到服务端的消息，用 SSE（Server-Sent Events）流承载服务端到客户端的流式响应。 这不是两个独立的连接，而是将 SSE 嵌入到 POST 请求的响应体中。

flowchart TB
    subgraph "旧方案：SSE + HTTP POST"
        C1[客户端] -->|"1. GET /sse（长连接）"| S1[服务端]
        S1 -->|"2. 返回 endpoint URL"| C1
        C1 -->|"3. POST /message"| S1
        S1 -->|"4. 通过 SSE 通道返回响应"| C1
    end
    subgraph "新方案：Streamable HTTP"
        C2[客户端] -->|"POST /mcp（JSON-RPC 消息）"| S2[服务端]
        S2 -->|"SSE 流式响应 或 JSON 响应"| C2
        C2 -->|"GET /mcp（可选，接收服务端主动消息）"| S2
    end
    style C1 fill:#f9f,stroke:#333
    style S1 fill:#bbf,stroke:#333
    style C2 fill:#f9f,stroke:#333
    style S2 fill:#bbf,stroke:#333

13.2 协议机制：三种 HTTP 方法的分工

Streamable HTTP 使用同一个 URL endpoint，通过三种 HTTP 方法实现不同的通信语义。

POST：客户端到服务端的消息通道

客户端通过 POST 请求发送 JSON-RPC 消息，Content-Type 为 application/json。2025-11-25 规范要求请求体必须是单条 JSON-RPC 请求、通知或响应——2025-03-26 版曾允许批量数组，2025-06-18 版已去掉批量。两个 SDK 在这里不一致：实测 TS 服务端仍接受数组（两个 ping 在同一条 SSE 流里各回一条结果），Python 服务端回 400 校验错误。关键的设计点在于 Accept 头——客户端必须同时声明接受 application/json 和 text/event-stream，因为服务端可能以任一格式响应。

服务端对 POST 请求有三种可能的响应方式：

1. 202 Accepted：当客户端发送的是通知（notification）或响应（response）时，服务端无需返回内容，直接返回 202。
2. 200 OK + JSON：当服务端选择非流式模式时，直接返回 JSON-RPC 响应。
3. 200 OK + SSE：当服务端选择流式模式时，返回一个 SSE 流，在流中逐步推送消息。

两个 SDK 的服务端默认都走 SSE；TS 传 enableJsonResponse: true、Python 传 json_response=True（传给 transport 的是 is_json_response_enabled）切到 JSON 模式。代价是请求执行期间的中间消息送不到客户端：实测一个会发 notifications/progress 的工具，SSE 模式下进度通知与结果依次到达，JSON 模式下只收到最后那条结果。原因是 TS 的 send() 在 JSON 模式下只收集最终响应、不写中间消息，Python 则一直读到响应为止，途中的其他消息只记一条 debug 日志就丢弃（src/mcp/server/streamable_http.py:543-550）。

来看 TypeScript 客户端中 _send 方法的核心逻辑（packages/client/src/client/streamableHttp.ts:541-685，节选并简化：省略了鉴权与错误状态码处理，以及「消息里有没有请求」的判断）：

// 设置请求头——同时声明接受 JSON 和 SSE
headers.set('content-type', 'application/json');
headers.set('accept', [...new Set(types)].join(', '));
// types 包含 'application/json' 和 'text/event-stream'

const response = await (this._fetch ?? fetch)(this._url, init);

// 从响应头中提取 session ID
const sessionId = response.headers.get('mcp-session-id');
if (sessionId) {
    this._sessionId = sessionId;
}

// 202 表示服务端已接收但无需返回内容
if (response.status === 202) {
    // 如果是 initialized 通知，开启 GET SSE 流
    if (isInitializedNotification(message)) {
        this._startOrAuthSse({ resumptionToken: undefined })
            .catch(error => this.onerror?.(error));
    }
    return;
}

// 根据响应的 Content-Type 决定处理方式
const contentType = response.headers.get('content-type');
if (contentType?.includes('text/event-stream')) {
    // 流式响应——交给 SSE 处理器
    this._handleSseStream(response.body, { onresumptiontoken }, false);
} else if (contentType?.includes('application/json')) {
    // 非流式响应——直接解析 JSON
    const data = await response.json();
    const responseMessages = Array.isArray(data)
        ? data.map(msg => JSONRPCMessageSchema.parse(msg))
        : [JSONRPCMessageSchema.parse(data)];
    for (const msg of responseMessages) {
        this.onmessage?.(msg);
    }
}

这段代码揭示了 Streamable HTTP 的一个重要设计决策：客户端不预设服务端的响应格式。同一个请求，服务端可以根据自身能力和当前场景选择 JSON 或 SSE 响应。这种灵活性使得简单的服务端可以只实现 JSON 响应，而复杂的服务端可以利用 SSE 进行流式传输。

GET：服务端主动推送通道

GET 请求用于建立一个独立的 SSE 长连接，使服务端能够在没有客户端请求的情况下主动推送消息——例如服务端发起的请求（server-initiated requests）和通知。这个通道是可选的：如果服务端返回 405 Method Not Allowed，客户端会静默忽略，不报错。（2026-07-28 版规范去掉了这个 GET 流，服务端主动的变更通知改由 subscriptions/listen 的长响应流承载。）

两个 SDK 的服务端都限制每个 session 只能有一个 GET 流，第二个 GET 回 409。来看 TS 服务端的处理逻辑（packages/server/src/server/streamableHttp.ts:433-437，省略了 onerror 回调）：

// 检查是否已有活跃的 GET 流（节选，省略 onerror 回调）
if (this._streamMapping.get(this._standaloneSseStreamId) !== undefined) {
    return this.createJsonErrorResponse(
        409, -32_000,
        'Conflict: Only one SSE stream is allowed per session'
    );
}

DELETE：会话终止

客户端通过 DELETE 请求显式终止会话。服务端可以返回 200 OK 确认终止，也可以返回 405 Method Not Allowed 表示不支持客户端主动终止会话。TS 客户端的 terminateSession() 发 DELETE 时把 405 也当作正常结果。实测服务端：TS 回 200 并关闭 transport；Python 有会话时回 200，此后再用这个会话 ID 请求一律得到 404（Not Found: Session has been terminated），无状态模式下则回 405。

sequenceDiagram
    participant C as 客户端
    participant S as 服务端
    Note over C,S: 初始化阶段
    C->>S: POST { method: "initialize", ... }
    S-->>C: SSE/JSON 响应 + mcp-session-id 头
    C->>S: POST { method: "notifications/initialized" }
    S-->>C: 202 Accepted
    Note over C,S: 可选：建立 GET 流
    C->>S: GET（Accept: text/event-stream）
    S-->>C: SSE 长连接（服务端主动推送通道）
    Note over C,S: 正常通信
    C->>S: POST { method: "tools/call", ... }
    S-->>C: SSE 流（进度更新 + 最终结果）
    Note over C,S: 会话终止
    C->>S: DELETE
    S-->>C: 200 OK

13.3 会话管理：有状态与无状态的权衡

Streamable HTTP 支持两种运行模式：有状态（stateful）和无状态（stateless）。TS 看服务端是否提供 sessionIdGenerator，Python 看 StreamableHTTPSessionManager 的 stateless 参数。

本节讲的是 2025-11-25 版的会话机制。2026-07-28 版规范已经去掉协议级会话与 Mcp-Session-Id 头（SEP-2567），需要跨调用保留状态的服务端，改为自己签发句柄、作为普通工具参数传递。

在有状态模式下，服务端在处理 initialize 请求时生成一个 session ID，通过 mcp-session-id 响应头返回给客户端。此后，客户端必须在每个请求中携带这个 session ID。来看服务端的会话验证逻辑（省略了 onerror 回调）：

// packages/server/src/server/streamableHttp.ts:847
private validateSession(req: Request): Response | undefined {
    if (this.sessionIdGenerator === undefined) {
        // 无状态模式，跳过验证
        return undefined;
    }
    if (!this._initialized) {
        return this.createJsonErrorResponse(
            400, -32_000, 'Bad Request: Server not initialized'
        );
    }
    const sessionId = req.headers.get('mcp-session-id');
    if (!sessionId) {
        // 缺少 session ID，返回 400
        return this.createJsonErrorResponse(
            400, -32_000, 'Bad Request: Mcp-Session-Id header is required'
        );
    }
    if (sessionId !== this.sessionId) {
        // session ID 不匹配，返回 404
        return this.createJsonErrorResponse(404, -32_001, 'Session not found');
    }
    return undefined;
}

这里有一个重要的设计细节：无效的 session ID 返回 404 Not Found，而非 403 Forbidden。这是规范的要求：服务端终止会话后，对带这个 ID 的请求 MUST 回 404；客户端收到 404 就 MUST 不带会话 ID、重新发送 InitializeRequest，而不是去修复认证。不过两个 SDK 的客户端 transport 都不会自动重新初始化：TS 抛出 SdkError，Python 把它转成一条 -32600「Session terminated」错误交给上层。服务端错误体里的 code 也不统一：实测未知会话 TS 回 -32001，Python 回 -32600。

Python 实现中（src/mcp/server/streamable_http.py:64、:157-158，节选），session ID 还有格式验证的约束（规范要求只含 0x21–0x7E 的可见 ASCII 字符）：

# 只允许可见 ASCII 字符（0x21-0x7E）
SESSION_ID_PATTERN = re.compile(r"^[\x21-\x7E]+$")

def __init__(self, mcp_session_id: str | None, ...):
    if mcp_session_id is not None and not SESSION_ID_PATTERN.fullmatch(mcp_session_id):
        raise ValueError(
            "Session ID must only contain visible ASCII characters (0x21-0x7E)"
        )

无状态模式适用于简单场景或 serverless 部署——每个请求独立处理，不需要在服务端维护状态。有状态模式则适用于需要上下文延续的场景，例如多轮对话中的工具调用。

Origin 校验与 DNS rebinding。 2025-11-25 规范在 Streamable HTTP 一节开头就有一段安全警告：服务端 MUST 校验所有连接的 Origin 头，Origin 存在但不合法时 MUST 回 403（这条 403 是 2025-11-25 版才写明的，PR #1439）；在本地运行时 SHOULD 只绑定 127.0.0.1，而不是 0.0.0.0；SHOULD 对所有连接做认证。否则，恶意网页可以借 DNS rebinding 访问本机上的 MCP 服务。两个 SDK 在这件事上的落点不同：

• Python 的 transport 内置了 TransportSecurityMiddleware（src/mcp/server/transport_security.py），校验 Host（不符回 421）和 Origin（不符回 403）。streamable_http_app() 在 host 为 127.0.0.1 / localhost / ::1、又没传 transport_security 时自动开启这项保护。实测带 Origin: http://evil.example 的请求得到 403 Invalid Origin header，带 Host: evil.example:80 的得到 421。
• TS transport 上的 allowedHosts / allowedOrigins / enableDnsRebindingProtection 默认关闭，并且已经标了 @deprecated（packages/server/src/server/streamableHttp.ts:116-135），官方建议改用外部中间件。例如 createMcpExpressApp() 绑定 localhost 时会自动挂上 Host 头校验（packages/middleware/express/src/express.ts:62-86）。但 packages/middleware 下的中间件只校验 Host、不校验 Origin，规范要求的 Origin 检查得自己补上。

13.4 SSE 流处理：priming event 与断线重连

SSE（Server-Sent Events）是 Streamable HTTP 中实现流式传输的核心机制。与 WebSocket 不同，SSE 是单向的（服务端到客户端），但正好契合 MCP 的需求——在一个 POST 请求的响应中，服务端需要向客户端推送多条消息。

Priming Event：建立可恢复性的基石

2025-11-25 规范规定，服务端开启 SSE 流后 SHOULD 立即发一个只有事件 ID、data 为空的事件，让客户端尽早拿到可以当 Last-Event-ID 用的 ID。两个 SDK 只在配置了 EventStore（支持可恢复性）、且客户端协议版本不低于 2025-11-25 时才发这个 "priming event"，它不携带任何 JSON-RPC 消息。

// packages/server/src/server/streamableHttp.ts:375-399（注释有改写）
private async writePrimingEvent(
    controller: ReadableStreamDefaultController<Uint8Array>,
    encoder: InstanceType<typeof TextEncoder>,
    streamId: string,
    protocolVersion: string
): Promise<void> {
    if (!this._eventStore) return;

    // 旧版客户端不支持空数据的 SSE 事件，会尝试将空字符串解析为 JSON 而崩溃
    if (protocolVersion < '2025-11-25') return;

    const primingEventId = await this._eventStore.storeEvent(
        streamId, {} as JSONRPCMessage
    );
    let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
    if (this._retryInterval !== undefined) {
        primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
    }
    controller.enqueue(encoder.encode(primingEvent));
}

注意版本检查 protocolVersion < '2025-11-25' 这行代码——这是一个向后兼容的防御措施。早期版本的客户端会将每个 SSE 事件的 data 字段当作 JSON 解析，空数据会导致解析失败；「先发空 data 事件、服务端可随时断开让客户端轮询」这套做法是 2025-11-25 版（SEP-1699）才引入的。这里用的版本号，initialize 请求取 params.protocolVersion，其余请求取 MCP-Protocol-Version 头，缺头按 2025-03-26 算（TS :720-723、Python :520-524），所以不带版本头的请求拿不到 priming event。

实测一下（TS 服务端，配了 EventStore 与 retryInterval: 3000，curl 发 tools/call，事件 ID 由示例 EventStore 生成，UUID 部分截短）：

HTTP/1.1 200 OK
content-type: text/event-stream
mcp-session-id: sess-ba8a84a5

id: b2305ed7-…_3
retry: 3000
data: 

event: message
id: b2305ed7-…_4
data: {"method":"notifications/progress","params":{"progressToken":"p7","progress":1,"total":4},"jsonrpc":"2.0"}

第一帧就是 priming event：有 id、有 retry、data 为空。Python 服务端的帧顺序相同，只是它用请求 ID 当 stream ID（下面会看到 str(request_id)），TS 则给每个 POST 流生成一个随机 UUID。

Python 实现的细微差异

对照 Python SDK 的真实等价实现（src/mcp/server/streamable_http.py:273-285），能看出两条 SDK 在同一 feature 上的不同工程选择：

if not self._event_store:
    return
# Priming events have empty data which older clients cannot handle.
if protocol_version < "2025-11-25":
    return
priming_event_id = await self._event_store.store_event(
    str(request_id),   # ← Convert RequestId to StreamId (str)
    None,              # ← Priming event has no payload
)
priming_event: dict[str, str | int] = {"id": priming_event_id, "data": ""}
if self._retry_interval is not None:
    priming_event["retry"] = self._retry_interval
await sse_stream_writer.send(priming_event)

差异一：payload 传 None 而非空对象

• TypeScript：storeEvent(streamId, {} as JSONRPCMessage)——cast 空对象通过类型检查
• Python：store_event(stream_id, None)——类型签名 JSONRPCMessage | None 显式允许 None

Python 的 signature 更诚实——priming event 就是"没有 payload"这个语义、用 None 表达比 {} 的空对象字面量精确。TypeScript 用 as JSONRPCMessage cast，是因为它的 EventStore.storeEvent(streamId, message: JSONRPCMessage) 签名不接受空值，只能拿空对象强转。两种实现都能工作，但语义清晰度 Python 占优。

差异二：str(request_id) 的显式类型转换

priming_event_id = await self._event_store.store_event(
    str(request_id),   # ← 注释明确标了 Convert RequestId to StreamId
    None,
)

JSON-RPC 规范里 id 可以是 string 或 number（甚至 null、但 MCP 不用），但 StreamId 是固定的 str 类型——源码注释原话："Convert RequestId to StreamId (str)"。在这条调用链上，其实早就转过了：_handle_post_request 一拿到请求就执行 request_id = str(message.id)（:527），message_router 给响应找流时用的也是 str(message.id)（:994）。这里的 str() 主要是让参数类型 RequestId（str | int）对上 StreamId。

真正要紧的是存和取两端用同一种键：假如一处用 int 42、另一处用 str "42"，按 dict 查找的 EventStore 会把它们当成两个流，重放时就找不到事件。Python SDK 的做法是在入口处统一转成 str，自己实现 EventStore 时也应沿用这个约定。

差异三：retry_interval 的可选注入

priming_event: dict[str, str | int] = {"id": priming_event_id, "data": ""}
if self._retry_interval is not None:
    priming_event["retry"] = self._retry_interval

retry 字段在 SSE 规范里让服务端主动告诉客户端断线后等多久再重连，2025-11-25 版 MCP 规范还要求客户端 MUST 遵守它。两个 SDK 都提供了这个参数（TS retryInterval、Python retry_interval），不配就不在 priming event 里发、客户端走自己默认的重连策略。这个默认值不由规范规定——HTML 规范只说 EventSource 的初始重连时间由实现决定，"probably in the region of a few seconds"。这是服务端控制重连节奏的通道——如果服务端知道自己的错误恢复通常要 30 秒，就可以建议客户端不要几秒就打回来，避免重连雪崩。

这种传输层级别的重连控制比纯应用层重试更精细——是 MCP 生产部署的高阶 tuning 点。普通 MCP server 不设 retry_interval，生产压力下需要防雪崩的场景才调。

断线重连机制

网络连接不可能永远稳定。Streamable HTTP 提供了一套完整的断线重连机制，包括指数退避和服务端控制的重试间隔。

客户端的重连核心逻辑体现在 _handleSseStream 和 _scheduleReconnection 方法中：

// packages/client/src/client/streamableHttp.ts:366-476（节选并简化：省略了 event 类型判断、
// 逐条消息的 try/catch、replayMessageId 改写，以及 catch 分支里与上面相同的重连判断）
private _handleSseStream(
    stream: ReadableStream<Uint8Array> | null,
    options: StartSSEOptions,
    isReconnectable: boolean
): void {
    let lastEventId: string | undefined;
    let hasPrimingEvent = false;  // 是否收到过带 ID 的事件
    let receivedResponse = false; // 是否已收到最终响应

    const processStream = async () => {
        try {
            const reader = stream
                .pipeThrough(new TextDecoderStream())
                .pipeThrough(new EventSourceParserStream({
                    onRetry: (retryMs: number) => {
                        // 捕获服务端指定的重试间隔
                        this._serverRetryMs = retryMs;
                    }
                }))
                .getReader();

            while (true) {
                const { value: event, done } = await reader.read();
                if (done) break;

                if (event.id) {
                    lastEventId = event.id;
                    hasPrimingEvent = true;
                    onresumptiontoken?.(event.id);
                }

                // 跳过空数据事件（priming event、keep-alive）
                if (!event.data) continue;

                // 解析并分发 JSON-RPC 消息
                const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
                if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
                    receivedResponse = true;
                }
                this.onmessage?.(message);
            }

            // 流正常关闭后，判断是否需要重连
            const canResume = isReconnectable || hasPrimingEvent;
            const needsReconnect = canResume && !receivedResponse;
            if (needsReconnect && !this._abortController?.signal.aborted) {
                this._scheduleReconnection({ resumptionToken: lastEventId, ... }, 0);
            }
        } catch (error) {
            // 网络断开等异常，同样尝试重连
            // ...
        }
    };
    processStream();
}

这段代码中有两个关键的布尔变量：

• hasPrimingEvent：标记是否收到过带 event ID 的事件。只有收到过 priming event，POST 请求的 SSE 流才被视为可恢复的。
• receivedResponse：标记是否已收到最终的 JSON-RPC 响应。如果已经收到了完整响应，即使流断开也不需要重连——请求已经完成了。

重连的退避策略由 _getNextReconnectionDelay 计算：

private _getNextReconnectionDelay(attempt: number): number {
    // 优先使用服务端指定的重试间隔
    if (this._serverRetryMs !== undefined) {
        return this._serverRetryMs;
    }
    // 否则使用指数退避
    const initialDelay = this._reconnectionOptions.initialReconnectionDelay; // 默认 1s
    const growFactor = this._reconnectionOptions.reconnectionDelayGrowFactor; // 默认 1.5
    const maxDelay = this._reconnectionOptions.maxReconnectionDelay; // 默认 30s
    return Math.min(initialDelay * Math.pow(growFactor, attempt), maxDelay);
}

默认配置是初始 1s、倍率 1.5、上限 30s、maxRetries 为 2（:21-26），所以默认只会等 1s、1.5s 各重试一次，后面的 2.25s 等档位要调大 maxRetries 才会出现；而且只有重连本身失败才累加计数，每次流重新断开都从第 0 次算起。服务端可以通过 SSE 的 retry 字段覆盖这个间隔，从而实现服务端对客户端轮询频率的控制。Python 客户端更简单：没有指数退避，有 retry 就等 retry 毫秒，没有就固定等 1 秒，同样最多 2 次（src/mcp/client/streamable_http.py:48-50）。

sequenceDiagram
    participant C as 客户端
    participant S as 服务端
    C->>S: POST { method: "tools/call", id: 1 }
    S-->>C: SSE: id=evt-001, data="" （priming event, retry: 5000）
    S-->>C: SSE: id=evt-002, data=进度通知
    Note over C,S: 网络中断
    Note over C: 等待 5 秒后重连
    C->>S: GET（Last-Event-ID: evt-002）
    S-->>C: SSE: 重放 evt-002 之后的事件
    S-->>C: SSE: id=evt-003, data=最终结果
    Note over C: 收到响应，不再重连

13.5 EventStore：事件持久化与重放

断线重连的另一半在服务端——当客户端带着 Last-Event-ID 头重新连接时，服务端需要能够重放在断线期间产生的事件。这就是 EventStore 的作用。（2026-07-28 版规范去掉了这套续传机制——Last-Event-ID 与 SSE 事件 ID 都不再使用，响应流一断，进行中的请求就丢了，客户端 MUST 用新的请求 ID 重发，见该版 changelog。）

EventStore 是一个接口，定义了事件存储和重放的契约：

// TypeScript 接口（packages/server/src/server/streamableHttp.ts:27-54，省略注释）
export 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>;
}

# Python 抽象类（src/mcp/server/streamable_http.py:82-113，省略 docstring）
class EventStore(ABC):
    @abstractmethod
    async def store_event(
        self, stream_id: StreamId, message: JSONRPCMessage | None
    ) -> EventId: ...

    @abstractmethod
    async def replay_events_after(
        self, last_event_id: EventId, send_callback: EventCallback
    ) -> StreamId | None: ...

设计上有几个值得注意的点：

第一，EventStore 是可选的。 不配置 EventStore 的服务端不支持断线恢复：SSE 流断开后，这条流上还没送达的消息就拿不回来了（会话本身仍然有效，不需要重新初始化）。这降低了简单部署场景的实现复杂度。Python 的无状态模式固定不带 EventStore（streamable_http_manager.py:157-162）。

第二，getStreamIdForEventId 方法是可选的。 TypeScript 中用 ? 标记；Python 的 EventStore 抽象类里根本没有这个方法。如果提供了这个方法，服务端可以在重放前检查 event ID 的合法性并进行冲突检测；如果没有提供，服务端依赖 replayEventsAfter 返回的 stream ID 进行映射。

第三，事件存储与流传输解耦。 在 Python 实现的 message_router 中，无论客户端是否在线，消息都会被存入 EventStore：

# src/mcp/server/streamable_http.py:1006-1028 节选（日志文字有简化）
request_stream_id = target_request_id if target_request_id is not None else GET_STREAM_KEY

# 无论客户端是否连接，都存储事件
event_id = None
if self._event_store:
    event_id = await self._event_store.store_event(request_stream_id, message)

if request_stream_id in self._request_streams:
    await self._request_streams[request_stream_id][0].send(
        EventMessage(message, event_id)
    )
else:
    # 客户端未连接，但事件已存储，重连时可重放
    logger.debug("Request stream not found, client might reconnect and replay.")

这种设计意味着即使客户端在服务端处理请求的过程中完全断开，所有中间事件都不会丢失。客户端重连后，通过 Last-Event-ID 头可以获取断线期间的所有事件。实测：Python 服务端配上 EventStore，curl 发一个耗时 4 秒、每秒报一次进度的 tools/call，1.5 秒时主动断开，再用最后一个事件 ID 发 GET：

GET /mcp
Accept: text/event-stream
Mcp-Session-Id: 487b1661011445eeb783ca62abb4c7e3
Last-Event-ID: 7_4

HTTP/1.1 200 OK
content-type: text/event-stream

id: 7_5
event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"p7","progress":2.0,"total":4.0}}

id: 7_6 …（progress 3）
id: 7_7 …（progress 4）

id: 7_8
event: message
data: {"jsonrpc":"2.0","id":7,"result":{"content":[{"type":"text","text":"done 4"}],"structuredContent":{"result":"done 4"},"isError":false}}

id: 7_9
data: 
retry: 3000

断线期间产生的三条进度和最终结果都补回来了（7_ 是请求 ID 7 作 stream ID，后缀是示例 EventStore 的计数）。末尾那个空事件，是 Python 在重放流上又发的一个 priming event（:912）。

TS 服务端不是这样。 它的 send() 只在 POST 流的映射还在时才调用 storeEvent（packages/server/src/server/streamableHttp.ts:986-997）。客户端一断开，ReadableStream 的 cancel 就把映射删掉，此后产生的消息既不存也不发；最终响应到来时找不到流，直接抛错（:1006-1008）。同样的实验换成 TS 服务端（HTTP 适配层把客户端断开传递为流取消），GET 重连得到 200 和空的 SSE 流，进度 2–4 与最终结果都没有回来。只有在断开前就已写入的事件，以及 GET 独立流上的消息（:959-971 不管连没连上都先存），才能靠 Last-Event-ID 找回。所以在 TS 上用 closeSSEStream() 做轮询时，要让客户端在下一条消息产生之前重连回来。

服务端处理重放请求的流程如下（TypeScript 实现，:485-565，节选并简化）：

private async replayEvents(lastEventId: string): Promise<Response> {
    // 1. 如果提供了 getStreamIdForEventId，先验证 event ID 合法性
    if (this._eventStore.getStreamIdForEventId) {
        const streamId = await this._eventStore.getStreamIdForEventId(lastEventId);
        if (!streamId) {
            return this.createJsonErrorResponse(400, -32_000, 'Invalid event ID format');
        }
        // 检查该 stream 是否已有活跃连接
        if (this._streamMapping.get(streamId) !== undefined) {
            return this.createJsonErrorResponse(409, -32_000,
                'Conflict: Stream already has an active connection');
        }
    }

    // 2. 创建新的 SSE 流
    const readable = new ReadableStream<Uint8Array>({ ... });

    // 3. 调用 EventStore 重放历史事件
    const replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, {
        send: async (eventId, message) => {
            this.writeSSEEvent(streamController, encoder, message, eventId);
        }
    });

    // 4. 注册新的 stream mapping，继续接收后续事件
    this._streamMapping.set(replayedStreamId, { controller, encoder, cleanup: ... });

    return new Response(readable, { headers });
}

13.6 认证与 401 重试

远程传输必然涉及认证。Streamable HTTP 在传输层集成了 OAuth 认证流程，其核心设计是透明的 401 重试机制——当请求收到 401 Unauthorized 响应、且 authProvider 提供了 onUnauthorized 时，传输层会自动尝试重新认证并重发请求，对上层协议透明；没有 onUnauthorized（例如需要用户跳转浏览器的交互式 OAuth）时，则抛出 UnauthorizedError，由应用完成授权后调用 finishAuth() 再重连。

客户端的 _send 方法通过 isAuthRetry 参数实现了"最多重试一次"的语义：

// packages/client/src/client/streamableHttp.ts:525-589 节选
private async _send(
    message: JSONRPCMessage | JSONRPCMessage[],
    options: ... | undefined,
    isAuthRetry: boolean  // 是否为认证重试
): Promise<void> {
    // ...
    if (response.status === 401 && this._authProvider) {
        if (response.headers.has('www-authenticate')) {
            const { resourceMetadataUrl, scope } = extractWWWAuthenticateParams(response);
            this._resourceMetadataUrl = resourceMetadataUrl;
            this._scope = scope;
        }
        if (this._authProvider.onUnauthorized && !isAuthRetry) {
            // 第一次 401：调用 onUnauthorized 刷新凭证，然后重试
            await this._authProvider.onUnauthorized({ response, serverUrl: this._url, ... });
            return this._send(message, options, true); // 标记为重试
        }
        if (isAuthRetry) {
            // 重试后仍然 401，抛出错误
            throw new SdkError(SdkErrorCode.ClientHttpAuthentication,
                'Server returned 401 after re-authentication', { status: 401 });
        }
        throw new UnauthorizedError();
    }
    // ...
}

同样的机制也应用在 GET SSE 流的建立过程中（_startOrAuthSse 方法），确保了 GET 流也能在认证过期时自动恢复。

此外还有一个 403 Forbidden 的处理（仅当传入的是 OAuthClientProvider 时）——当服务端返回 insufficient_scope 错误时，客户端会提取新的 scope 信息，尝试用更高权限重新认证。为防止无限循环，代码通过 _lastUpscopingHeader 追踪上一次尝试的 WWW-Authenticate 头，如果两次相同就终止重试：

if (this._lastUpscopingHeader === wwwAuthHeader) {
    throw new SdkError(SdkErrorCode.ClientHttpForbidden,
        'Server returned 403 after trying upscoping', { status: 403 });
}

13.7 服务端架构：流映射与消息路由

服务端的实现需要解决一个核心的架构问题：如何将多个并发请求的响应路由到正确的 SSE 流？

TypeScript 服务端使用两层映射来解决这个问题：

1. _requestToStreamMapping: Map<RequestId, string>——将每个请求 ID 映射到一个 stream ID。
2. _streamMapping: Map<string, StreamMapping>——将 stream ID 映射到实际的 SSE 流控制器。

当服务端需要发送消息时，send 方法的逻辑如下：

// packages/server/src/server/streamableHttp.ts:943（节选并简化：省略了 eventStore 存储——即 eventId 的来源、
// 独立流拒发响应的检查、JSON 模式分支，以及找不到流时抛出的异常）
async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void> {
    let requestId = options?.relatedRequestId;
    if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
        requestId = message.id;
    }

    // 没有关联的请求 ID——发送到 GET 独立流
    if (requestId === undefined) {
        const standaloneSse = this._streamMapping.get(this._standaloneSseStreamId);
        if (standaloneSse?.controller && standaloneSse?.encoder) {
            this.writeSSEEvent(standaloneSse.controller, standaloneSse.encoder, message, eventId);
        }
        return;
    }

    // 有关联的请求 ID——发送到对应的 POST 响应流
    const streamId = this._requestToStreamMapping.get(requestId);
    const stream = this._streamMapping.get(streamId);

    if (stream?.controller && stream?.encoder) {
        this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
    }

    // 如果是最终响应（result 或 error），检查是否所有请求都已完成
    if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
        this._requestResponseMap.set(requestId, message);
        const relatedIds = [...this._requestToStreamMapping.entries()]
            .filter(([_, sid]) => sid === streamId)
            .map(([id]) => id);
        const allResponsesReady = relatedIds.every(id => this._requestResponseMap.has(id));
        if (allResponsesReady) {
            stream.cleanup(); // 关闭 SSE 流
        }
    }
}

Python 实现采用了不同的架构——用 anyio.create_memory_object_stream() 开出一对内存流（返回的是 MemoryObjectSendStream / MemoryObjectReceiveStream，anyio 里没有 MemoryObjectStream 这个类型）当内部消息通道，再由一个专门的 message_router 协程负责分发（src/mcp/server/streamable_http.py:984 定义 message_router、:1038 用 tg.start_soon 拉起；每个请求的内存流在 _handle_post_request 里建立，:529）：

flowchart LR
    subgraph "服务端内部架构"
        WS[write_stream<br/>上层协议写入] --> MR[message_router<br/>消息路由器]
        MR -->|"request_id=1"| RS1[request_stream_1<br/>POST 响应流]
        MR -->|"request_id=2"| RS2[request_stream_2<br/>POST 响应流]
        MR -->|"无 request_id"| GS[GET_stream<br/>独立 SSE 流]
        MR -->|"配置了才存"| ES[EventStore<br/>事件持久化]
    end
    RS1 --> SSE1[SSE 响应 1]
    RS2 --> SSE2[SSE 响应 2]
    GS --> SSE3[SSE 长连接]
    style MR fill:#ff9,stroke:#333
    style ES fill:#9f9,stroke:#333

这里的 message_router 是一个长期运行的协程，它从 write_stream 读取上层协议产生的消息，根据消息类型和关联的请求 ID 分发到对应的请求流。这种设计将"消息从哪里来"和"消息到哪里去"彻底解耦。

13.8 协议版本协商

MCP-Protocol-Version 头不是 Streamable HTTP 一开始就有的：2025-06-18 版规范起才要求 HTTP 客户端在后续请求上带协商好的版本（PR #548）。在初始化请求中，版本信息通过 JSON-RPC 消息的 params.protocolVersion 字段传递；在后续请求中，版本通过这个 HTTP 头传递。

TS 服务端的版本验证逻辑如下（packages/server/src/server/streamableHttp.ts:889-898，省略 onerror；真实的错误信息还会附上支持的版本列表）：

// 节选：省略 onerror 回调，错误信息简化
private validateProtocolVersion(req: Request): Response | undefined {
    const protocolVersion = req.headers.get('mcp-protocol-version');
    if (protocolVersion !== null
        && !this._supportedProtocolVersions.includes(protocolVersion)) {
        return this.createJsonErrorResponse(
            400, -32_000,
            `Bad Request: Unsupported protocol version: ${protocolVersion}`
        );
    }
    return undefined;
}

注意这里的逻辑：如果没有提供版本头，默认接受；只有提供了但不在支持列表中时才拒绝。这是一种渐进增强的设计——老客户端不发送版本头也能工作。Python（src/mcp/server/streamable_http.py:848-868）逻辑相同：缺头时按 2025-03-26 处理，不在 SUPPORTED_PROTOCOL_VERSIONS 里就回 400。实测两边对 MCP-Protocol-Version: 2099-01-01 都回 400，错误信息里都列出了支持的版本；TS 的列表比 Python 多一个 2024-10-07。

这与 2025-11-25 规范 basic/transports.mdx「Protocol Version Header」一节一致：客户端 MUST 在初始化之后的每个 HTTP 请求上带 MCP-Protocol-Version；服务端如果没收到这个头、又没有别的途径判断版本（例如初始化时协商的结果），SHOULD 假定协议版本为 2025-03-26；如果收到的是无效或不受支持的版本，则 MUST 返回 400 Bad Request。

13.9 与其他传输方式的对比

理解 Streamable HTTP 的定位，需要将它与其他传输方式进行对比。

特性	stdio	旧版 SSE+POST	Streamable HTTP	WebSocket
通信方向	双向	双向（两个连接）	双向（POST+GET）	全双工
连接数	进程管道	2 个 HTTP 连接	1-2 个 HTTP 连接	1 个连接
流式传输	天然支持	支持	支持	支持
断线恢复	不支持	不支持	支持（EventStore）	需自行实现
基础设施兼容	不适用	良好	优秀	中等
无状态部署	不适用	困难	支持	困难
实现复杂度	低	中	高	中

Streamable HTTP vs 旧版 SSE+POST： 旧方案需要客户端先通过 GET 建立 SSE 连接，服务端返回一个 endpoint URL，客户端再用这个 URL 发送 POST 请求。两个连接通过 URL 关联。Streamable HTTP 将 SSE 嵌入 POST 响应，消除了这种耦合。更重要的是，Streamable HTTP 支持服务端选择 JSON 或 SSE 响应，使得简单场景无需 SSE 即可工作。

Streamable HTTP vs WebSocket： WebSocket 的优势在于全双工——任何一方可以随时发送消息。但 WebSocket 需要专门的基础设施支持，升级成长连接后，也难以沿用按请求工作的 HTTP 缓存、网关鉴权、负载均衡等成熟机制；而且 2025-11-25 规范只定义了 stdio 和 Streamable HTTP 两种标准传输，WebSocket 属于自定义传输（见第 14 章）。Streamable HTTP 虽然不是真正的全双工，但通过 GET 流和 POST 响应流的组合，在 HTTP 语义之上实现了等效的双向通信能力。

何时选择 Streamable HTTP： 当 MCP 服务需要跨网络访问时，Streamable HTTP 是首选。它兼容几乎所有的 HTTP 基础设施，支持无状态部署（适合 serverless），并且通过 EventStore 提供了可选的断线恢复能力。只有在局域网内对延迟极其敏感、且基础设施完全可控的场景下，WebSocket 才值得考虑。

13.9.1 两个 SDK 的 Streamable HTTP 实现规模对比

把 TS 和 Python 两个官方 SDK 的 Streamable HTTP 实现文件按行数实测（wc -l 物理行，含空行与注释；不含 *.examples.ts 与测试，与首页的实现代码口径一致）——

TS SDK（mcp-typescript-sdk）——

路径	行	角色
packages/server/src/server/streamableHttp.ts	1038	WebStandardStreamableHTTPServerTransport + EventStore interface（line 27）+ stream mapping；文件顶层只导出 5 个 interface/type 和这 1 个类，其余逻辑全在这个大类里
packages/client/src/client/streamableHttp.ts	761	StreamableHTTPClientTransport——POST 发送 + GET 流接收 + reconnect 状态机
packages/middleware/node/src/streamableHttp.ts	204	Node.js IncomingMessage / ServerResponse 适配壳
总计	2003	—

Python SDK（mcp-python-sdk）——

路径	行	角色
src/mcp/server/streamable_http.py	1056	StreamableHTTPServerTransport + EventStore(ABC)（line 82）+ priming event + replay 逻辑
src/mcp/server/streamable_http_manager.py	296	Session 生命周期管理——按 mcp-session-id 把请求路由到各自的 transport，可选的空闲超时清理（session_idle_timeout，默认关闭）；TS SDK 没有对应物，见下文
src/mcp/client/streamable_http.py	588	客户端 transport
总计	1940	—

两条值得记住的工程对比——

1. 服务端 transport 主文件 1038 vs 1056——几乎一模一样——TS（1038）和 Python（1056）的差距只有 18 行——和 §15.9.4 OAuth 服务端那种"一边有一边没有"的不对称完全相反——Streamable HTTP 在两个 SDK 里都是一等公民。规范对 POST/GET/DELETE 的语义写得很具体，两边照着实现，服务端主文件体量也就相近
2. Python 有独立的 296 行会话管理器，TS 没有——TS 的 transport 一个实例只对应一个会话：validateSession 只拿请求头和自己的 this.sessionId 比（见 §13.3），多会话之间的路由（Map<sessionId, transport>、在 onsessioninitialized 里登记、在关闭时删除）留给应用自己写，官方示例 examples/server/src/simpleStreamableHttp.ts 就是这么做的。Python 则由 StreamableHTTPSessionManager 替应用做了这件事：按请求头查 _server_instances，无会话 ID 的请求新建 transport（ID 为 uuid4().hex），未知 ID 回 404

EventStore 抽象在两边都是 user-implementable interface——TS export interface EventStore（streamableHttp.ts:27）+ Python class EventStore(ABC)（streamable_http.py:82）——但要注意：两个 SDK 的库代码里都没有内置任何实现。InMemoryEventStore 只存在于示例目录（TS examples/server/src/inMemoryEventStore.ts、Python examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/event_store.py 与 examples/servers/sse-polling-demo/mcp_sse_polling_demo/event_store.py），不是可 import 的公共 API。也就是说，想要断线恢复，Redis / Postgres / 甚至内存版都得自己写或从示例里抄。这符合 §13.5 讨论的"持久化是可选的、最简实现仍然能跑"渐进增强原则。

客户端 TS 761 vs Python 588 = 1.29 倍——比 §15.9.4 的 OAuth 客户端（auth.ts 对 oauth2.py 按物理行 2.7 倍；因两边文件切分方式不同，把 Python 的 utils.py 算进去、同口径约 1.8 倍）小得多，两边在 Streamable HTTP 上比在 OAuth 上"对称"得多。

13.9.2 客户端实现里的四个状态：session、版本、SSE 与恢复点

Python 客户端的 StreamableHTTPTransport 很适合用来校准本章的协议解释。src/mcp/client/streamable_http.py:72-100 只保存三个跨请求状态：endpoint URL、session_id、protocol_version，并在 _prepare_headers() 里把它们转换为 mcp-session-id 与 mcp-protocol-version 头。streamable_http.py:102-108 还单独识别 initialize request 与 initialized notification，说明 Streamable HTTP 虽然运行在 HTTP 上，但会话状态仍由 MCP 生命周期驱动，而不是由 HTTP 连接本身驱动。

第二个状态是初始化响应里的版本。streamable_http.py:117-124 会尝试把初始化 response 解析成 InitializeResult，再把协商出的 protocol_version 存入 transport；后续请求就能通过 header 带上版本。这个实现呼应第 4 章的原则：版本协商只在 initialize 时决定，之后 HTTP 层只是携带结果，避免每个 handler 重新猜测对端协议。

第三个状态是 SSE 响应里的事件边界。streamable_http.py:129-179 的 _handle_sse_event() 明确区分空 data 的 priming event、可解析成 JSON-RPC 的 message event、以及未知 event。空 data 如果带 id，会触发 resumption callback，但不会被当成业务消息；真正的 message 才会进入 jsonrpc_message_adapter.validate_json()。这就解释了为什么 priming event 不是“多余心跳”：它可以先给客户端一个恢复点，让网络断开后有机会从某个事件 id 继续，而不是从头重跑请求。

第四个状态是断线后的 Last-Event-ID。streamable_http.py:328-372 处理 SSE response 时会持续记录 last_event_id 和服务器给出的 retry 间隔；如果流结束时还没有得到最终 response，并且手上有 last_event_id，就进入 _handle_reconnection()。streamable_http.py:374-405 的重连逻辑再把 Last-Event-ID 放入 GET 请求头。注意它不是盲目重发原始 POST：重发 POST 可能导致工具重复执行，而 GET + Last-Event-ID 表达的是“请从这个事件之后继续给我结果”。这就是 Streamable HTTP 相比旧 SSE+POST 更适合长任务的关键差异。

工程上可以据此形成一条排查顺序：如果会话串了，先看 mcp-session-id 是否被保存和回传；如果版本不一致，看 initialize response 是否成功解析；如果流式响应丢消息，看 SSE event id 与 resumption callback；如果重连后重复执行工具，说明实现把恢复误做成了重试。四个状态分清楚，Streamable HTTP 的复杂度就从“HTTP、SSE、JSON-RPC 混在一起”降成了可以逐个排查的四个状态。

还有一个工程边界值得单独强调：POST 响应可以是 JSON，也可以是 SSE。JSON 响应适合短请求，服务端一次性返回 result；SSE 响应适合长请求，服务端可以先给恢复点，再持续推送 progress、log 或最终 response。客户端不能根据“我是通过 POST 发出的请求”就假设一定收到 JSON；它必须看 content-type 并进入对应解析器。这个分支是很多自研客户端兼容失败的来源：短任务测试全过，一到长任务就把 text/event-stream 当普通 JSON 读，最后表现成解析错误或超时。

对服务端实现者来说，EventStore 也不应被误解成必须组件。没有 EventStore 时，Streamable HTTP 仍然可以完成基本远程调用，只是断线后无法精确重放；有 EventStore 时，事件 id、Last-Event-ID 和持久化记录才构成恢复闭环。生产系统要根据工具幂等性决定是否实现它：只读查询可以接受重试，写入型工具更需要恢复而不是重放执行，否则网络抖动会变成重复提交。

所以验收 Streamable HTTP 不能只测 happy path。至少要测四类场景：初始化响应为 JSON；长任务响应为 SSE；SSE 中途断开后带 Last-Event-ID 重连；客户端带旧 protocol version 或错误 session id。只有这四类都通过，才能说明实现真的覆盖了“可流式、可恢复、可协商、可隔离”四个核心目标。

13.10 本章小结

Streamable HTTP 是 MCP 协议中最复杂的传输层实现，也是远程部署场景的核心基础设施。它的设计体现了几个重要的工程哲学：

1. 在约束中寻找最优解。 HTTP 的半双工特性是硬约束，但通过将 SSE 嵌入响应体、通过独立的 GET 流接收服务端推送，Streamable HTTP 在不破坏 HTTP 语义的前提下实现了等效的双向通信。

2. 渐进增强而非全有全无。 EventStore 是可选的，GET 流是可选的，会话管理是可选的，JSON 与 SSE 响应可以共存。一个最简单的 Streamable HTTP 服务端只需要处理 POST 请求并返回 JSON 响应、对 GET 回 405，就满足了 2025-11-25 规范的基本要求。

3. 面向故障设计。 Priming event、resumption token、指数退避重连、事件持久化与重放——这些机制共同构成了一套容错体系。配置了 EventStore 时，长时间运行的操作不会因为瞬时断线而丢失进度（TS 服务端对 POST 流的限制见 §13.5）。

4. 传输层的认证透明性。 在 authProvider 能刷新凭证时，401 自动重试机制使得上层协议不需要关心认证状态的变化。这种关注点分离让协议层可以专注于业务语义，而传输层负责处理所有的网络和认证复杂性。