深入解析MCP四層架構(gòu)與JSON-RPC 2.0生產(chǎn)級(jí)實(shí)現(xiàn)

想讓你的AI Agent真正連接外部世界，而不是只會(huì)聊天？Model Context Protocol（MCP）就是那把鑰匙。但網(wǎng)上資料要么太淺，要么太散。今天咱們就拆解清楚：MCP的四層架構(gòu)到底怎么設(shè)計(jì)，以及如何用JSON-RPC 2.0寫(xiě)出生產(chǎn)級(jí)的Server。

MCP不只是協(xié)議，是四層樂(lè)高積木

很多人把MCP理解成簡(jiǎn)單的“AI調(diào)用工具的協(xié)議”，這低估了它的設(shè)計(jì)野心。MCP的核心是四層分層架構(gòu)，每一層解決一個(gè)明確問(wèn)題，讓開(kāi)發(fā)者可以像搭樂(lè)高一樣構(gòu)建能力。

1. 傳輸層（Transport Layer）：這是最底層，負(fù)責(zé)“怎么連”。它定義了客戶(hù)端（比如你的AI應(yīng)用）和服務(wù)器（提供工具或數(shù)據(jù)的服務(wù)）之間如何建立連接和交換原始字節(jié)。MCP支持多種傳輸方式，比如標(biāo)準(zhǔn)的輸入輸出（stdio）和HTTP with Server-Sent Events（SSE）。技術(shù)價(jià)值在于解耦——你的Server邏輯完全不用關(guān)心連接是來(lái)自本地進(jìn)程還是遠(yuǎn)程網(wǎng)絡(luò)，換傳輸層就像換插座一樣簡(jiǎn)單。
2. 消息層（Message Layer）：這一層定義了“說(shuō)什么話”。它建立在JSON-RPC 2.0之上，規(guī)定了所有通信都必須是結(jié)構(gòu)化的JSON-RPC消息。請(qǐng)求（Request）、響應(yīng)（Response）、通知（Notification）三種消息類(lèi)型，構(gòu)成了清晰、無(wú)歧義的對(duì)話規(guī)則。技術(shù)價(jià)值是標(biāo)準(zhǔn)化和可預(yù)測(cè)性，任何遵循此層的組件都能無(wú)縫對(duì)話。
3. 能力層（Capability Layer）：這是MCP的“功能菜單”。服務(wù)器在這里聲明自己能提供什么：是提供工具（Tools）讓AI執(zhí)行操作（如發(fā)郵件、查數(shù)據(jù)庫(kù)），還是提供資源（Resources）讓AI讀取上下文（如文件內(nèi)容、API文檔），或是提供提示模板（Prompts）。客戶(hù)端則聲明自己需要什么。技術(shù)價(jià)值是動(dòng)態(tài)發(fā)現(xiàn)和協(xié)商，連接建立后雙方就知道彼此的能力邊界。
4. 語(yǔ)義層（Semantic Layer）：最頂層，也是最體現(xiàn)前瞻性的一層。它關(guān)注“什么意思”。比如，如何用JSON Schema精確描述一個(gè)工具的輸入輸出參數(shù)？如何讓AI理解一個(gè)“資源”代表的是用戶(hù)文檔還是系統(tǒng)日志？技術(shù)價(jià)值是讓交互從“能調(diào)用”升級(jí)到“能理解”，為更復(fù)雜的Agent協(xié)作打下基礎(chǔ)。

這四層分離的設(shè)計(jì)，讓MCP既靈活又強(qiáng)大。你可以只實(shí)現(xiàn)傳輸層和消息層，做一個(gè)最簡(jiǎn)Server；也可以完整實(shí)現(xiàn)四層，構(gòu)建一個(gè)功能豐富、語(yǔ)義清晰的智能服務(wù)。

用JSON-RPC 2.0打造生產(chǎn)級(jí)Server：穩(wěn)定高于一切

理解了架構(gòu)，我們來(lái)點(diǎn)實(shí)在的。如何基于JSON-RPC 2.0寫(xiě)出一個(gè)能抗住生產(chǎn)環(huán)境壓力的MCP Server？關(guān)鍵不在于實(shí)現(xiàn)所有RPC方法，而在于處理好穩(wěn)定性、錯(cuò)誤處理和流式響應(yīng)。

JSON-RPC 2.0的核心很簡(jiǎn)單：一個(gè)請(qǐng)求包含“jsonrpc”: “2.0”、“method”、“params”和“id”。但生產(chǎn)環(huán)境中，魔鬼在細(xì)節(jié)里。

一個(gè)健壯的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. 嚴(yán)格校驗(yàn)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. 構(gòu)造標(biāo)準(zhǔn)成功響應(yīng)
            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:
            # 生產(chǎn)環(huán)境必須捕獲所有異常，避免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.xmhny.cn/usr/uploads/covers/cover_mcp_20260531_081513.jpg)

    # 具體方法實(shí)現(xiàn)（示例）
    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:
        # 返回工具列表，每個(gè)工具必須有清晰的JSON Schema描述
        return {
            “tools”: [
                {
                    “name”: “get_weather”,
                    “description”: “獲取指定城市的當(dāng)前天氣”,
                    “inputSchema”: {
                        “type”: “object”,
                        “properties”: {
                            “city”: {“type”: “string”, “description”: “城市名稱(chēng)，如‘北京’”}
                        },
                        “required”: [“city”]
                    }
                }
            ]
        }

    async def _handle_tools_call(self, params: Dict) -> Dict:
        tool_name = params.get(“name”)
        arguments = params.get(“arguments”, {})
        # 這里執(zhí)行實(shí)際工具邏輯，并返回內(nèi)容列表
        if tool_name == “get_weather”:
            city = arguments.get(“city”, “北京”)
            weather_data = await self._fetch_weather_api(city) # 假設(shè)的異步API調(diào)用
            return {
                “content”: [
                    {“type”: “text”, “text”: f“{city}當(dāng)前天氣：{weather_data}”}
                ]
            }
        else:
            raise ValueError(f“Unknown tool: {tool_name}”)

生產(chǎn)級(jí)要點(diǎn)：

• 嚴(yán)格的輸入校驗(yàn)：絕不信任客戶(hù)端輸入，對(duì)每條消息進(jìn)行格式和參數(shù)校驗(yàn)。
• 全面的錯(cuò)誤處理：使用標(biāo)準(zhǔn)的JSON-RPC錯(cuò)誤碼（-32700解析錯(cuò)誤，-32600無(wú)效請(qǐng)求等），并捕獲所有未處理異常，返回友好的錯(cuò)誤信息，而不是讓進(jìn)程崩潰。
• 清晰的Schema定義：工具的inputSchema是AI理解如何調(diào)用的關(guān)鍵，必須詳細(xì)、準(zhǔn)確。
• 異步非阻塞：使用asyncio，確保一個(gè)慢操作（如網(wǎng)絡(luò)請(qǐng)求）不會(huì)阻塞整個(gè)Server。

生態(tài)拐點(diǎn)：首個(gè)支持流式Tool Response的開(kāi)源Server

就在最近，開(kāi)源社區(qū)出現(xiàn)了第一個(gè)完整支持流式Tool Response的MCP Server實(shí)現(xiàn)。為什么說(shuō)這是一個(gè)里程碑式的拐點(diǎn)？

在之前，一個(gè)工具調(diào)用（比如“分析這份100頁(yè)的PDF并生成報(bào)告”）的流程是：AI發(fā)送請(qǐng)求 -> Server執(zhí)行（可能需要幾分鐘）-> 執(zhí)行完畢后一次性返回完整結(jié)果。這導(dǎo)致兩個(gè)問(wèn)題：1) 用戶(hù)等待時(shí)間長(zhǎng)，體驗(yàn)差；2) 容易超時(shí)。

而流式Tool Response允許Server在工具執(zhí)行過(guò)程中，像SSE一樣分塊、逐步地返回中間結(jié)果或進(jìn)度。例如，分析PDF時(shí)，可以每分析完一章就返回一章的摘要。

技術(shù)實(shí)現(xiàn)上，這依賴(lài)于MCP消息層對(duì)JSON-RPC通知（Notification）的運(yùn)用。Server可以在處理一個(gè)tools/call請(qǐng)求的同時(shí)，向客戶(hù)端發(fā)送多個(gè)notifications/tools/progress通知，實(shí)時(shí)更新進(jìn)度或部分內(nèi)容。

這標(biāo)志著Server生態(tài)進(jìn)入發(fā)展拐點(diǎn)，因?yàn)?/strong>：

1. 體驗(yàn)升級(jí)：AI Agent可以實(shí)時(shí)反饋工作進(jìn)度，從“黑盒等待”變?yōu)椤巴该鬟M(jìn)程”，用戶(hù)信任度大幅提升。
2. 場(chǎng)景解鎖：使得需要長(zhǎng)時(shí)間運(yùn)行的工具（代碼編譯、大數(shù)據(jù)分析、長(zhǎng)文檔處理）變得實(shí)用。
3. 架構(gòu)演進(jìn)：推動(dòng)Server開(kāi)發(fā)者必須考慮異步、流式的架構(gòu)設(shè)計(jì)，整體提升生態(tài)的技術(shù)水位。

下一步行動(dòng)：從理解到動(dòng)手

理論讀千遍，不如動(dòng)手寫(xiě)一遍。你的下一步可以很明確：

1. 本地跑起來(lái)：用上面的代碼骨架，在你的電腦上用stdio傳輸方式啟動(dòng)一個(gè)最簡(jiǎn)Server，用MCP客戶(hù)端工具（如Claude Desktop的開(kāi)發(fā)者模式）連接它，調(diào)用一下get_weather工具。
2. 挑戰(zhàn)流式響應(yīng)：在你的Server中，嘗試為get_weather工具添加一個(gè)“模擬查詢(xún)延遲”，并使用notifications/tools/progress每秒返回一次“正在連接氣象衛(wèi)星…”的進(jìn)度通知。
3. 發(fā)布你的第一個(gè)工具：想一個(gè)能解決你實(shí)際小問(wèn)題的工具（比如“總結(jié)網(wǎng)頁(yè)內(nèi)容”、“轉(zhuǎn)換文件格式”），用MCP實(shí)現(xiàn)它，并開(kāi)源到GitHub。生態(tài)的繁榮，始于每一個(gè)具體的工具。

MCP的四層架構(gòu)給了我們清晰的藍(lán)圖，JSON-RPC 2.0提供了穩(wěn)定的通信基石，而流式響應(yīng)則點(diǎn)燃了新的可能性。現(xiàn)在，輪到你來(lái)建造了。