1. 从一条 JSON-RPC 消息说起MCP 到底在传什么如果你刚接触 MCPModel Context Protocol大概率会被ClientSession、Transport、SSE这些词绕晕。但把外壳剥掉MCP 在网络上跑的东西其实非常朴素它就是一条条符合 JSON-RPC 2.0 规范的文本消息。你完全可以把 MCP 想象成两个进程在用「带编号的纸条」对话——每张纸条上写着「我要调用哪个方法、参数是什么、编号是多少」对方回一张「编号相同、结果是啥」的纸条。编号也就是id是整套机制的灵魂因为它是唯一能把「请求」和「响应」配对起来的线索。MCP 能做什么它让大模型应用Client以统一协议去发现和调用外部能力Server比如列工具、调工具、读资源。适合谁适合想自己写 MCP Server、或者想搞懂 SDK 内部通信链路、排查「握手失败」「工具列表拉不到」「SSE 断流」这类问题的开发者。这篇笔记不堆概念而是从 JSON-RPC 报文格式出发逐层拆到ClientSession与Transport含 SSE怎么协作最后给你一套能直接跑的本地 SSE 端点验证流程。全程围绕 MCP、JSON-RPC、ClientSession、Transport、SSE 这几个关键词展开读完你应该能自己画出这条链路。先明确一个分层心智模型后面所有内容都挂在这上面层级组件职责协议层ClientSession维护 JSON-RPC 状态机、用 id 做请求响应关联、序列化/反序列化传输层Transport只管把字节搬来搬去屏蔽是管道还是网络向上暴露读写流关键点在于ClientSession不知道数据是走本地管道还是走 HTTPTransport也不懂tools/call是什么意思。两者通过统一的读写接口对接这就是 MCP 关注点分离设计的精髓。2. 前置准备TaoToken 接入与本地环境在动手拆 Transport 之前得先把「模型侧」和「协议侧」两条线准备好。模型侧我用 TaoToken 来提供兼容的对话与工具调用能力它的接口是标准 OpenAI 风格接入成本低协议侧则是本地跑一个 MCP Server 做实验对象。TaoToken 的官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个地址不加 UTM 参数。你需要先去控制台生成一个 API Key后续所有请求都用它做鉴权。环境上准备三样东西Python 3.10、mcp官方 SDK、以及一个能发 HTTP 请求的工具curl 或 httpx 都行。安装 SDKpip install mcp httpx如果你打算长期做编码类 Agent 开发反复调试工具调用链路可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合高频、长会话的场景只是临时验证模型行为的话用模型对话页https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 就够了。API Key 的创建入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入细节可以对照文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。注意MCP 的 Transport 层和模型 API 是两条独立的链路。模型负责「决定要不要调工具」MCP 负责「怎么把这次调用送到 Server 并拿回结果」。别把两者混在一起排查否则很容易定位错方向。3. 可复制配置JSON-RPC 报文与 Transport 骨架3.1 握手三阶段的真实报文MCP 生命周期严格遵循 JSON-RPC 2.0分握手协商、资源发现、业务执行三段。第一段握手是整个链路最容易出问题的地方先把三条报文写清楚。Step 1Client 发initialize请求带上自己的信息和能力声明{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: my-client, version: 0.1.0 }, capabilities: { sampling: {} } } }Step 2Server 返回确认包含元数据和能力清单{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, serverInfo: { name: demo-server, version: 0.1.0 }, capabilities: { tools: { listChanged: true } } } }这里有个容易忽略的点Server 此时只声明「我有工具调用能力」并不返回具体工具列表。这是刻意降低握手开销的设计工具 Schema 要等 Step 4 才拉。Step 3Client 发单向通知没有id也不需要响应{ jsonrpc: 2.0, method: notifications/initialized }这条通知一发双方状态机进入 Running握手才算真正完成。很多人卡在「initialize 成功了但 tools/list 报错」八成就是漏发了这条 notification。3.2 资源发现与工具调用Step 4主动拉工具列表{ jsonrpc: 2.0, id: 2, method: tools/list, params: {} }Server 返回完整定义含名称、描述和 JSON Schema 参数结构{ jsonrpc: 2.0, id: 2, result: { tools: [ { name: get_weather, description: 查询指定城市天气, inputSchema: { type: object, properties: { city: { type: string } }, required: [city] } } ] } }Step 5调用工具id换成新的唯一值{ jsonrpc: 2.0, id: 3, method: tools/call, params: { name: get_weather, arguments: { city: Hangzhou } } }Step 6Server 用相同id返回结果{ jsonrpc: 2.0, id: 3, result: { content: [{ type: text, text: Hangzhou: 22C, cloudy }] } }ClientSession在这一步做的事就是发出请求时把id3标记为 Pending 并 await收到响应后按id匹配、解包、恢复执行流。如果id对不上这条响应就会被丢弃表现为「调用一直挂起」。3.3 Transport 配置骨架Transport 层对上层屏蔽介质差异。Stdio 适合本地进程间通信SSE 适合分布式和远程调试。SSE 的通信模式是下行用 SSE 长连接做服务端推送上行用 HTTP POST 发指令从而模拟全双工。下面是一个 SSE Transport 的配置骨架重点是sse_read_timeout和timeout这两个参数它们直接决定长连接会不会被过早掐断from mcp.client.sse import sse_client async def connect_sse(): async with sse_client( urlhttp://127.0.0.1:8000/sse, headers{Authorization: Bearer your-token}, timeout5.0, # 建立连接的超时 sse_read_timeout300.0, # 长连接读取超时别设太小 ) as (read_stream, write_stream): # read_stream / write_stream 交给 ClientSession return read_stream, write_streamStdio 的骨架则简单得多它直接启动子进程生命周期和 Client 绑定from mcp import StdioServerParameters from mcp.client.stdio import stdio_client params StdioServerParameters( commandpython, args[demo_server.py], envNone, )4. 验证请求本地 SSE 端点跑通握手与往返光看报文不够得真跑一遍。我用官方 SDK 起一个带 SSE 的本地 Server然后用ClientSession连上去观察握手和消息往返。先写一个最小 Server暴露一个get_weather工具# demo_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server) mcp.tool() def get_weather(city: str) - str: 查询指定城市天气 return f{city}: 22C, cloudy if __name__ __main__: mcp.run(transportsse)启动它默认监听本地 SSE 端点python demo_server.py # 输出类似Uvicorn running on http://127.0.0.1:8000然后写 Client 侧用ClientSession完成握手、拉工具、调工具三步import asyncio from mcp import ClientSession from mcp.client.sse import sse_client async def main(): async with sse_client(http://127.0.0.1:8000/sse) as (read, write): async with ClientSession(read, write) as session: # Step 1-3握手 init await session.initialize() print(server:, init.serverInfo.name) print(capabilities:, init.capabilities) # Step 4资源发现 tools await session.list_tools() for t in tools.tools: print(tool:, t.name) # Step 5-6工具调用 result await session.call_tool(get_weather, {city: Hangzhou}) print(result:, result.content[0].text) asyncio.run(main())成功的话你会看到类似输出server: demo-server capabilities: tools{listChanged: True} tool: get_weather result: Hangzhou: 22C, cloudy这里session.initialize()内部其实帮你做了三件事发initialize、收响应、发notifications/initialized。所以你自己手写报文时千万别漏第三步。call_tool返回后ClientSession已经按id把响应匹配好并解包成了 Python 对象你拿到的就是干净的content。如果你想验证模型侧能不能正确触发工具调用可以把工具 Schema 喂给 TaoToken 的对话接口观察它是否生成符合inputSchema的arguments。这一步用模型对话页https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 手动试最快。5. 本篇常见错排查握手后 tools/list 报「not initialized」最常见原因是漏发notifications/initialized。用 SDK 时initialize()会自动发但如果你手写 Transport 或自己拼报文必须补上这条无id的通知。调用一直挂起、永远不返回九成是id不匹配。检查 Server 返回的id是否和请求完全一致类型也要一致1和1在严格实现里不等价。另外确认ClientSession没有在响应到达前被关闭。SSE 连接几秒后断开sse_read_timeout设太小。长连接场景下这个值要留足默认值往往不够。同时检查中间是否有反向代理提前掐断空闲连接。Stdio 模式下 Server 无响应多半是 Server 往 stdout 打了日志污染了 JSON-RPC 流。记住 stdio 传输里 stdout 是协议专用通道日志必须走 stderr。工具参数校验失败对照tools/list返回的inputSchema检查argumentsrequired字段一个都不能少类型也要匹配。鉴权头没带上SSE 的headers参数容易漏配导致 401。确认Authorization拼写和 token 有效性token 在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 管理。6. 继续深入的方向把这条链路跑通后你会发现 MCP 的复杂度几乎全在「状态机 id 关联 传输解耦」这三件事上。想继续深入建议从两个方向切一是自己实现一个最小 Transport只实现read/write两个方法体会ClientSession是怎么和它对接的二是研究listChanged通知看 Server 如何在运行中动态推送工具变更。如果你要把这套链路接到真实编码 Agent 里做长会话调试Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 在会话保持和工具调用稳定性上更省心接入过程中遇到协议层报错优先翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面把鉴权和请求格式讲得比较细。我自己的习惯是先用本地 SSE 端点把握手和往返验证通过再去接模型这样出问题时能立刻判断是协议层还是模型层省掉大量来回试错。