MCP 协议设计与实现

第14章 SSE 与 WebSocket

作者杨艺韬 · 7,373 字

第14章 SSE 与 WebSocket

前面几章我们分析了 STDIO 和 Streamable HTTP 两种传输机制。本章聚焦于 MCP 协议中另外两种重要的传输方式：SSE（Server-Sent Events） 和 WebSocket。

SSE 曾是 MCP 在网络环境下的主力传输方案，目前已被 Streamable HTTP 取代并标记为废弃；WebSocket 则作为 Python SDK 的独有实现提供了全双工通信能力。理解它们的设计思路和当前定位，对于在实际项目中做出正确的技术选型至关重要。

本章要点

SSE 的双通道设计（GET 长连接 + POST 独立请求）及其必然的复杂性
SSE 被废弃的三个核心原因：双通道脆弱性、无法 Server→Client 请求、连接恢复困难
WebSocket 的全双工优势与 MCP 中的有限地位（仅 Python SDK）
四种传输方式的选型决策树
向后兼容策略——优先 Streamable HTTP，失败降级到 SSE
迁移路线图——服务端先行，客户端渐进，监控再下线

14.1 SSE 传输的架构设计

SSE 是一种基于 HTTP 的单向流式传输技术。浏览器（或客户端）通过一个长连接的 GET 请求接收服务端推送的事件流，而客户端向服务端发送消息则需要通过独立的 POST 请求完成。这种非对称的通信模型决定了 SSE 传输在 MCP 中的架构形态。

14.1.1 为什么早期 MCP 选 SSE

回到 2024 年 11 月 MCP 发布时的技术选项：

候选方案	优势	劣势
WebSocket	全双工，低开销	防火墙/代理兼容性差，浏览器限制
SSE	标准 HTTP，通过所有代理	单向，需要辅助 POST 通道
长轮询	最简单	开销大，延迟不稳定
HTTP/2 Server Push	高效	浏览器支持撤回，复杂
gRPC-Web	高效，类型安全	需要 proxy，学习曲线

MCP 选择 SSE 的核心理由：

原生 HTTP 兼容——通过任何 HTTP 反向代理、CDN、企业防火墙
浏览器原生支持——EventSource API，零依赖
文本协议可调试——wireshark/curl 可直接看
实现成熟——RFC 8895 标准，大量 SDK

14.1.2 双通道通信模型

SSE 传输的核心设计是将一条逻辑上的双向通信链路拆分成两条物理通道：

sequenceDiagram
    participant Client as MCP 客户端
    participant Server as MCP 服务端

    Note over Client,Server: 阶段一：建立 SSE 连接
    Client->>Server: GET /sse (建立 SSE 长连接)
    Server-->>Client: event: endpoint<br/>data: /messages?sessionId=xxx

    Note over Client,Server: 阶段二：双向通信
    Client->>Server: POST /messages?sessionId=xxx<br/>{"jsonrpc":"2.0","method":"initialize",...}
    Server-->>Client: event: message<br/>data: {"jsonrpc":"2.0","result":{...}}

    Client->>Server: POST /messages?sessionId=xxx<br/>{"jsonrpc":"2.0","method":"tools/list",...}
    Server-->>Client: event: message<br/>data: {"jsonrpc":"2.0","result":{...}}

    Note over Client,Server: 阶段三：关闭连接
    Client->>Server: 关闭 EventSource

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

连接建立阶段：客户端首先发起 GET 请求建立 SSE 长连接，服务端通过 endpoint 事件告知客户端后续 POST 请求应发送到哪个 URL
endpoint 事件的安全校验：客户端必须验证 endpoint URL 的 origin 与初始连接的 origin 一致，防止中间人攻击将消息重定向到恶意服务器
会话绑定：endpoint URL 中通常包含 sessionId 参数，将 SSE 连接与后续的 POST 请求关联到同一个会话

14.1.3 endpoint 事件的安全考量

endpoint 事件的 origin 校验是 MCP 协议中一个精细的安全设计。考虑这样的攻击场景：

恶意中间人攻击 (CSRF 变种):
1. 用户访问恶意网页
2. 恶意网页诱导用户连接合法 MCP Server
3. 恶意 Server 返回 endpoint URL 指向 attacker.com
4. 客户端把所有后续消息发送到 attacker.com
5. 敏感数据泄露

对策：SDK 强制校验 endpoint URL 的 origin 必须与初始连接一致。这是一个简单但有效的同源策略：

if (this._endpoint.origin !== this._url.origin) {
    throw new Error(`Endpoint origin does not match connection origin`);
}

这种校验必须在 SDK 层实现，不能依赖 server 端诚实——因为这个校验就是在防止服务端作恶。

14.2 TypeScript SDK 的 SSE 客户端实现

TypeScript SDK 中的 SSEClientTransport 类实现了 Transport 接口，完整展示了 SSE 传输的工作机制。

连接建立

连接建立的核心逻辑在 _startOrAuth 方法中。该方法创建 EventSource 实例并监听两类事件：

// 监听 endpoint 事件，获取 POST 请求的目标 URL
this._eventSource.addEventListener('endpoint', (event: Event) => {
    const messageEvent = event as MessageEvent;
    this._endpoint = new URL(messageEvent.data, this._url);

    // 安全校验：endpoint URL 的 origin 必须与连接 URL 一致
    if (this._endpoint.origin !== this._url.origin) {
        throw new Error(`Endpoint origin does not match connection origin`);
    }
    resolve();
});

// 监听 message 事件，接收服务端推送的 JSON-RPC 消息
this._eventSource.onmessage = (event: Event) => {
    const messageEvent = event as MessageEvent;
    const message = JSONRPCMessageSchema.parse(JSON.parse(messageEvent.data));
    this.onmessage?.(message);
};

消息发送

发送消息时，send 方法通过 POST 请求将 JSON-RPC 消息发送到之前接收到的 endpoint URL：

async _send(message: JSONRPCMessage, isAuthRetry: boolean): Promise<void> {
    if (!this._endpoint) {
        throw new SdkError(SdkErrorCode.NotConnected, 'Not connected');
    }
    const headers = await this._commonHeaders();
    headers.set('content-type', 'application/json');

    const response = await (this._fetch ?? fetch)(this._endpoint, {
        method: 'POST',
        headers,
        body: JSON.stringify(message),
        signal: this._abortController?.signal
    });

    if (response.status === 401 && this._authProvider && !isAuthRetry) {
        // 认证失败：尝试刷新 token 后重试一次
        const result = await this._authProvider.onUnauthorized(response);
        if (result === 'AUTHORIZED') {
            return this._send(message, true);  // 递归重试，但 isAuthRetry=true 防止无限循环
        }
        throw new UnauthorizedError();
    }

    if (!response.ok) {
        throw new Error(`HTTP ${response.status}: ${await response.text()}`);
    }
}

这里有一个精妙的设计：认证失败（401）时，_send 方法会调用 authProvider.onUnauthorized 尝试刷新凭证，然后以 isAuthRetry = true 递归调用自身进行重试，但只重试一次。同样的认证重试逻辑也存在于 SSE 连接建立阶段的 onerror 处理中。

断线重连的挑战

SSE 的一个痛点是断线重连。EventSource 自动重连，但重连后 session 已失效：

1. SSE 连接断开（网络抖动）
2. EventSource 自动重连 GET /sse
3. 服务端看到新连接，分配新的 sessionId
4. 但客户端还在用旧 endpoint URL！
5. POST 请求失败：sessionId invalid

SDK 必须监听 endpoint 事件的重新到达，替换 endpoint URL。但已发出的请求可能丢失——这是 SSE 架构的固有缺陷。

14.3 Python SDK 的 SSE 客户端实现

Python SDK 的 sse_client 采用了异步上下文管理器模式，使用 httpx 和 httpx_sse 库实现 SSE 连接，并通过 anyio 的内存流实现消息的异步传递。

双协程架构

其核心架构由两个并发协程构成：

sse_reader：从 SSE 事件流中读取消息，解析后写入 read_stream
post_writer：从 write_stream 中读取待发送的消息，通过 POST 请求发送到服务端

async def sse_reader(task_status):
    async for sse in event_source.aiter_sse():
        match sse.event:
            case "endpoint":
                endpoint_url = urljoin(url, sse.data)
                # 安全校验：验证 origin 一致性
                if urlparse(endpoint_url).netloc != urlparse(url).netloc:
                    raise ValueError("Endpoint origin mismatch")
                task_status.started(endpoint_url)
            case "message":
                message = types.jsonrpc_message_adapter.validate_json(sse.data)
                await read_stream_writer.send(SessionMessage(message))

async def post_writer(endpoint_url: str):
    async for session_message in write_stream_reader:
        response = await client.post(
            endpoint_url,
            json=session_message.message.model_dump(by_alias=True, mode="json")
        )
        response.raise_for_status()

task_status 机制

值得注意的是，sse_reader 使用了 anyio 的 task_status 机制：只有当 endpoint 事件到达后，post_writer 才会启动。这确保了在发送任何消息之前，endpoint URL 已经就绪。

async with anyio.create_task_group() as tg:
    # sse_reader 先启动，等待 endpoint 事件
    endpoint_url = await tg.start(sse_reader)
    # endpoint 事件到达后，post_writer 才启动
    tg.start_soon(post_writer, endpoint_url)

这避免了一个常见的竞态：如果 post_writer 和 sse_reader 并行启动，post_writer 可能在 endpoint_url 就绪前就想发送消息。

session_id 回调

Python SDK 还提供了 on_session_created 回调，允许调用者在会话建立时获取 sessionId，便于实现会话恢复等高级功能：

async def sse_client(url: str, on_session_created: Callable[[str], None] | None = None):
    # ...
    # 当 endpoint URL 确定后，从中提取 sessionId
    parsed = urlparse(endpoint_url)
    query = parse_qs(parsed.query)
    session_id = query.get("sessionId", [None])[0]
    if session_id and on_session_created:
        on_session_created(session_id)

14.4 WebSocket 传输

WebSocket 提供了真正的全双工通信能力，客户端和服务端可以在同一条连接上同时发送和接收消息。在 MCP 生态中，WebSocket 传输仅在 Python SDK 中实现，TypeScript SDK 并未提供对应支持。

14.4.1 为什么 TS SDK 没有 WebSocket

一个常见的疑问：为什么 TypeScript SDK 不实现 WebSocket？几个原因：

浏览器环境的限制——浏览器 WebSocket 不支持自定义 header，OAuth token 很难传递
Streamable HTTP 足够好——新协议已经解决了 SSE 的问题，没必要再加一种传输
节能减员——四种传输协议要维护，成本高
Python SDK 场景更需要——Python 多用于服务端/脚本，全双工更有价值

这也解释了为什么 WebSocket 在 MCP 中是”实验性”的。

14.4.2 全双工模型

与 SSE 的双通道设计不同，WebSocket 的架构更加简洁：

sequenceDiagram
    participant Client as MCP 客户端
    participant Server as MCP 服务端

    Note over Client,Server: WebSocket 握手
    Client->>Server: GET /ws (Upgrade: websocket)<br/>Sec-WebSocket-Protocol: mcp

    Server-->>Client: 101 Switching Protocols<br/>Sec-WebSocket-Protocol: mcp

    Note over Client,Server: 全双工通信（同一连接）
    Client->>Server: {"jsonrpc":"2.0","method":"initialize",...}
    Server-->>Client: {"jsonrpc":"2.0","result":{...}}

    par 并行消息
        Server-->>Client: {"jsonrpc":"2.0","method":"notifications/progress",...}
    and
        Client->>Server: {"jsonrpc":"2.0","method":"tools/call",...}
    end

    Server-->>Client: {"jsonrpc":"2.0","result":{...}}

    Note over Client,Server: 关闭连接
    Client->>Server: WebSocket Close Frame
    Server-->>Client: WebSocket Close Ack

WebSocket 方案有两个显著优势：

无需 endpoint 协商——连接建立后即可直接通信
真正的全双工——双方可以在任意时刻发送消息，无需等待对方响应

14.4.3 Python SDK 的 WebSocket 实现

Python SDK 的 websocket_client 同样采用异步上下文管理器模式，代码简洁而完整：

@asynccontextmanager
async def websocket_client(url: str):
    async with ws_connect(url, subprotocols=[Subprotocol("mcp")]) as ws:
        # 创建内存流用于消息传递
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

        async def ws_reader():
            async for raw_text in ws:
                message = types.jsonrpc_message_adapter.validate_json(raw_text)
                await read_stream_writer.send(SessionMessage(message))

        async def ws_writer():
            async for session_message in write_stream_reader:
                msg_dict = session_message.message.model_dump(
                    by_alias=True, mode="json", exclude_unset=True
                )
                await ws.send(json.dumps(msg_dict))

        async with anyio.create_task_group() as tg:
            tg.start_soon(ws_reader)
            tg.start_soon(ws_writer)
            yield (read_stream, write_stream)
            tg.cancel_scope.cancel()

四个关键设计决策

这段代码展现了几个关键设计决策：

1. mcp 子协议

通过 subprotocols=[Subprotocol("mcp")] 声明使用 MCP 子协议，服务端据此识别这是 MCP 通信而非普通 WebSocket 连接。这是 RFC 6455 规定的标准机制，让一个 WebSocket 端点可以承载多种协议。

2. 零缓冲内存流

anyio.create_memory_object_stream(0) 创建的是无缓冲流，这意味着发送方会阻塞直到接收方准备好。这种背压机制防止了消息在内存中无限堆积——如果下游处理慢，上游自然就慢下来，而不是 OOM。

3. 统一的流抽象

无论是 SSE 还是 WebSocket，Python SDK 都将传输层抽象为 (read_stream, write_stream) 元组，上层的 Session 和 Client 类无需关心底层使用的是哪种传输协议。这是一个漂亮的 Transport Layer Abstraction。

4. 优雅关闭

当调用者退出 async with 块时，tg.cancel_scope.cancel() 会取消所有子任务，WebSocket 连接随之关闭。不需要手动清理资源。

与 SSE 实现对比，WebSocket 版本的代码量明显更少，不需要 endpoint 协商、不需要维护两个独立的 HTTP 通道、也不需要处理 SSE 事件类型的分发逻辑。

14.5 四种传输方式横向对比

MCP 协议目前支持四种传输方式，各有其适用场景。

14.5.1 对比表

维度	STDIO	Streamable HTTP	SSE（已废弃）	WebSocket
通信模型	进程管道	HTTP 请求/响应 + 可选 SSE 流	GET 流 + POST 请求	全双工
部署模型	本地进程	远程 HTTP 服务	远程 HTTP 服务	远程 WebSocket 服务
SDK 支持	TS + Python	TS + Python	TS + Python	仅 Python
是否需要服务端	否（子进程）	是	是	是
认证支持	无（依赖 OS）	OAuth 2.0	OAuth 2.0	无内建支持
连接开销	进程启动 ~50ms	零（HTTP 复用）	初始 ~100ms	握手 ~50ms
吞吐量	极高（IPC）	高	中	高
延迟	最低	低	中	低
状态保持	无状态	可选 session	强状态（endpoint）	强状态（连接）
防火墙友好	N/A	✅ 非常友好	✅ 友好	⚠️ 一般
代理兼容	N/A	✅ 完美	⚠️ 需要长连接支持	⚠️ 需要 Upgrade 支持
浏览器可用	❌	✅	✅	✅
调试难度	低（文本 pipe）	低（HTTP 工具）	中	中（需专用工具）
协议状态	稳定	当前推荐	已废弃	实验性
推荐场景	桌面 CLI/IDE	云端服务	遗留兼容	Python 全双工

14.5.2 选型决策树

graph TD
    Start{部署场景?}
    Start -->|本地集成| Local
    Start -->|远程服务| Remote

    Local[STDIO<br/>简单可靠]

    Remote{语言?}
    Remote -->|TypeScript| TS{新项目?}
    Remote -->|Python| PY{需要 Server→Client 高频推送?}

    TS -->|是| StreamableTS[Streamable HTTP<br/>推荐]
    TS -->|否，兼容老服务端| Compat[Streamable HTTP<br/>+ SSE 降级]

    PY -->|是| WS[WebSocket<br/>全双工]
    PY -->|否| StreamablePY[Streamable HTTP<br/>推荐]

    style Local fill:#dcfce7,stroke:#22c55e
    style StreamableTS fill:#dcfce7,stroke:#22c55e,stroke-width:2px
    style StreamablePY fill:#dcfce7,stroke:#22c55e,stroke-width:2px
    style Compat fill:#fef3c7,stroke:#f59e0b
    style WS fill:#dbeafe,stroke:#3b82f6

14.5.3 详细选型指南

场景 1：IDE 插件、CLI 工具

推荐：STDIO
理由：最简单、最可靠，无需网络配置，启动延迟低
注意：进程管理需要处理——子进程崩溃要能感知并重启

场景 2：云端 MCP 服务

推荐：Streamable HTTP
理由：当前协议推荐的网络传输方案，兼容 HTTP 基础设施
注意：注意 session affinity（分布式部署时同一 session 要路由到同一节点）

场景 3：需要兼容现存 SSE 服务端

推荐：Streamable HTTP + SSE 降级
理由：新协议优先，失败时降级到旧协议
注意：只在过渡期使用，等 SSE 端点下线后移除降级逻辑

场景 4：Python 全双工高频推送

推荐：WebSocket
理由：延迟低、代码简洁
注意：仅在 Python 场景，生态兼容性有限

场景 5：企业内网，严格防火墙

推荐：Streamable HTTP
理由：最标准的 HTTP 形态，通过所有企业代理
注意：避免 WebSocket——很多企业防火墙不允许 Upgrade

14.6 SSE 的废弃与迁移

14.6.1 为什么 SSE 被废弃

TypeScript SDK 中的 SSEClientTransport 类已被标记为 @deprecated，官方建议迁移到 StreamableHTTPClientTransport。

废弃的核心原因在于 SSE 传输的三个架构缺陷：

缺陷 1：双通道脆弱性

SSE 需要维护 SSE 连接（GET）和 POST 通道两条链路，任何一条中断都会导致通信失败。而且两条通道的失败模式不一样——SSE 失败会自动重连，POST 失败会立即返回错误。这种不对称让错误处理变得复杂。

缺陷 2：服务端不能主动请求

SSE 是单向流，服务端无法通过 SSE 通道发起 JSON-RPC 请求（注意是 request，不是 notification）。而 MCP 1.0 之后引入了 sampling/createMessage 等需要 Server→Client 请求的机制，SSE 无法支持。

这个问题只能通过在客户端建立另一个 SSE 连接（反向）来 hack，但这又把架构复杂度推高了一个数量级。

缺陷 3：连接恢复困难

SSE 连接断开后，需要重新建立连接并获取新的 endpoint URL，旧 endpoint 上的未完成请求可能丢失。无法无缝恢复——这对长对话型的 MCP 会话是致命的。

14.6.2 Streamable HTTP 如何解决这些问题

Streamable HTTP 通过以下方式解决了上述所有问题：

核心设计:
- 所有请求都走 POST /mcp（统一入口）
- 响应可以是同步 JSON 或流式 SSE
- session 通过 cookie 或 header 传递
- Server→Client 请求通过响应流传递
- 断线重连 = 重新 POST + 携带 session ID

具体改进：

SSE 缺陷	Streamable HTTP 的解法
双通道脆弱	单一 HTTP 入口
服务端不能主动请求	响应流可携带 Server→Client 请求
连接恢复困难	无状态 HTTP，session 通过 ID 恢复

14.6.3 向后兼容策略

尽管 SSE 已被废弃，但由于大量现存服务端仍在使用该传输方式，客户端需要在过渡期间支持两种协议。TypeScript SDK 官方示例提供了一个清晰的降级策略：

async function connectWithBackwardsCompatibility(
  url: string,
  clientInfo: ClientInfo,
): Promise<{ client: Client; transport: Transport; type: string }> {
    try {
        // 优先尝试 Streamable HTTP
        const transport = new StreamableHTTPClientTransport(new URL(url));
        const client = new Client(clientInfo);
        await client.connect(transport);
        return { client, transport, type: 'streamable-http' };
    } catch (error) {
        // 判断是否是"协议不支持"的错误
        if (!isProtocolMismatch(error)) {
            throw error;  // 其他错误直接抛出
        }

        // 降级到 SSE
        console.warn('Falling back to legacy SSE transport');
        const sseTransport = new SSEClientTransport(new URL(url));
        const sseClient = new Client(clientInfo);
        await sseClient.connect(sseTransport);
        return { client: sseClient, transport: sseTransport, type: 'sse' };
    }
}

function isProtocolMismatch(error: unknown): boolean {
    if (!(error instanceof Error)) return false;
    // 服务端返回 404/405 通常说明不支持 Streamable HTTP
    return error.message.includes('404') || error.message.includes('405');
}

这个模式的关键在于：先尝试现代协议，如果服务端不支持（返回 4xx 错误），则回退到旧协议。

14.6.4 迁移路线图

对于正在维护 MCP 服务端的开发者，建议按以下步骤完成迁移：

gantt
    title SSE → Streamable HTTP 迁移时间线
    dateFormat YYYY-MM-DD
    section 服务端
    在 SSE 基础上增加 Streamable HTTP 端点: 2026-01-01, 30d
    两个端点并行运行: 2026-02-01, 90d
    section 客户端
    采用降级策略更新客户端: 2026-01-15, 60d
    部署新版本到主流用户: 2026-03-15, 60d
    section 监控
    观察客户端流量分布: 2026-03-01, 120d
    确认 SSE 流量 < 5%: crit, 2026-06-30, 30d
    section 完全迁移
    标记 SSE 端点为 deprecated: 2026-07-01, 30d
    移除 SSE 端点: 2026-08-01, 30d

四步迁移法：

服务端先行：在现有 SSE 端点的基础上，增加 Streamable HTTP 端点支持。两个端点可以并行运行
客户端适配：采用上述降级策略更新客户端代码，确保优先使用新协议
监控与切换：观察客户端连接方式的统计数据，当绝大多数活跃客户端已切换到 Streamable HTTP，并且旧端点错误率、客服反馈和业务关键路径都稳定后，再考虑下线 SSE 端点
完全迁移：移除 SSE 相关代码，简化服务端架构

关键监控指标：

新旧协议的连接数比例（阈值由业务兼容窗口和客户分布决定）
各客户端版本的协议分布
降级发生的频率（高说明客户端版本旧）

14.7 实战：为你的 MCP Server 增加 Streamable HTTP 支持

以下是一个参考实现，展示如何让一个现有的 SSE-only 服务端同时支持 Streamable HTTP：

import express from 'express';
import { Server as MCPServer } from '@modelcontextprotocol/sdk/server';
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamable-http';
import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse';

const app = express();

// ========== 现有 SSE 端点（保留）==========
app.get('/sse', async (req, res) => {
    const transport = new SSEServerTransport('/messages', res);
    const server = new MCPServer({ /* ... */ });
    await server.connect(transport);
});

app.post('/messages', express.json(), async (req, res) => {
    // SSE session 查找 + 消息投递
});

// ========== 新增 Streamable HTTP 端点 ==========
app.post('/mcp', express.json(), async (req, res) => {
    const sessionId = req.headers['mcp-session-id'] as string;
    const transport = new StreamableHTTPServerTransport({ sessionId });
    const server = new MCPServer({ /* ... */ });
    await server.connect(transport);
    // Streamable HTTP 在响应流中处理所有消息
});

// ========== 服务端能力广播 ==========
app.get('/.well-known/mcp-server', (req, res) => {
    res.json({
        transports: {
            'streamable-http': { url: '/mcp' },
            'sse': { url: '/sse', deprecated: true },
        },
    });
});

app.listen(3000);

客户端可以通过 .well-known 发现服务端支持的传输协议，自动选择最优的：

async function autoConnect(baseUrl: string): Promise<Client> {
    const wellKnown = await fetch(`${baseUrl}/.well-known/mcp-server`).then(r => r.json());

    // 优先选择非 deprecated 的 transport
    const transports = Object.entries(wellKnown.transports)
        .sort(([, a], [, b]) => (a.deprecated ? 1 : 0) - (b.deprecated ? 1 : 0));

    for (const [type, config] of transports) {
        try {
            const transport = createTransport(type, baseUrl + config.url);
            const client = new Client({ name: 'my-client', version: '1.0' });
            await client.connect(transport);
            return client;
        } catch (e) {
            continue;
        }
    }

    throw new Error('No compatible transport');
}

14.8 三个反模式

反模式一：把 SSE 当成实时推送通道

现象：开发者把 MCP 的 SSE 通道当成一般 WebSocket 用，往里面塞大量 notification。

问题：SSE 是 MCP 的响应通道，不是通用推送通道。高频 notification 会让客户端消化不过来。

对策：Server→Client 推送应该克制；需要高频推送时考虑 WebSocket。

反模式二：跨 session 共用 endpoint

现象：服务端生成一个 endpoint URL，所有客户端共用。

问题：消息会串——客户端 A 的请求响应可能发给客户端 B。

对策：每个 SSE 连接分配独立的 endpoint URL，包含唯一 sessionId。

反模式三：无脑 WebSocket

现象：开发者觉得 WebSocket 看起来”高级”，不分场景都用。

问题：浏览器限制、企业防火墙兼容性、TS SDK 不支持。

对策：优先 Streamable HTTP。只在明确需要全双工 + Python 场景下选 WebSocket。

14.9 SDK 源码实况：TS SDK 已抛弃服务端 SSE 和客户端 WebSocket

章节 §14.6 讨论 SSE 废弃——实际上 TypeScript SDK 已经把”废弃”做到了源码级——比文档讲的更彻底。

实测（基于本地 mcp-typescript-sdk 仓库）：

文件	状态
`packages/client/src/client/sse.ts`	存在、311 行——客户端仍能连接旧 SSE 服务端
`packages/server/src/server/sse.ts`	不存在——TS SDK 服务端完全没有 SSE 实现
`packages/client/src/client/websocket.ts`	不存在——WebSocket 客户端已被显式移除
`packages/server/src/server/websocket.ts`	不存在

TS SDK 服务端只剩两种 transport——stdio.ts + streamableHttp.ts。

WebSocket 移除的 changeset（.changeset/remove-websocket-transport.md）原文——

Remove WebSocketClientTransport. WebSocket is not a spec-defined transport; use stdio or Streamable HTTP. The Transport interface remains exported for custom implementations. See #142.

两个值得记住的事实——

WebSocket 被官方判定为”非规范传输”——MCP 1.0 的 spec 只承认 stdio + Streamable HTTP——WebSocket 是 SDK 自由选择、TS SDK 选择不实现、Python SDK 仍保留（85 行）作为遗产
TS SDK 服务端从未有过 SSE——章节 §14.7 示例代码里 import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse' 引用的是 legacy 包 @modelcontextprotocol/sdk——已被 @modelcontextprotocol/server + @modelcontextprotocol/client 取代——生产代码不要再用 legacy 包路径

Python SDK 保留 SSE 服务端（241 行）+ WebSocket 服务端（52 行）——是因为 Python 生态用户基数小、breaking change 影响面大、所以维持向后兼容更久。

结论的强化——本章 §14.10 选型铁律应该升级为——

本地 → STDIO
云端 (TS) → 只能 Streamable HTTP（SSE/WS 都不存在）
云端 (Python) → Streamable HTTP（SSE/WS 是遗产、新项目别选）
WebSocket 全双工特殊场景 → 重新考虑设计（既然不在 spec、未来可能 Python SDK 也会移除）

14.9.1 sse 实现的尺寸对比与根因

文件	行	备注
TS `client/sse.ts`	311	自己实现 EventSource（Node 不原生支持）
Python `client/sse.py`	160	用 `httpx_sse` 库、不用自己 parse SSE 帧
Python `server/sse.py`	241	服务端比客户端复杂（要管多 session、endpoint event 等）
Python `server/websocket.py`	52	用 `websockets` 库、协议细节完全外包
Python `client/websocket.py`	85	同上

TS 客户端 SSE 比 Python 大近一倍——根因：Node.js 没有原生 EventSource（浏览器有），TS SDK 必须自己实现完整的 SSE wire format parser（按 SSE spec：event: / data: / id: / retry: 四种字段、多行 data 拼接、\n\n 边界等）——这部分代码占 sse.ts 的近 50%。Python 直接 import httpx_sse、把 parse 完全外包——160 行就够。

WebSocket 实现都很短（85 / 52 行）——因为 websockets (Python) 和 ws (Node) 这类库已经处理了所有协议细节（握手、frame、ping/pong、close）。SDK 层只需要 wrap 成 MCP 的 Transport 接口——薄薄一层。

这条规律——协议越成熟、SDK 包装层越薄。SSE 在 Node.js 还需要 SDK 自己 parse 是因为 Node 标准库没跟上（浏览器原生有 EventSource、Node 一直没加）。WebSocket 因为有成熟库、SDK 实现极简。

14.9.2 旧 SSE 与 WebSocket 的真实边界：Python 客户端最清楚

旧 SSE 的客户端实现直接暴露了它的“双通道”本质。mcp-python-sdk/src/mcp/client/sse.py:35-60 先用 GET 连接 SSE endpoint；sse.py:65-92 的 reader 等待服务端发来 endpoint 事件，并检查 endpoint 的 scheme 与 host 是否和原始 SSE URL 一致；只有通过同源检查后，writer 才能把客户端消息 POST 到这个 endpoint。也就是说，旧 SSE 并不是一条全双工连接，而是“GET 接收 + POST 发送 + endpoint 事件负责配对”的组合。

这个组合的安全边界很脆。sse.py:74-84 的同源检查不是装饰，它防止恶意或错误的 SSE 服务端把客户端引导到另一个 origin 发送 JSON-RPC 请求。对浏览器外的 CLI 客户端来说，这类问题更容易被忽略，因为没有浏览器同源策略自动兜底；SDK 必须自己检查。另一方面，sse.py:93-106 只把 message event 解析成 JSON-RPC，空消息会跳过，未知 event 只是 warning。这使旧 SSE 能兼容心跳和少量扩展，但也说明它的协议语义高度依赖 event 名称。

WebSocket 的边界完全不同。mcp-python-sdk/src/mcp/client/websocket.py:15-31 把 WebSocket 客户端定义成与服务端对称的 MCP transport；websocket.py:41-44 连接时请求 mcp subprotocol；websocket.py:46-68 用两个协程分别读写 JSON-RPC 文本消息。这里没有 endpoint 事件、没有 POST 发送路径、也没有 SSE event id；好处是模型简单，坏处是所有恢复、认证、代理兼容和负载均衡问题都要落在 WebSocket 连接层处理。

这解释了为什么本章不应把 WebSocket 描述成“更先进的 SSE”。它们解决的是不同约束：旧 SSE 借 HTTP GET/POST 穿过普通基础设施，代价是双连接关联复杂；WebSocket 提供真正全双工，代价是基础设施和恢复语义更重；Streamable HTTP 则试图保留 HTTP 生态，把流式响应、session id、protocol version、Last-Event-ID 放到同一个传输模型里。迁移时最重要的不是追求某个传输“最新”，而是确认你的服务是否需要断线恢复、是否部署在受限代理后、是否允许长连接、是否能承受重复执行风险。

从实现取舍看，Python SDK 同时保留 SSE client 与 WebSocket client，可以服务旧系统和实验场景；TS SDK 的主线则更集中在 stdio 与 Streamable HTTP。对生产团队而言，旧 SSE 最适合被当作兼容层维护，WebSocket 适合私有网络或双方都可控的专用通道，跨网络的默认选项应优先评估 Streamable HTTP。这个结论不是因为某个传输“更潮”，而是因为状态恢复、认证集成、HTTP 运维工具链和协议版本协商最终都要在生产里付账。

迁移时可以把三种传输放进同一张运维视角的表里：

维度	旧 SSE	WebSocket	Streamable HTTP
客户端发送	POST 到 endpoint	同一连接发送文本帧	POST JSON-RPC
服务端推送	SSE message event	同一连接发送文本帧	POST 响应流或 GET 流
恢复语义	依赖自定义重连	通常需要应用自建	Last-Event-ID + EventStore
代理兼容	HTTP 友好但双通道复杂	取决于代理升级支持	复用 HTTP 语义
适合角色	兼容旧客户端	双方可控的专用链路	远程默认候选

这张表的重点不是给某个传输打分，而是提醒实现者：传输选择会把复杂度推到不同地方。旧 SSE 把复杂度推给 endpoint 配对；WebSocket 把复杂度推给连接存活与恢复；Streamable HTTP 把复杂度推给 content-type、session header 和事件存储。一个团队如果没有能力长期维护这些状态，就应选择复杂度最符合自己运维工具链的方案。

还有一个经常被忽略的迁移风险：客户端库的“自动降级”必须可观测。若连接 Streamable HTTP 失败后静默退回旧 SSE，短期会让用户觉得稳定，长期会让团队误判迁移进度。更好的做法是在客户端指标中记录首选传输、实际传输、失败原因和 server 版本；服务端也记录每条连接的 transport 类型与协议版本。这样，当旧 SSE 流量迟迟不降时，团队能区分是老客户端没升级、代理挡住了 Streamable HTTP，还是新端点存在兼容问题。

WebSocket 也有类似陷阱。它在本地或内网 demo 里常显得最简单，因为一条连接双向收发；但一到企业代理、移动网络、长时间空闲连接和滚动发布，连接保持与重连策略就会成为主要复杂度。MCP 的消息层本身已经能表达 request、notification、response；传输层不需要再追求“看起来最实时”，而要选择最容易被基础设施正确承载的形态。

因此，本章的迁移建议可以压缩成一句工程原则：不要按协议形态选传输，要按故障恢复责任选传输。如果你愿意维护 endpoint 配对，旧 SSE 可以继续兼容；如果你愿意维护连接状态，WebSocket 可以服务专用场景；如果你希望把认证、代理、可观测和恢复尽量放回 HTTP 体系，Streamable HTTP 才是更自然的长期方向。

这条原则同样适用于自研客户端：先写清楚故障责任，再决定传输形态。

否则迁移只是在搬运复杂度。

14.10 本章小结

本章深入分析了 MCP 协议中 SSE 和 WebSocket 两种传输方式的设计与实现。

核心要点回顾

SSE 是 MCP 早期的主力网络传输方案——双通道设计（GET 长连接 + POST 请求）
SSE 已被废弃——三个核心缺陷让它不适合 MCP 1.0 之后的能力需求
WebSocket 提供全双工——但仅 Python SDK 实现，场景有限
Streamable HTTP 是当前推荐——统一 HTTP 模型解决了 SSE 的所有缺陷
迁移需要循序渐进——服务端先行，客户端降级，监控再下线

选型铁律

本地 → STDIO
云端 → Streamable HTTP
旧系统兼容 → Streamable HTTP + SSE 降级
Python 全双工特殊场景 → WebSocket