MCP协议底层原理深度剖析：从JSON-RPC 2.0到多传输层实现
引言
2026年，AI Agent已然成为技术圈最炙手可热的方向。从单体的聊天机器人到能够自主调用工具、执行复杂任务的多智能体系统，支撑这一切的底层基础设施正在被一场静默的协议革命重塑。而这场革命的核心，就是MCP（Model Context Protocol）——模型上下文协议。
Anthropic 在 2024 年底提出的 MCP，被业界称为"AI 时代的 USB-C 接口"。但大多数开发者对它的理解停留在"能让 AI 调用工具"的层面，对其底层协议设计、传输层实现、生命周期管理等核心机制缺乏深入认知。
本文将以底层工程视角，从 JSON-RPC 2.0 协议基础出发，深入剖析 MCP 的协议层设计、双传输层实现（stdio/SSE）、连接生命周期、以及生产级实践，并附带完整的 Python 代码示例，帮助读者建立对 MCP 协议的全景技术认知。
---
一、MCP 协议栈全景
MCP 的整体架构可分为三层：
┌─────────────────────────────────────┐ │ 应用层 (Application) │ │ ┌─────────┐ ┌─────────┐ │ │ │ Host │◄─────►│ Server │ │ │ │(Client) │ │(Tool) │ │ │ └────┬────┘ └────┬────┘ │ ├───────┼──────────────────┼─────────┤ │ │ 协议层 │ │ │ │ JSON-RPC 2.0 │ │ │ │ × MCP 原语 │ │ ├───────┼──────────────────┼─────────┤ │ │ 传输层 │ │ │ ┌────┴────┐ ┌────┴────┐ │ │ │ stdio │ or │ SSE │ │ │ └─────────┘ └─────────┘ │ └─────────────────────────────────────┘• **应用层**：Host（宿主，如 Claude Desktop、IDE 插件）和 Server（工具/数据源提供方）
• **协议层**：基于 JSON-RPC 2.0 的消息格式 + MCP 定义的原语（Tools / Resources / Prompts）
• **传输层**：stdio（本地进程通信）或 SSE（远程 HTTP 通信）
---
二、协议层基石：JSON-RPC 2.0 深度分析
2.1 JSON-RPC 2.0 消息规范
MCP 的协议层完全建立在 JSON-RPC 2.0 之上。JSON-RPC 是一种轻量级、无状态的远程过程调用协议，使用 JSON 作为数据格式。为什么选择 JSON-RPC 而不是 gRPC 或 REST？原因有三：
1.极简：协议规范只有一页纸，实现成本极低
2.传输无关：可在 stdio、TCP、HTTP、WebSocket 等任意传输层上运行
3.天然支持异步通知：无需等待响应的"通知"消息，适合流式场景
JSON-RPC 2.0 定义了三种消息类型：
请求（Request）：
{ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "get_weather", "arguments": { "city": "Beijing" } } }响应（Response）——成功：
{ "jsonrpc": "2.0", "id": 1, "result": { "content": [ {"type": "text", "text": "北京当前温度：28°C"} ] } }响应（Response）——错误：
{ "jsonrpc": "2.0", "id": 1, "error": { "code": -32603, "message": "Internal error", "data": {"details": "API rate limit exceeded"} } }通知（Notification）——无 id，无需响应：
{ "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} }2.2 MCP 标准错误码
| 错误码 | 含义 | 说明 |
|--------|------|------|
| -32700 | Parse error | JSON 解析错误 |
| -32600 | Invalid Request | 请求结构无效 |
| -32601 | Method not found | 方法不存在 |
| -32602 | Invalid params | 参数无效 |
| -32603 | Internal error | 服务器内部错误 |
| -32000 ~ -32099 | Server error | 自定义服务器错误 |
| -32100 | Resource not found | 资源未找到（MCP 扩展） |
| -32101 | Tool execution error | 工具执行错误（MCP 扩展） |
2.3 MCP 核心原语
MCP 在 JSON-RPC 2.0 之上定义了三大核心原语，构成了协议的功能语义：
Tools（工具）——"做什么"
• 定义可被 AI 调用的外部工具
• 包含名称、描述、输入参数 schema（JSON Schema）
• 调用方式：`tools/call` 方法
Resources（资源）——"读什么"
• 暴露数据源（文件、数据库、API 响应等）
• 支持 URI 模式进行资源定位
• 读取方式：`resources/read` 方法
Prompts（提示模板）——"怎么说"
• 预定义的提示词模板
• 包含模板参数和交互逻辑
• 获取方式：`prompts/get` 方法
这三者的设计哲学可以概括为：Tools 写、Resources 读、Prompts 说，形成了一个完整的交互三角。
---
三、传输层详解：stdio vs SSE
3.1 stdio 传输：本地进程间通信
stdio 传输是 MCP 最基础也是最高效的传输方式。它通过子进程的标准输入（stdin）和标准输出（stdout）进行 JSON-RPC 消息的双向传输。
Python 服务端实现：
import asyncio import json import sys from mcp.server import Server from mcp.server.stdio import stdio_server # 创建 MCP 服务器实例 server = Server( name="my-tool-server", version="1.0.0", capabilities={ "tools": {}, # 声明支持工具调用 } ) # 注册工具 @server.list_tools() async def list_tools(): from mcp.types import Tool return [ Tool( name="calculator", description="执行数学运算", inputSchema={ "type": "object", "properties": { "expr": { "type": "string", "description": "数学表达式，如 2 + 3 * 4" } }, "required": ["expr"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): from mcp.types import TextContent if name == "calculator": expr = arguments["expr"] try: result = eval(expr, {"__builtins__": {}}, {}) return [TextContent(type="text", text=str(result))] except Exception as e: return [TextContent(type="text", text=f"错误：{str(e)}")] raise ValueError(f"未知工具: {name}") async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ == "__main__": asyncio.run(main())Python 客户端连接：
import asyncio from mcp import ClientSession, StdioClientTransport from mcp.client.stdio import get_default_environment async def main(): # 配置 stdio 传输：启动服务端子进程 transport = StdioClientTransport( command="python", args=["server.py"], env=get_default_environment() ) async with ClientSession(transport) as session: # 1. 初始化握手 await session.initialize() # 2. 列出可用工具 tools = await session.list_tools() print(f"可用工具: {[t.name for t in tools]}") # 3. 调用工具 result = await session.call_tool( "calculator", {"expr": "2 + 3 * 4"} ) print(f"计算结果: {result.content[0].text}") asyncio.run(main())stdio 传输的优势：
• 零网络开销，延迟最低（微秒级）
• 安全性高——子进程在本地运行，无网络暴露面
• 适合 CLI 工具、本地集成、开发调试
3.2 SSE 传输：远程 HTTP 流式通信
SSE（Server-Sent Events）是一种服务器向客户端推送数据的 HTTP 技术。MCP 的 SSE 方案采用双向混合通信：服务器通过 SSE 向客户端推送消息，客户端通过 HTTP POST 向服务器发送消息。
# SSE 服务端（使用 Starlette） from mcp.server import Server from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route server = Server("example-server", capabilities={"tools": {}}) sse = SseServerTransport("/messages") async def handle_sse(request): """SSE 端点：服务器→客户端流式推送""" async with sse.connect_sse( request.scope, request.receive, request.send ) as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options() ) async def handle_messages(request): """消息端点：客户端→服务器 POST""" await sse.handle_post_message( request.scope, request.receive, request.send ) starlette_app = Starlette( routes=[ Route("/sse", endpoint=handle_sse), Route("/messages", endpoint=handle_messages, methods=["POST"]), ] )SSE 客户端连接：
from mcp.client.sse import sse_client from mcp import ClientSession async def main(): async with sse_client("http://localhost:8000/sse") as streams: async with ClientSession(*streams) as session: await session.initialize() tools = await session.list_tools() print(f"远程可用工具: {[t.name for t in tools]}") asyncio.run(main())SSE vs stdio 对比：
| 维度 | stdio | SSE |
|------|-------|-----|
| 通信方式 | 进程内管道 | HTTP + 流 |
| 延迟 | 纳秒~微秒级 | 毫秒级 |
| 部署模式 | 本地子进程 | 远程服务器 |
| 安全性 | 天然隔离 | 需要认证/TLS |
| 适用场景 | CLI、本地集成 | 远程API、微服务 |
| 连接数 | 1:1 | 1:N |
3.3 自定义传输层实现
MCP 的 Transport 接口非常简洁，只需要实现三个方法：
from typing import AsyncContextManager, AsyncIterator from anyio import create_memory_object_stream from mcp.types import JSONRPCMessage @contextmanager async def custom_transport(): """自定义传输实现""" # 创建双向内存流 read_writer, read_stream = create_memory_object_stream[JSONRPCMessage](0) write_stream, write_reader = create_memory_object_stream[JSONRPCMessage](0) async def message_handler(): """消息处理主循环""" async with read_writer: async for message in write_reader: # 处理消息逻辑... pass async with anyio.create_task_group() as tg: tg.start_soon(message_handler) try: yield read_stream, write_stream finally: tg.cancel_scope.cancel()这种设计使得 MCP 可以运行在任何传输层之上——WebSocket、Unix Socket、甚至 MQTT——只需实现 Transport 接口。
---
四、连接生命周期：从握手到关闭
MCP 的连接生命周期包括三个阶段：
第一阶段：初始化握手（Handshake）
客户端和服务器在建立连接后首先进行协议版本和能力协商：
客户端 → 服务器: initialize (协议版本 + 客户端能力) 服务器 → 客户端: initialized (服务器能力 + 协议版本) 客户端 → 服务器: initialized (确认通知)# 初始化请求 { "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {}, "resources": {} }, "clientInfo": { "name": "my-agent", "version": "1.0.0" } } } # 初始化响应 { "jsonrpc": "2.0", "id": 0, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {}, "prompts": {} }, "serverInfo": { "name": "weather-tool", "version": "1.0.0" } } }**关键设计点**：握手阶段是**严格的先后顺序**——在初始化完成之前，服务器不得接受任何工具调用请求。这避免了协议版本不兼容导致的解析错误。
第二阶段：正常运行（Operation）
握手完成后，客户端可以自由调用工具、读取资源、获取提示模板。这个阶段的通信是完全异步的——客户端可以同时发出多个请求，服务器可以按任意顺序响应。
# 并发调用示例 async with ClientSession(transport) as session: await session.initialize() # 并发发送三个请求 task1 = session.call_tool("weather", {"city": "北京"}) task2 = session.call_tool("weather", {"city": "上海"}) task3 = session.list_tools() results = await asyncio.gather(task1, task2, task3)第三阶段：优雅关闭
# 客户端关闭 await session.close() # 服务器收到关闭信号后清理资源---
五、生产级实践：多连接池与负载均衡
在生产环境中，单一 MCP 客户端往往需要管理多个服务器连接。这里给出一个多连接池的实现方案：
import asyncio from mcp import ClientSession from typing import Dict, Optional class MCPConnectionPool: """MCP 连接池：管理和复用多个 MCP 服务器连接""" def __init__(self): self._sessions: Dict[str, ClientSession] = {} self._locks: Dict[str, asyncio.Lock] = {} async def register_server(self, name: str, transport): """注册一个 MCP 服务器""" self._locks[name] = asyncio.Lock() session = ClientSession(transport) async with session: await session.initialize() self._sessions[name] = session async def call_tool(self, server_name: str, tool_name: str, arguments: dict): """在指定服务器上调用工具（带锁保护）""" async with self._locks.get(server_name, asyncio.Lock()): session = self._sessions.get(server_name) if not session: raise ConnectionError(f"服务器 {server_name} 未注册") return await session.call_tool(tool_name, arguments) async def discover_tools(self) -> Dict[str, list]: """发现所有注册服务器的可用工具""" result = {} for name, session in self._sessions.items(): async with self._locks[name]: tools = await session.list_tools() result[name] = tools return result async def close_all(self): """关闭所有连接""" for name, session in self._sessions.items(): await session.close() self._sessions.clear()---
六、MCP 协议的演进趋势
站在 2026 年 7 月的节点回望，MCP 协议已经经历了近两年的迭代，呈现出几个明确的演进方向：
1.A2A 协议的融合：Google 提出的 Agent-to-Agent 协议正在与 MCP 形成互补——MCP 解决"人→工具"的连接，A2A 解决"Agent→Agent"的协作。两者正在走向融合标准。
2.流式响应标准化：MCP 正在推进对 SSE 流式工具调用的原生支持，避免当前"全量返回后再推送"的延迟问题。
3.安全审计体系：随着 MCP 工具市场（MCP Hub）的爆发式增长，Skill 安全审计、依赖扫描、沙箱执行等安全机制正在成为协议规范的一部分。
4.边缘计算适配：轻量级 MCP 运行时正在被设计用于边缘设备，支持在资源受限的环境中运行 MCP 服务器。
---
结语
MCP 协议的核心设计哲学是"最小约定，最大自由"——它不做任何假设，不限制任何能力，只是定义了消息应该长什么样、怎么传输、何时建立连接。正是这种极简的克制，让它成为了 AI Agent 生态中不可或缺的基础设施。
理解 MCP 的底层原理，不只是为了会用某个 SDK，而是为了在面对复杂生产环境时，能够做出正确的架构决策。当你需要优化工具调用延迟时，你会想起 stdio vs SSE 的取舍；当你设计多 Agent 协作系统时，你会思考连接池和负载均衡；当你面对安全问题，你会回到传输层和握手阶段的防护设计。
MCP 不是魔法，是工程。掌握它的底层原理，你就能在 AI Agent 的浪潮中，从"使用者"成长为"构建者"。
---
本文封面图来源于 Unsplash。
