深入解析MCP四层架构与JSON-RPC 2.0生产级实现

想让你的AI Agent真正连接外部世界，而不是只会聊天？Model Context Protocol（MCP）就是那把钥匙。但网上资料要么太浅，要么太散。今天咱们就拆解清楚：MCP的四层架构到底怎么设计，以及如何用JSON-RPC 2.0写出生产级的Server。

MCP不只是协议，是四层乐高积木

很多人把MCP理解成简单的“AI调用工具的协议”，这低估了它的设计野心。MCP的核心是四层分层架构，每一层解决一个明确问题，让开发者可以像搭乐高一样构建能力。

1. 传输层（Transport Layer）：这是最底层，负责“怎么连”。它定义了客户端（比如你的AI应用）和服务器（提供工具或数据的服务）之间如何建立连接和交换原始字节。MCP支持多种传输方式，比如标准的输入输出（stdio）和HTTP with Server-Sent Events（SSE）。技术价值在于解耦——你的Server逻辑完全不用关心连接是来自本地进程还是远程网络，换传输层就像换插座一样简单。
2. 消息层（Message Layer）：这一层定义了“说什么话”。它建立在JSON-RPC 2.0之上，规定了所有通信都必须是结构化的JSON-RPC消息。请求（Request）、响应（Response）、通知（Notification）三种消息类型，构成了清晰、无歧义的对话规则。技术价值是标准化和可预测性，任何遵循此层的组件都能无缝对话。
3. 能力层（Capability Layer）：这是MCP的“功能菜单”。服务器在这里声明自己能提供什么：是提供工具（Tools）让AI执行操作（如发邮件、查数据库），还是提供资源（Resources）让AI读取上下文（如文件内容、API文档），或是提供提示模板（Prompts）。客户端则声明自己需要什么。技术价值是动态发现和协商，连接建立后双方就知道彼此的能力边界。
4. 语义层（Semantic Layer）：最顶层，也是最体现前瞻性的一层。它关注“什么意思”。比如，如何用JSON Schema精确描述一个工具的输入输出参数？如何让AI理解一个“资源”代表的是用户文档还是系统日志？技术价值是让交互从“能调用”升级到“能理解”，为更复杂的Agent协作打下基础。

这四层分离的设计，让MCP既灵活又强大。你可以只实现传输层和消息层，做一个最简Server；也可以完整实现四层，构建一个功能丰富、语义清晰的智能服务。

用JSON-RPC 2.0打造生产级Server：稳定高于一切

理解了架构，我们来点实在的。如何基于JSON-RPC 2.0写出一个能抗住生产环境压力的MCP Server？关键不在于实现所有RPC方法，而在于处理好稳定性、错误处理和流式响应。

JSON-RPC 2.0的核心很简单：一个请求包含“jsonrpc”: “2.0”、“method”、“params”和“id”。但生产环境中，魔鬼在细节里。

一个健壮的Server骨架（Python示例）：

import asyncio
import json
from typing import Any, Dict, Optional

class ProductionMCPServer:
    def __init__(self):
        self.capabilities = {
            "tools": {"listChanged": False},
            "resources": {"subscribe": False, "listChanged": False}
        }

    async def handle_message(self, raw_message: str) -> Optional[str]:
        """处理单条JSON-RPC消息的核心逻辑"""
        try:
            message = json.loads(raw_message)
            # 1. 严格校验JSON-RPC 2.0格式
            if not all(k in message for k in (“jsonrpc”, “method”, “id”)):
                return self._error_response(None, -32600, “Invalid Request”)

            method = message[“method”]
            params = message.get(“params”, {})
            msg_id = message[“id”]

            # 2. 路由到具体处理方法
            if method == “initialize”:
                result = await self._handle_initialize(params)
            elif method == “tools/list”:
                result = await self._handle_tools_list()
            elif method == “tools/call”:
                result = await self._handle_tools_call(params)
            else:
                return self._error_response(msg_id, -32601, “Method not found”)

            # 3. 构造标准成功响应
            return json.dumps({
                “jsonrpc”: “2.0”,
                “result”: result,
                “id”: msg_id
            })

        except json.JSONDecodeError:
            return self._error_response(None, -32700, “Parse error”)
        except Exception as e:
            # 生产环境必须捕获所有异常，避免Server崩溃
            return self._error_response(message.get(‘id’), -32000, f“Server error: {str(e)}”)

    def _error_response(self, id: Any, code: int, message: str) -> str:
        return json.dumps({
            “jsonrpc”: “2.0”,
            “error”: {“code”: code, “message”: message},
            “id”: id
        })


![配图](https://yitb.com/usr/uploads/covers/cover_mcp_20260531_081513.jpg)

    # 具体方法实现（示例）
    async def _handle_initialize(self, params: Dict) -> Dict:
        return {
            “protocolVersion”: “2025-03-26”,
            “capabilities”: self.capabilities,
            “serverInfo”: {“name”: “ProductionServer”, “version”: “1.0.0”}
        }

    async def _handle_tools_list(self) -> Dict:
        # 返回工具列表，每个工具必须有清晰的JSON Schema描述
        return {
            “tools”: [
                {
                    “name”: “get_weather”,
                    “description”: “获取指定城市的当前天气”,
                    “inputSchema”: {
                        “type”: “object”,
                        “properties”: {
                            “city”: {“type”: “string”, “description”: “城市名称，如‘北京’”}
                        },
                        “required”: [“city”]
                    }
                }
            ]
        }

    async def _handle_tools_call(self, params: Dict) -> Dict:
        tool_name = params.get(“name”)
        arguments = params.get(“arguments”, {})
        # 这里执行实际工具逻辑，并返回内容列表
        if tool_name == “get_weather”:
            city = arguments.get(“city”, “北京”)
            weather_data = await self._fetch_weather_api(city) # 假设的异步API调用
            return {
                “content”: [
                    {“type”: “text”, “text”: f“{city}当前天气：{weather_data}”}
                ]
            }
        else:
            raise ValueError(f“Unknown tool: {tool_name}”)

生产级要点：

• 严格的输入校验：绝不信任客户端输入，对每条消息进行格式和参数校验。
• 全面的错误处理：使用标准的JSON-RPC错误码（-32700解析错误，-32600无效请求等），并捕获所有未处理异常，返回友好的错误信息，而不是让进程崩溃。
• 清晰的Schema定义：工具的inputSchema是AI理解如何调用的关键，必须详细、准确。
• 异步非阻塞：使用asyncio，确保一个慢操作（如网络请求）不会阻塞整个Server。

生态拐点：首个支持流式Tool Response的开源Server

就在最近，开源社区出现了第一个完整支持流式Tool Response的MCP Server实现。为什么说这是一个里程碑式的拐点？

在之前，一个工具调用（比如“分析这份100页的PDF并生成报告”）的流程是：AI发送请求 -> Server执行（可能需要几分钟）-> 执行完毕后一次性返回完整结果。这导致两个问题：1) 用户等待时间长，体验差；2) 容易超时。

而流式Tool Response允许Server在工具执行过程中，像SSE一样分块、逐步地返回中间结果或进度。例如，分析PDF时，可以每分析完一章就返回一章的摘要。

技术实现上，这依赖于MCP消息层对JSON-RPC通知（Notification）的运用。Server可以在处理一个tools/call请求的同时，向客户端发送多个notifications/tools/progress通知，实时更新进度或部分内容。

这标志着Server生态进入发展拐点，因为：

1. 体验升级：AI Agent可以实时反馈工作进度，从“黑盒等待”变为“透明进程”，用户信任度大幅提升。
2. 场景解锁：使得需要长时间运行的工具（代码编译、大数据分析、长文档处理）变得实用。
3. 架构演进：推动Server开发者必须考虑异步、流式的架构设计，整体提升生态的技术水位。

下一步行动：从理解到动手

理论读千遍，不如动手写一遍。你的下一步可以很明确：

1. 本地跑起来：用上面的代码骨架，在你的电脑上用stdio传输方式启动一个最简Server，用MCP客户端工具（如Claude Desktop的开发者模式）连接它，调用一下get_weather工具。
2. 挑战流式响应：在你的Server中，尝试为get_weather工具添加一个“模拟查询延迟”，并使用notifications/tools/progress每秒返回一次“正在连接气象卫星…”的进度通知。
3. 发布你的第一个工具：想一个能解决你实际小问题的工具（比如“总结网页内容”、“转换文件格式”），用MCP实现它，并开源到GitHub。生态的繁荣，始于每一个具体的工具。

MCP的四层架构给了我们清晰的蓝图，JSON-RPC 2.0提供了稳定的通信基石，而流式响应则点燃了新的可能性。现在，轮到你来建造了。