人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载本篇文章围绕官方 python-sdk 仓库中的examples/clients/sse-polling-client演示客户端展开完整讲解 SSEServer-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通过streamable_http_client(url)建立读写流在ClientSession中执行await session.initialize()完成协议握手调用session.list_tools()打印服务器暴露的工具列表调用process_batch工具并打印最终结果。整个process_batch调用期间服务器会周期性关闭 SSE 流触发客户端重连而客户端无需任何额外配置即可自动完成重连与消息恢复——这正是该演示希望传达的核心能力。客户端命令行选项详解mcp-sse-polling-client使用 Click 定义命令行参数全部选项及其含义如下表选项默认值说明--urlhttp://localhost:3000/mcp服务器 SSE 端点 URL--items10要处理的条目数量会作为process_batch的参数传给服务器--checkpoint-every3检查点间隔即每处理多少个条目关闭一次 SSE 流触发重连--log-levelINFO日志级别可传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_storeInMemoryEventStore(), retry_intervalretry_interval, debugTrue, )两个关键参数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是整条链路的发动机每处理一个条目通过ctx.session.send_log_message(...)发送一条进度日志当i % checkpoint_every 0且尚未处理完时调用await ctx.close_sse_stream()主动关闭当前 SSE 流并不取消处理器短暂anyio.sleep(0.2)等待客户端重连该值必须大于retry_interval100ms处理完成后返回CallToolResult。关闭流之后继续发出的进度消息不再走原连接而是全部写入事件存储客户端重连后由事件存储补发——这就是断开期间不丢消息的保证。底层源码EventStore 接口与重放机制事件存储是 SSE 可续传能力的契约层。接口定义在 streamable_http.pystore_event(stream_id, message) - EventId存储一个事件并返回生成的事件 IDmessage为None时表示引导事件priming event。replay_events_after(last_event_id, send_callback) - StreamId | None根据客户端上报的Last-Event-ID把该事件之后的所有事件通过回调补发给客户端。演示服务器使用内存实现 InMemoryEventStore它按流维护一个固定容量默认max_events_per_stream100的双端队列为每个事件生成uuid4作为事件 ID并在重放时跳过message为None的引导事件只补发真实 JSON-RPC 消息。客户端侧则完全透明streamable_http_client传输层自动处理引导事件、retry:提示和Last-Event-ID重连ClientSession不暴露任何续传配置。从调用方视角看await session.call_tool(...)的返回与流未断开时别无二致。一次完整轮询的时序综合客户端与服务器两端的实现一次带检查点的process_batch调用完整时序如下客户端 POST 发起tools/call请求服务器返回 SSE 响应流首注入引导事件并携带retry:提示服务器逐条处理条目通过 SSE 推送进度消息到达检查点时服务器调用close_sse_stream()关闭当前 SSE 响应释放连接槽位继续处理并持续向事件存储写入进度客户端传输层发现流结束按retry:提示等待后携带Last-Event-ID重新发起连接服务器的事件存储将Last-Event-ID之后的事件全部重放给客户端处理完毕服务器发送最终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.pyEventStore接口与retry_interval语义、src/mcp/server/context.pyclose_sse_stream回调同主题的可运行故事用例examples/stories/sse_polling/README.md以及相邻场景standalone_get独立流关闭与reconnect2026 时代重连可对照阅读。赞分享人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐如何优化dlib-android-app性能提升人脸检测速度的5个实用技巧如何优化dlib android app性能提升人脸检测速度的5个实用技巧 dlib android app是一款基于dlib android库开发的Andr人工智能MCP 服务MCP ClientsTypeScript SDK 实战基于 SEP-1699 的 SSE 服务端主动断开与 Last-Event-ID 重连恢复机制TypeScript SDK 实战基于 SEP 1699 的 SSE 服务端主动断开与 Last Event ID 重连恢复机制 导读 在 typescrip人工智能MCP 服务MCP Clientsqwen-code 的 ACP-over-HTTP 可断点续传会话流基于 SSE Last-Event-ID 的事件重放设计qwen code 的 ACP over HTTP 可断点续传会话流基于 SSE Last Event ID 的事件重放设计 导读 本篇技术指南围绕 sse人工智能AI Agent代码智能体工具调用交互助手CLIQwen上一篇凌晨3点你的Annotators服务雪崩了怎么办一份“反脆弱”的LLM运维手册下一篇如何用终极跨平台串口调试工具提升硬件开发效率SerialPortAssistant完全指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考