在之前的讨论中，我们深入剖析了 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
