MCP 协议解读：STDIO 高效通信与 JSON-RPC 实战

本文深度解析 MCP 协议的传输机制与消息格式，涵盖 stdio、SSE 及自定义传输方式，剖析 JSON-RPC 2.0 的请求、响应与通知设计。
结合 RooCode 开源实现与天气查询案例，揭秘如何通过 MCP 实现跨进程通信与高效服务集成，为开发者提供从协议原理到落地实践的完整指南。

MCP 协议解读：STDIO 高效通信与 JSON-RPC 实战

Standard Input/Output (stdio)

stdio 传输允许通过标准输入和输出流进行通信。这对于本地集成和命令行工具特别有用。

使用 stdio 的情况：

• 构建命令行工具
• 实现本地集成
• 需要简单的进程通信
• 使用 shell 脚本

server 示例

app = Server("example-server")async with stdio_server() as streams:await app.run(streams[0],streams[1],app.create_initialization_options())

client 示例

params = StdioServerParameters(command="./server",args=["--option", "value"]
)async with stdio_client(params) as streams:async with ClientSession(streams[0], streams[1]) as session:await session.initialize()

Server-Sent Events (SSE)

SSE 传输通过 HTTP POST 请求实现服务器到客户端的流式传输，并支持客户端到服务器的通信。
目前非本地程序开发，都可使用 SSE 实现，已有的 HTTP 接口都可方便的改下 Controller 层对外提供服务。
不过 MCP 目前官方还没有标准的鉴权方式（roadmap 已规划），只能每个 mcp-server 通过参数自定义实现

支持使用 SSE 的情况：

• 仅需要服务器到客户端的流式传输
• 在受限网络中工作
• 实现简单更新

server 示例

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Routeapp = Server("example-server")
sse = SseServerTransport("/messages")# Define handler functions
async def handle_sse(scope, receive, send):async with sse.connect_sse(scope, receive, send) as streams:await app.run(streams[0], streams[1], app.create_initialization_options())async def handle_messages(scope, receive, send):await sse.handle_post_message(scope, receive, send)# Create and run Starlette app
starlette_app = Starlette(routes=[Route("/sse", endpoint=handle_sse),Route("/messages", endpoint=handle_messages, methods=["POST"]),]
)

client 示例

async with sse_client("http://localhost:8000/sse") as streams:async with ClientSession(streams[0], streams[1]) as session:await session.initialize()

自定义传输协议

MCP 使得实现针对特定需求的自定义传输变得简单。任何传输实现只需符合传输接口：
您可以实现自定义传输，用于：

• 自定义网络协议
• 专用通信通道
• 与现有系统集成
• 性能优化

示例

@contextmanager
async def create_transport(read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],write_stream: MemoryObjectSendStream[JSONRPCMessage]
):"""Transport interface for MCP.Args:read_stream: Stream to read incoming messages fromwrite_stream: Stream to write outgoing messages to"""async with anyio.create_task_group() as tg:try:# Start processing messagestg.start_soon(lambda: process_messages(read_stream))# Send messagesasync with write_stream:yield write_streamexcept Exception as exc:# Handle errorsraise excfinally:# Clean uptg.cancel_scope.cancel()await write_stream.aclose()await read_stream.aclose()

消息格式 JSON-RPC

MCP 使用 JSON-RPC 2.0 作为其传输格式。

JSON-RPC

目前存在两种 JSON-RPC 规范，分别是 JSON-RPC 1.0 和 JSON-RPC2.0。

• JSON-RPC 1.0 在多个方面缺乏实用性，特别是缺少名称参数和错误消息解释，引起的麻烦超出了预期。使得它更像是一种点对点的交流方式。
• JSON RPC 2.0 规范则解决了 1.0 版本的缺陷，表现得更加先进。使用客户机 - 服务器模式，同时增加了多种传输方式的支持。

具体区别：

• 结构变化：2.0 版本引入了强制性的 jsonrpc:“2.0” 字段，将 id 字段变为可选，并规范了响应结构（成功必须有 result，错误必须有 error）。
• 功能增强：2.0 版本新增了 批处理功能，改进了 通知机制，并支持两种参数传递方式（按位置的数组和按名称的对象）。
• 错误处理优化：2.0 版本标准化了 错误对象格式（code, message, data）并定义了预设错误代码范围，使错误处理更规范和一致。

MCP 的传输层负责将 MCP 协议消息转换为 JSON-RPC 格式进行传输，并将接收到的 JSON-RPC 消息转换回 MCP 协议消息。
MCP 的 RPC 消息有请求，响应，通知 3 种类型。

请求 Requests

{jsonrpc: "2.0",id: number | string,method: string,params?: object
}

客户端请求类型

class ClientRequest(RootModel[PingRequest| InitializeRequest| CompleteRequest| SetLevelRequest| GetPromptRequest| ListPromptsRequest| ListResourcesRequest| ListResourceTemplatesRequest| ReadResourceRequest| SubscribeRequest| UnsubscribeRequest| CallToolRequest| ListToolsRequest]
):pass

响应 Responses

{jsonrpc: "2.0",id: number | string,result?: object,error?: {code: number,message: string,data?: unknown}
}

服务端响应类型

class ServerResult(RootModel[EmptyResult| InitializeResult| CompleteResult| GetPromptResult| ListPromptsResult| ListResourcesResult| ListResourceTemplatesResult| ReadResourceResult| CallToolResult| ListToolsResult]
):pass

通知 Notifications

{jsonrpc: "2.0",method: string,params?: object
}

通知类型

# 客户端通知
class ClientNotification(RootModel[CancelledNotification| ProgressNotification| InitializedNotification| RootsListChangedNotification]
):pass# 服务端通知
class ServerNotification(RootModel[CancelledNotification| ProgressNotification| LoggingMessageNotification| ResourceUpdatedNotification| ResourceListChangedNotification| ToolListChangedNotification| PromptListChangedNotification]
):pass

MCP 开源实现分析 -Cline/RooCode

1. 每个 vscode 实例启动时，插件 RooCode 都会自动启动 2 个进程。
2. RooCode 支持配置全局的 MCP-Server，也支持在 Projet 级别配置 MCP-Server,从而实现按项目隔隔离参数配置；

MCP 客户端配置

{"MCPServers": {"weather": {"command": "uv","args": ["--directory","/mnt/github/MCP/weather","run","weather.py"],"alwaysAllow": ["get_forecast","get_alerts"],"disabled": false}}
}

客户端进程分析

1. uv 作为父进程，负责运行和管理 mcp-server
2. MCP.run(transport='stdio') 使用 python 虚拟环境运行 weather.py，对外使用 stdio 通信
   py

# ps -ef --forest | grep weather.py
## `uv` 是一个进程管理工具，它负责启动和管理子进程（在这里是 weather.py）。它本身会作为一个父进程存在。
\_ uv --directory /mnt/github/MCP/weather run weather.py
## 初始化了 `FastMCP` 服务，并通过 `MCP.run(transport='stdio')` 开启了一个基于标准输入/输出的通信服务。\_ /mnt/github/MCP/weather/.venv/bin/python3 weather.py

结语

看完这篇解析，是不是对 MCP 协议的设计思路更清晰了？无论是本地命令行工具还是分布式服务，MCP 都能用灵活的传输方式和标准的 JSON-RPC 帮你 " 搭桥 "。
如果你正头疼进程通信或服务集成，不妨试试 MCP 框架——毕竟，好的协议就像隐形的桥梁，默默支撑着每一次高效交互。😃
今天分享的主要是传输层的实现细节，宏观的 MCP 工作原理见昨天分享的 模型无关的AI集成革命：MCP协议解读。

---

这儿的文章都会同步发送到：极客工具 XTool

🌟 极客工具：共享开源力量，成就超级个体