当前位置：首页 > news >正文

MCP协议底层原理深度剖析：从JSON-RPC 2.0到多传输层实现

news 2026/7/27 23:35:13

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()

---