深入解析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
        })


![配圖](http://www.nhjb.com.cn/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提供了穩定的通信基石，而流式響應則點燃了新的可能性。現在，輪到你來建造了。