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。