MCP四層架構(gòu)解析與JSON-RPC 2.0生產(chǎn)級Server實現(xiàn)指南

深入解析MCP四層架構(gòu)與JSON-RPC 2.0生產(chǎn)級實現(xiàn)
想讓你的AI Agent真正連接外部世界,而不是只會聊天?Model Context Protocol(MCP)就是那把鑰匙。但網(wǎng)上資料要么太淺,要么太散。今天咱們就拆解清楚:MCP的四層架構(gòu)到底怎么設計,以及如何用JSON-RPC 2.0寫出生產(chǎn)級的Server。
MCP不只是協(xié)議,是四層樂高積木
很多人把MCP理解成簡單的“AI調(diào)用工具的協(xié)議”,這低估了它的設計野心。MCP的核心是四層分層架構(gòu),每一層解決一個明確問題,讓開發(fā)者可以像搭樂高一樣構(gòu)建能力。
- 傳輸層(Transport Layer):這是最底層,負責“怎么連”。它定義了客戶端(比如你的AI應用)和服務器(提供工具或數(shù)據(jù)的服務)之間如何建立連接和交換原始字節(jié)。MCP支持多種傳輸方式,比如標準的輸入輸出(stdio)和HTTP with Server-Sent Events(SSE)。技術(shù)價值在于解耦——你的Server邏輯完全不用關(guān)心連接是來自本地進程還是遠程網(wǎng)絡,換傳輸層就像換插座一樣簡單。
- 消息層(Message Layer):這一層定義了“說什么話”。它建立在JSON-RPC 2.0之上,規(guī)定了所有通信都必須是結(jié)構(gòu)化的JSON-RPC消息。請求(Request)、響應(Response)、通知(Notification)三種消息類型,構(gòu)成了清晰、無歧義的對話規(guī)則。技術(shù)價值是標準化和可預測性,任何遵循此層的組件都能無縫對話。
- 能力層(Capability Layer):這是MCP的“功能菜單”。服務器在這里聲明自己能提供什么:是提供工具(Tools)讓AI執(zhí)行操作(如發(fā)郵件、查數(shù)據(jù)庫),還是提供資源(Resources)讓AI讀取上下文(如文件內(nèi)容、API文檔),或是提供提示模板(Prompts)??蛻舳藙t聲明自己需要什么。技術(shù)價值是動態(tài)發(fā)現(xiàn)和協(xié)商,連接建立后雙方就知道彼此的能力邊界。
- 語義層(Semantic Layer):最頂層,也是最體現(xiàn)前瞻性的一層。它關(guān)注“什么意思”。比如,如何用JSON Schema精確描述一個工具的輸入輸出參數(shù)?如何讓AI理解一個“資源”代表的是用戶文檔還是系統(tǒng)日志?技術(shù)價值是讓交互從“能調(diào)用”升級到“能理解”,為更復雜的Agent協(xié)作打下基礎(chǔ)。
這四層分離的設計,讓MCP既靈活又強大。你可以只實現(xiàn)傳輸層和消息層,做一個最簡Server;也可以完整實現(xiàn)四層,構(gòu)建一個功能豐富、語義清晰的智能服務。
用JSON-RPC 2.0打造生產(chǎn)級Server:穩(wěn)定高于一切
理解了架構(gòu),我們來點實在的。如何基于JSON-RPC 2.0寫出一個能抗住生產(chǎn)環(huán)境壓力的MCP Server?關(guān)鍵不在于實現(xiàn)所有RPC方法,而在于處理好穩(wěn)定性、錯誤處理和流式響應。
JSON-RPC 2.0的核心很簡單:一個請求包含“jsonrpc”: “2.0”、“method”、“params”和“id”。但生產(chǎn)環(huán)境中,魔鬼在細節(jié)里。
一個健壯的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. 構(gòu)造標準成功響應
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
})

# 具體方法實現(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:
# 返回工具列表,每個工具必須有清晰的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”, {})
# 這里執(zhí)行實際工具邏輯,并返回內(nèi)容列表
if tool_name == “get_weather”:
city = arguments.get(“city”, “北京”)
weather_data = await self._fetch_weather_api(city) # 假設的異步API調(diào)用
return {
“content”: [
{“type”: “text”, “text”: f“{city}當前天氣:{weather_data}”}
]
}
else:
raise ValueError(f“Unknown tool: {tool_name}”)生產(chǎn)級要點:
- 嚴格的輸入校驗:絕不信任客戶端輸入,對每條消息進行格式和參數(shù)校驗。
- 全面的錯誤處理:使用標準的JSON-RPC錯誤碼(-32700解析錯誤,-32600無效請求等),并捕獲所有未處理異常,返回友好的錯誤信息,而不是讓進程崩潰。
- 清晰的Schema定義:工具的
inputSchema是AI理解如何調(diào)用的關(guān)鍵,必須詳細、準確。 - 異步非阻塞:使用
asyncio,確保一個慢操作(如網(wǎng)絡請求)不會阻塞整個Server。
生態(tài)拐點:首個支持流式Tool Response的開源Server
就在最近,開源社區(qū)出現(xiàn)了第一個完整支持流式Tool Response的MCP Server實現(xiàn)。為什么說這是一個里程碑式的拐點?
在之前,一個工具調(diào)用(比如“分析這份100頁的PDF并生成報告”)的流程是:AI發(fā)送請求 -> Server執(zhí)行(可能需要幾分鐘)-> 執(zhí)行完畢后一次性返回完整結(jié)果。這導致兩個問題:1) 用戶等待時間長,體驗差;2) 容易超時。
而流式Tool Response允許Server在工具執(zhí)行過程中,像SSE一樣分塊、逐步地返回中間結(jié)果或進度。例如,分析PDF時,可以每分析完一章就返回一章的摘要。
技術(shù)實現(xiàn)上,這依賴于MCP消息層對JSON-RPC通知(Notification)的運用。Server可以在處理一個tools/call請求的同時,向客戶端發(fā)送多個notifications/tools/progress通知,實時更新進度或部分內(nèi)容。
這標志著Server生態(tài)進入發(fā)展拐點,因為:
- 體驗升級:AI Agent可以實時反饋工作進度,從“黑盒等待”變?yōu)椤巴该鬟M程”,用戶信任度大幅提升。
- 場景解鎖:使得需要長時間運行的工具(代碼編譯、大數(shù)據(jù)分析、長文檔處理)變得實用。
- 架構(gòu)演進:推動Server開發(fā)者必須考慮異步、流式的架構(gòu)設計,整體提升生態(tài)的技術(shù)水位。
下一步行動:從理解到動手
理論讀千遍,不如動手寫一遍。你的下一步可以很明確:
- 本地跑起來:用上面的代碼骨架,在你的電腦上用stdio傳輸方式啟動一個最簡Server,用MCP客戶端工具(如Claude Desktop的開發(fā)者模式)連接它,調(diào)用一下
get_weather工具。 - 挑戰(zhàn)流式響應:在你的Server中,嘗試為
get_weather工具添加一個“模擬查詢延遲”,并使用notifications/tools/progress每秒返回一次“正在連接氣象衛(wèi)星…”的進度通知。 - 發(fā)布你的第一個工具:想一個能解決你實際小問題的工具(比如“總結(jié)網(wǎng)頁內(nèi)容”、“轉(zhuǎn)換文件格式”),用MCP實現(xiàn)它,并開源到GitHub。生態(tài)的繁榮,始于每一個具體的工具。
MCP的四層架構(gòu)給了我們清晰的藍圖,JSON-RPC 2.0提供了穩(wěn)定的通信基石,而流式響應則點燃了新的可能性。現(xiàn)在,輪到你來建造了。