在之前的讨论中,我们深入剖析了 MCP 的协议层,揭示了 BaseSession 如何在 JSON-RPC 之上完成 SessionMessage 的帧化、请求–响应关联、并发收发与通知分发。本文则从'消息如何被打包'转向'消息如何被传输',探讨由传输层(Transport layer)承载这些帧化后的 SessionMessage,将它们在进程或网络之间来回搬运的实现细节。
一、MCP 的传输方式
MCP 支持多种传输方式以适配不同场景:
Stdio Transport:最轻量的 stdio 通道,借助标准输入/输出即可完成同机进程间的双向通信,适合本地进程间的通信。
HTTP + SSE Transport:基于 HTTP 的单向推流与单向发送组合(SSE + POST),适合浏览器前端与微服务后端的分离部署。服务器端使用 Server-Sent Events 推送消息给客户端,客户端通过 HTTP POST 发送请求。
Streamable HTTP Transport:更完善的 Streamable HTTP 场景中,POST 请求能返回 JSON 或切换到 SSE 流,在同一路径上混合使用 HTTP POST(发送请求)和 SSE(接收流式响应),支持 mcp-session-id 会话管理、断点续传以及 JSON/SSE 混合响应。
WebSocket Transport:基于 WebSocket 全双工通道,建立后客户端和服务端都可以随时推送 JSON-RPC 消息。适合低延迟、高频率的实时双向交互。
所有传输都遵循 JSON-RPC 2.0 格式。无论选哪一种,Host 启动时都会分别构造 ClientSession 和 ServerSession,将它们的读写流对接到相应的底层通道——本地时可把一端的 stdout 连到另一端的 stdin,远程时则通过 HTTP 或 WS 长连接互联。
标准输入/输出(stdio)
在 MCP 中,最轻量的本地进程间通信方式是通过标准输入/输出(stdio)来传递 JSON-RPC 消息。你只需要在主机(Host)和工具进程之间把 stdout 接入对方的 stdin,就能用同一套 BaseSession 协议去读写消息,而不必开启任何网络端口。
1.什么是 stdio
在操作系统和 C 语言标准库的语境里,stdio(Standard I/O,标准输入/输出)指的是一组预定义的、用于进程与外界交换数据的'流'接口。每个运行中的进程在启动时都会自动打开三条标准流:
- stdin(标准输入):文件描述符 0,默认从键盘或上游程序的输出中读取数据。
- stdout(标准输出):文件描述符 1,默认将程序的正常输出写到终端或下游程序的输入中。
- stderr(标准错误):文件描述符 2,默认将错误信息写到终端,通常不与 stdout 混用,以便错误可以单独重定向或捕获。
在 C 语言中,这些流由 <stdio.h> 头文件提供高层次的封装。从操作系统层面来看,这三条标准流其实都是文件描述符,对应于打开的文件或管道,可以借助管道将一个进程的 stdout 连接到另一个进程的 stdin,无需网络通信就能完成数据交换。
在 MCP 里,利用 stdio 的方式进行本地进程间通信,就是把'主机'的 stdout 通过管道接入'工具进程'的 stdin,同时把工具进程的 stdout 接到主机的 stdin,再使用 JSON-RPC 的消息格式在这两条流上来回读写。
这样做的好处有三点:
- 无需网络端口:整个通信走的是进程内部的管道,避免了网络配置和安全隔离的复杂度。
- 轻量透明:直接复用操作系统和语言自身对 stdio 的管理,既简单,性能开销又低。
- 兼容性好:几乎所有编程语言和操作系统都原生支持 stdio,使得跨语言、跨平台的工具集成都非常方便。
因此,stdio 在这里就是'最小代价、零配置'的进程间消息通道基础。
2.服务端的 stdio 实现
在服务端(也就是工具进程)里,stdio_server() 作为一个异步上下文管理器,重包装了系统的 sys.stdin 和 sys.stdout,确保它们都以 UTF-8 文本流的形式可读可写。
进入上下文后,两个协程分别启动:一个不断从 stdin 读取整行文本、把每行当作 JSON 去反序列化成 JSONRPCMessage,再封装为 SessionMessage 发入 read_stream;另一个从 write_stream 中取出需要下发给客户端的 SessionMessage,序列化成 JSON 并逐行写到 stdout。
# 服务端:stdio_server 的核心实现
@asynccontextmanager
async def ():
stdin:
stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding=))
stdout:
stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding=))
read_w, read = anyio.create_memory_object_stream()
write, write_r = anyio.create_memory_object_stream()
():
read_w:
line stdin:
:
msg = types.JSONRPCMessage.model_validate_json(line)
Exception e:
read_w.send(e)
read_w.send(SessionMessage(msg))
():
write_r:
session_msg write_r:
json_str = session_msg.message.model_dump_json(by_alias=, exclude_none=)
stdout.write(json_str + )
stdout.flush()
anyio.create_task_group() tg:
tg.start_soon(reader)
tg.start_soon(writer)
read, write


