MCP SSE 轮询客户端实践：基于 python-sdk 的自动重连与 Last-Event-ID 断点续传

本篇文章围绕官方 python-sdk 仓库中的 examples/clients/sse-polling-client 演示客户端展开，完整讲解 SSE（Server-Sent Events）轮询模式下客户端的自动重连机制（SEP-1699）：包括如何启动配套的 mcp-sse-polling-demo 服务器、如何运行客户端并配置参数，以及底层 EventStore、close_sse_stream()、retry: 提示等核心机制如何协同工作。读完本文，你将掌握在 MCP Streamable HTTP 传输下，客户端如何在服务器主动关闭 SSE 流后自动重连、借助 Last-Event-ID 无缝恢复未接收消息的完整实战方案。

SSE 轮询模式解决了什么问题

MCP 的 Streamable HTTP 传输基于 HTTP + SSE。在常规设计中，客户端发起一次 tools/call 请求后，服务器通过 SSE 流持续把进度通知、日志消息和最终结果推送给客户端。但当一个工具调用运行时间较长（例如批量处理 100 个任务）时，长时间占住一条打开的 HTTP 连接会让服务器端的连接槽位吃紧，也给网络层带来长时间空闲连接被中间设备断开的隐患。

SSE 轮询模式（SEP-1699）的思路是：服务器在长任务执行过程中主动关闭 SSE 流，释放连接；客户端检测到流结束后立即按 retry: 提示重新发起连接，并通过 Last-Event-ID 让服务器把断开期间错过的消息全部补发。对客户端调用方而言，整个过程是透明的——await client.call_tool(...) 最终正常返回，仿佛连接从未断开过。

快速上手：先跑服务器，再跑客户端

示例以两个独立进程协作的方式演示完整链路。先启动 SSE 轮询演示服务器：

uv run mcp-sse-polling-demo --port 3000

服务器会监听 http://127.0.0.1:3000，在 /mcp 端点提供 Streamable HTTP 服务，并暴露一个名为 process_batch 的工具。随后在另一个终端启动配套客户端：

uv run mcp-sse-polling-client --url http://localhost:3000/mcp

客户端会完成如下动作（对应 main.py 中的 run_demo）：

1. 通过 streamable_http_client(url) 建立读写流；
2. 在 ClientSession 中执行 await session.initialize() 完成协议握手；
3. 调用 session.list_tools() 打印服务器暴露的工具列表；
4. 调用 process_batch 工具并打印最终结果。

整个 process_batch 调用期间，服务器会周期性关闭 SSE 流触发客户端重连，而客户端无需任何额外配置即可自动完成重连与消息恢复——这正是该演示希望传达的核心能力。

客户端命令行选项详解

mcp-sse-polling-client 使用 Click 定义命令行参数，全部选项及其含义如下表：

选项	默认值	说明
--url	http://localhost:3000/mcp	服务器 SSE 端点 URL
--items	10	要处理的条目数量（会作为 process_batch 的参数传给服务器）
--checkpoint-every	3	检查点间隔，即每处理多少个条目关闭一次 SSE 流触发重连
--log-level	INFO	日志级别，可传 DEBUG、INFO、WARNING、ERROR 等

自定义选项的完整用法：

uv run mcp-sse-polling-client --url http://localhost:3000/mcp --items 20 --checkpoint-every 5

该命令会让服务器连续处理 20 个条目，每处理 5 个条目关闭一次 SSE 流，从而在单次调用中触发多次重连，直观检验断点续传的可靠性。

值得留意的一个细节：仓库中的 README.md 将 --log-level 的默认值标注为 DEBUG，而 main.py 中 Click 选项的实际默认值为 "INFO"。以源码为准，默认日志级别为 INFO；如需观察重连与重放的详细过程，可显式传入 --log-level DEBUG。另外客户端在 main() 中会主动把 httpx2、httpcore2 两个库的日志级别压到 WARNING，以抑制底层 HTTP 库的噪音输出（见 main.py）。

服务器端：谁在制造断流与重连

要理解客户端的自动重连，必须同时看清服务器端的行为。演示服务器入口位于 server.py，核心代码如下：

starlette_app = app.streamable_http_app(
    event_store=InMemoryEventStore(),
    retry_interval=retry_interval,
    debug=True,
)

两个关键参数：

• event_store：传入 InMemoryEventStore 后，服务器即为每个 SSE 事件生成事件 ID，并在每条响应流开头注入一个"引导事件"（priming event），客户端因此始终握有一个可用于重连的 Last-Event-ID。不传 event_store 时，可续传能力与 ctx.close_sse_stream 回调都不会生效（该回调在 ServerRequestContext 上默认为 None，见 context.py）。
• retry_interval：服务器通过 SSE 的 retry: 字段告知客户端重连前的等待毫秒数。服务器端默认 100ms，对应启动命令 --retry-interval 100。

process_batch 工具的处理逻辑（server.py）是整条链路的发动机：

1. 每处理一个条目，通过 ctx.session.send_log_message(...) 发送一条进度日志；
2. 当 i % checkpoint_every == 0 且尚未处理完时，调用 await ctx.close_sse_stream() 主动关闭当前 SSE 流（并不取消处理器）；
3. 短暂 anyio.sleep(0.2) 等待客户端重连（该值必须大于 retry_interval 100ms）；
4. 处理完成后返回 CallToolResult。

关闭流之后继续发出的进度消息不再走原连接，而是全部写入事件存储；客户端重连后由事件存储补发——这就是"断开期间不丢消息"的保证。

底层源码：EventStore 接口与重放机制

事件存储是 SSE 可续传能力的契约层。接口定义在 streamable_http.py：

• store_event(stream_id, message) -> EventId：存储一个事件并返回生成的事件 ID；message 为 None 时表示引导事件（priming event）。
• replay_events_after(last_event_id, send_callback) -> StreamId | None：根据客户端上报的 Last-Event-ID，把该事件之后的所有事件通过回调补发给客户端。

演示服务器使用内存实现 InMemoryEventStore，它按流维护一个固定容量（默认 max_events_per_stream=100）的双端队列，为每个事件生成 uuid4 作为事件 ID，并在重放时跳过 message 为 None 的引导事件，只补发真实 JSON-RPC 消息。

客户端侧则完全透明：streamable_http_client 传输层自动处理引导事件、retry: 提示和 Last-Event-ID 重连，ClientSession 不暴露任何续传配置。从调用方视角看，await session.call_tool(...) 的返回与流未断开时别无二致。

一次完整轮询的时序

综合客户端与服务器两端的实现，一次带检查点的 process_batch 调用完整时序如下：

1. 客户端 POST 发起 tools/call 请求，服务器返回 SSE 响应，流首注入引导事件并携带 retry: 提示；
2. 服务器逐条处理条目，通过 SSE 推送进度消息；
3. 到达检查点时，服务器调用 close_sse_stream() 关闭当前 SSE 响应（释放连接槽位），继续处理并持续向事件存储写入进度；
4. 客户端传输层发现流结束，按 retry: 提示等待后，携带 Last-Event-ID 重新发起连接；
5. 服务器的事件存储将 Last-Event-ID 之后的事件全部重放给客户端；
6. 处理完毕，服务器发送最终 CallToolResult，客户端打印结果。

每一步都能在源码中得到印证：服务器侧断流与等待重连见 server.py，事件重放见 event_store.py，客户端侧透明的重连消费见 main.py。

生产环境注意事项

该示例旨在演示机制，直接用于生产前需要注意以下几点（与仓库内 examples/stories/sse_polling/README.md 的提醒一致）：

• InMemoryEventStore 仅用于演示：采用顺序递增的队列且无驱逐策略，生产环境应将 EventStore 接口落地为持久化存储（如数据库），以支持跨进程、跨重启的消息重放；
• 注意协议版本演进：Last-Event-ID 可续传与带会话的传输属于 2025 版协议（2025-11-25 规范）的能力，在 2026-07-28 协议（SEP-2575）中已被移除且无对等替代；2026 时代更接近的模式是基于持久化 DiscoverResult 的客户端重连（对应仓库中的 reconnect 示例）。若面向旧协议部署，请显式协商到 2025 时代版本；
• 重连等待与检查点节奏：服务器在检查点后 sleep(0.2) 必须大于 retry_interval，否则客户端尚未重连完成就开始下一轮写入，容易造成消息在重放窗口之外；
• 连接安全：演示代码因进程内测试场景关闭了 DNS 重绑定防护，真实部署应保留传输安全设置。

配套资源

• 客户端完整源码：mcp_sse_polling_client/main.py，打包配置见 pyproject.toml；
• 服务器完整源码：mcp_sse_polling_demo/server.py、mcp_sse_polling_demo/event_store.py；
• 协议规范实现：src/mcp/server/streamable_http.py（EventStore 接口与 retry_interval 语义）、src/mcp/server/context.py（close_sse_stream 回调）；
• 同主题的可运行故事用例：examples/stories/sse_polling/README.md，以及相邻场景 standalone_get（独立流关闭）与 reconnect（2026 时代重连）可对照阅读。