1. 为什么 MCP 协议需要双向通信从一次工具调用卡死说起如果你正在给大模型接工具链大概率遇到过这种场景模型调用一个本地脚本查数据脚本跑完了但模型那边迟迟收不到结果或者工具执行到一半想上报进度却只能等整个请求结束才把结果一次性吐回来。传统 HTTP 的请求-响应模式在这里会显得很别扭——客户端发一个请求服务端处理完再返回中间没有任何通道让服务端主动说话。MCP 协议Model Context Protocol要解决的就是这个问题。它是一套让大模型与外部工具、数据源之间标准化对话的协议核心能力是双向通信客户端可以发请求服务端也可以主动推送通知、进度、日志甚至反向请求。适合谁适合正在搭建 MCP 工具链的开发者尤其是需要本地进程调用和远程服务混合编排的场景。MCP 协议定义了两种主要传输方式Stdio 和 SSE。Stdio 走标准输入输出适合本地进程间通信SSE 走 HTTP 服务器推送事件适合远程服务。两者底层都跑 JSON-RPC 2.0 消息格式区别在于消息怎么在两端之间流动。这篇文章会把两种传输的配置、消息流、验证方法拆开讲给出可以直接复制的片段并用日志和抓包确认双向消息是否真的通了。我试过在同一个工具链里混用两种传输本地文件操作走 Stdio远程知识库查询走 SSE踩过的坑主要集中在消息 ID 对不上和 SSE 连接被中间层掐断。下面按实际搭建顺序展开。2. TaoToken 前置准备拿到 Base URL、API Key 和 Model ID在写 MCP 客户端配置之前需要先有一个能调用模型的入口。TaoToken 提供兼容 OpenAI 风格的 APIMCP 客户端在发起模型请求时需要三件套Base URL、API Key、Model ID。这三样在后续的 Stdio 和 SSE 配置里都会用到。先到官网注册并进入控制台。控制台里可以创建 API Key复制出来保存好后面配置文件里要用。模型对话页面可以用来快速验证 Key 是否可用不用写代码就能发一条消息看返回。具体操作路径打开 https://taotoken.net/api-keys 创建一个新的 Key命名建议带上用途比如 mcp-local-test方便后面排查是哪个 Key 出的问题。创建后立即复制页面刷新后就不再完整显示。模型 ID 在模型对话页面或者文档里可以查到常见的有 claude 系列和 gpt 系列。MCP 工具链里通常用 Claude 系列做工具调用因为对 function calling 支持比较完整。Base URL 统一用 https://taotoken.net/api 注意不要在后面多加斜杠也不要带 UTM 参数配置文件里写干净地址。如果你打算长期跑编码类 Agent可以了解 Coding Plan它更适合高频工具调用场景。接入文档在 https://taotoken.net/doc 里面有各语言 SDK 的示例。拿到三件套后先别急着写 MCP 配置用 curl 验证一下 Key 是否有效curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer 你的API_KEY \ -H Content-Type: application/json \ -d { model: claude-3-5-sonnet, messages: [{role: user, content: ping}] }返回里有 choices 字段且 content 非空说明 Key 和 Base URL 都没问题。这一步很重要因为后面 MCP 配置出错时你要能区分是模型入口的问题还是 MCP 传输层的问题。3. 可复制配置Stdio 与 SSE 的 JSON/TOML 片段这一节给出两种传输方式的实际配置。MCP 客户端种类很多这里以常见的配置文件格式为例路径和字段名保持通用你按自己用的客户端调整。3.1 Stdio 配置片段Stdio 模式下MCP 客户端会启动一个本地子进程通过 stdin/stdout 交换 JSON-RPC 消息。配置里需要写清楚启动命令、参数、环境变量。以 JSON 配置为例放在客户端的 mcp servers 配置段{ mcpServers: { local-weather: { command: python3, args: [/Users/you/mcp/weather_server.py, --transportstdio], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: 你的API_KEY, TAOTOKEN_MODEL: claude-3-5-sonnet } } } }关键点command 必须是可执行文件路径args 里第一个是脚本路径后面是传给脚本的参数。env 里放三件套服务端脚本读取这些变量去调用模型。注意不要把 Key 硬编码在脚本里放 env 里方便轮换。如果你的客户端用 TOML等价写法[mcp_servers.local-weather] command python3 args [/Users/you/mcp/weather_server.py, --transportstdio] [mcp_servers.local-weather.env] TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY 你的API_KEY TAOTOKEN_MODEL claude-3-5-sonnetStdio 服务端脚本的最小骨架用 Python 写import sys, json, os def handle(request): method request.get(method) req_id request.get(id) if method query_weather: city request[params][city] result f{city} 当前气温 28 度 return {jsonrpc: 2.0, id: req_id, result: result} return {jsonrpc: 2.0, id: req_id, error: {code: -32601, message: method not found}} for line in sys.stdin: line line.strip() if not line: continue req json.loads(line) resp handle(req) sys.stdout.write(json.dumps(resp) \n) sys.stdout.flush()注意 flush否则消息会卡在缓冲区里客户端一直等不到响应。3.2 SSE 配置片段SSE 模式下客户端先连一个 SSE 端点拿会话再通过 HTTP POST 发消息。配置里写 URL 和鉴权头。{ mcpServers: { remote-knowledge: { url: https://your-mcp-server.example.com/sse, headers: { Authorization: Bearer 你的MCP_SERVER_TOKEN }, env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: 你的API_KEY, TAOTOKEN_MODEL: claude-3-5-sonnet } } } }SSE 服务端的关键是保持连接不中断并且每条消息以data:开头、以两个换行结尾。一个最小实现from flask import Flask, Response, request import json, queue, threading app Flask(__name__) client_queues {} app.route(/sse) def sse(): client_id request.args.get(client_id, default) q queue.Queue() client_queues[client_id] q def stream(): while True: msg q.get() yield fdata: {json.dumps(msg)}\n\n return Response(stream(), mimetypetext/event-stream) app.route(/message, methods[POST]) def message(): body request.get_json() client_id body.get(client_id, default) if client_id in client_queues: client_queues[client_id].put(body) return {ok: True}生产环境要加心跳每 15 秒发一个空注释:\n\n防止 Nginx 或网关把空闲连接断掉。Nginx 侧配置proxy_read_timeout 3600sproxy_buffering off否则 SSE 会被缓冲消息延迟很大。三件套在 SSE 场景下同样要写全Base URL 是 https://taotoken.net/api Key 是你在控制台创建的Model ID 按实际用的填。如果客户端支持把这三样放在环境变量里服务端和客户端都能读到。4. 验证请求用日志和抓包确认双向消息收发正常配置写完不代表通了。MCP 的双向通信要验证两件事客户端发的请求服务端收到了服务端推的消息客户端收到了。下面给两种传输的验证方法。4.1 Stdio 验证最直接的办法是把 stdin/stdout 都 tee 到日志文件。启动服务端时python3 weather_server.py --transportstdio 2server.err.log | tee server.out.log然后在客户端发一个请求观察 server.out.log 里有没有对应的 JSON 行。正常情况你会看到类似{jsonrpc:2.0,id:1,method:query_weather,params:{city:北京}} {jsonrpc:2.0,id:1,result:北京 当前气温 28 度}如果只有请求没有响应检查服务端脚本有没有 flush或者是不是在某个分支里 return 了但没写 stdout。如果响应有但客户端没收到检查客户端是不是按行读取有些客户端会等 EOF 才解析那就永远等不到。再验证服务端主动推送。在服务端脚本里加一段定时发通知的逻辑import threading, time def push_notifications(): while True: time.sleep(5) note {jsonrpc: 2.0, method: notifications/progress, params: {percent: 50}} sys.stdout.write(json.dumps(note) \n) sys.stdout.flush() threading.Thread(targetpush_notifications, daemonTrue).start()客户端日志里应该每 5 秒出现一条 notifications/progress没有 id 字段因为通知不需要响应。这就是双向通信里服务端主动说话的部分。4.2 SSE 验证SSE 用 curl 直接连看事件流curl -N https://your-mcp-server.example.com/sse?client_idtest-N 关闭缓冲你会看到data: {...}一行行刷出来。如果卡住不动先检查服务端有没有发心跳再检查中间有没有反向代理缓冲。发消息用另一个终端curl -X POST https://your-mcp-server.example.com/message \ -H Content-Type: application/json \ -d {client_id:test,jsonrpc:2.0,id:1,method:query_weather,params:{city:上海}}第一个终端里应该出现对应的响应事件。如果 POST 返回 200 但 SSE 流里没有检查 client_id 是否一致服务端队列是不是按 client_id 存的。抓包可以用 tcpdump 看 HTTP/2 帧但更实用的是在服务端加访问日志记录每条消息的 id、method、方向。MCP 的 JSON-RPC 消息 id 是关联请求和响应的关键id 对不上就会出现「响应找不到请求」的报错。验证模型调用是否正常可以在服务端处理请求时调用 TaoToken 的模型对话接口把工具执行结果交给模型润色。如果模型返回正常说明三件套配置没问题如果返回 401那就是 Key 或 Base URL 的问题跟 MCP 传输层无关。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节对照真实报错给出定位思路。MCP 工具链的报错经常混在一起要分层排查模型入口层、MCP 传输层、客户端解析层。5.1 401 Unauthorized报错长这样Error: 401 Unauthorized {error:{message:Invalid API key,type:invalid_request_error}}这是模型入口层的错跟 Stdio/SSE 无关。检查三件套Base URL 是不是 https://taotoken.net/api Key 有没有复制完整Model ID 是不是当前账号可用的。常见坑是 Key 前后带了空格或者配置文件里用了中文引号。用 curl 单独测一次排除 MCP 配置的干扰。5.2 local proxy failed报错local proxy failed: dial tcp 127.0.0.1:7890: connect: connection refused这个错说明客户端或服务端在尝试走本地代理端口但那个端口没有服务在听。MCP 的 Stdio 模式是本地进程通信不需要代理SSE 模式如果服务端在本地也不需要。检查环境变量里有没有 HTTP_PROXY、HTTPS_PROXY 指向了一个不存在的端口清掉再试。配置文件里也不要写 proxy 字段。5.3 reading choices 相关报错报错json: cannot unmarshal object into Go struct field ChatCompletion.choices of type []...或者 Python 里KeyError: choices。这通常是模型返回的不是标准 chat completion 格式可能是返回了错误对象但客户端直接去读 choices。先打印原始返回体看是不是 401 或 429 被包装成了 200。如果是流式返回要按 SSE 格式逐块解析不能一次性 json.loads 整个响应。5.4 OAuth 相关报错报错OAuth token exchange failed: invalid_grantSSE 服务端如果启用了 OAuth 鉴权客户端拿到的 token 过期或 scope 不对就会报这个。检查 token 的有效期以及请求的 scope 是否包含 MCP 服务需要的权限。如果是 PKCE 模式确认 code_verifier 和 code_challenge 匹配。调试阶段可以先关掉 OAuth用静态 Bearer token 验证传输层是否通再逐步加鉴权。5.5 消息 ID 对不上这个不报错但表现为客户端一直等响应直到超时。检查服务端返回的 id 是不是原样带回请求的 id。JSON-RPC 2.0 要求响应里的 id 必须和请求一致通知消息没有 id。如果服务端用了自增 id 而不是回传客户端就匹配不上。在服务端日志里打印请求 id 和响应 id对比一下。5.6 SSE 连接被断开表现为事件流跑几秒就断客户端重连后会话丢失。检查 Nginx 的 proxy_read_timeout 和 proxy_buffering检查服务端有没有发心跳。如果用了负载均衡确认 SSE 连接是粘性的或者用 Redis 做消息总线让任意节点都能推消息。6. 把 MCP 双向通信接进你的工具链走到这里Stdio 和 SSE 两种传输的配置、验证、排错都过了一遍。实际搭建时建议先用 Stdio 跑通一个本地工具确认 JSON-RPC 消息能双向流动再切到 SSE 接远程服务。Stdio 的好处是链路短出问题容易定位SSE 的好处是能跨机器、能多客户端但中间层多排查成本高。三件套在两种模式下都要写全Base URL 用 https://taotoken.net/api API Key 在 https://taotoken.net/api-keys 创建Model ID 按实际用的填。模型对话页面可以用来快速验证 Key接入文档在 https://taotoken.net/doc 有各语言示例。如果长期跑编码类 AgentCoding Plan 比按次调用更划算。最后留一个实用技巧在服务端每条消息处理前后都打一行日志格式统一成[方向][id][method]比如[recv][1][query_weather]和[send][1][result]。这样出问题时 grep 一下就能看出消息在哪个环节断了比抓包快得多。MCP 的双向通信本质就是消息在两端之间来回跑把消息轨迹记清楚大部分问题都能自己定位。