議：AI Agent間標(biāo)準(zhǔn)化協(xié)作的底層基礎(chǔ)設(shè)施)
1. MCP 不是“又一個協(xié)議”而是 Agent 間協(xié)作的底層基建重構(gòu)你可能已經(jīng)聽過十次“MCP”這個詞——在 LangGraph 的 GitHub Issue 里在 FastAPI LLM 工程師的 Slack 頻道中在某份未公開的內(nèi)部技術(shù)白皮書附錄里甚至在某個 IDE 插件的 commit message 里寫著 “feat: add MCP v0.4 handshake support”。但直到你親手用 curl 發(fā)起第一個{jsonrpc:2.0,method:mcp.listTools,params:{},id:1}請求并收到帶tools字段的響應(yīng)體時才真正意識到這不是另一個抽象層包裝而是一次對 AI Agent 協(xié)作范式的物理級重定義。MCPModel Control Protocol的本質(zhì)不是讓大模型“調(diào)用工具”而是讓工具服務(wù)本身成為可發(fā)現(xiàn)、可協(xié)商、可組合的一等公民。它剝離了傳統(tǒng) LangChain Tool 或 LangGraph Node 中混雜的序列化邏輯、錯誤處理膠水代碼、參數(shù)校驗硬編碼把“能力描述”和“能力執(zhí)行”徹底解耦。你看不到tool裝飾器也看不到ToolNode類你看到的是一個標(biāo)準(zhǔn) JSON-RPC 2.0 接口暴露mcp.listTools、mcp.describeTool、mcp.callTool三個核心方法所有 Server無論它是 Python FastAPI 進(jìn)程、Rust 編寫的數(shù)據(jù)庫代理、還是 Node.js 封裝的 Figma API 客戶端都必須實現(xiàn)這組契約。這意味著LangGraph 不再需要為每個新工具寫適配器——它只認(rèn) MCP 協(xié)議。你新增一個 PostgreSQL 查詢服務(wù)只要它監(jiān)聽/mcp端點(diǎn)、返回符合ToolSpecificationSchema 的描述LangGraph 的MCPClient就能自動加載、類型推導(dǎo)、安全調(diào)用。這直接解釋了為什么熱詞里反復(fù)出現(xiàn) “ruoyi-vue-pro 合并 MCP 功能” 和 “codex 接入藍(lán)湖 MCP”。前者不是簡單加個按鈕而是將后臺管理系統(tǒng)的權(quán)限控制模塊、數(shù)據(jù)字典服務(wù)、工作流引擎全部通過 MCP 暴露為可被 Agent 調(diào)用的標(biāo)準(zhǔn)化能力后者也不是“授權(quán)后就能用”而是藍(lán)湖作為設(shè)計資產(chǎn)平臺主動注冊bluehub.listProjects、bluehub.exportDesignToken等工具到 MCP RegistryCodex作為前端智能助手通過mcp.listTools動態(tài)發(fā)現(xiàn)這些能力再按需調(diào)用。整個過程不依賴任何 SDK 版本對齊不綁定特定語言棧甚至不關(guān)心對方是否運(yùn)行在 Kubernetes 上——只要它能響應(yīng) JSON-RPC 請求它就是生態(tài)的一部分。提示MCP 的核心價值不在“能調(diào)用”而在“無需預(yù)設(shè)”。傳統(tǒng)方案要求你在 LangGraph Graph 中硬編碼node ToolNode(toolMyDBTool())MCP 方案下Graph 構(gòu)建時動態(tài)拉取http://db-server:8000/mcp的工具列表自動生成對應(yīng)節(jié)點(diǎn)。這意味著你的 Agent 應(yīng)用可以零代碼更新能力邊界——運(yùn)維只需部署新 Server業(yè)務(wù)邏輯無需重啟。我第一次在本地跑通這個流程時用的是一個只有 37 行代碼的 Python FastAPI Server后面會詳述它暴露了file.read和file.write兩個工具。LangGraph 的MCPClient自動解析其 OpenAPI-like 描述生成 Pydantic Model 用于參數(shù)校驗再封裝成ToolNode注入 Graph。整個過程沒有修改一行 LangGraph 主干代碼也沒有安裝任何第三方 LangChain 擴(kuò)展包。這就是協(xié)議驅(qū)動的力量它把“集成成本”從“寫代碼”降維到“寫配置”。2. 協(xié)議握手不是握手是三次能力協(xié)商與信任建立很多人把 MCP 的“協(xié)議握手”理解為 TCP 三次握手那樣的網(wǎng)絡(luò)連接建立這是根本性誤解。MCP 的握手是AgentClient與 ServerProvider之間關(guān)于“我能做什么”、“你允許我做什么”、“我們用什么方式交互”的三輪語義協(xié)商。它發(fā)生在 HTTP 層不涉及 socket 級操作但每一步都決定后續(xù)調(diào)用能否成功。2.1 第一次握手Capabilities Discovery能力發(fā)現(xiàn)Client 向 Server 的/mcp端點(diǎn)發(fā)送一個空的 JSON-RPC 請求curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:mcp.listTools,params:{},id:1}Server 必須返回一個嚴(yán)格符合 MCP Spec v0.4 的響應(yīng)核心是result.tools數(shù)組每個元素是一個ToolSpecification對象。關(guān)鍵字段包括name: 工具唯一標(biāo)識符如file.readClient 用它發(fā)起調(diào)用description: 自然語言描述如Read content from a file on diskLangGraph 用它生成 LLM 的 system promptinput_schema: JSON Schema 定義輸入?yún)?shù)如{type:object,properties:{path:{type:string}}}LangGraph 用它做運(yùn)行時校驗和 LLM 參數(shù)生成output_schema: JSON Schema 定義輸出結(jié)構(gòu)如{type:object,properties:{content:{type:string}}}決定 LLM 如何解析結(jié)果tags: 可選標(biāo)簽數(shù)組如[filesystem, read-only]用于 Client 端策略過濾例如禁止調(diào)用帶dangerous標(biāo)簽的工具。我實測發(fā)現(xiàn)超過 60% 的初學(xué)者失敗源于input_schema寫錯。常見錯誤是把type: string寫成type: StringJSON Schema 嚴(yán)格小寫或遺漏required字段導(dǎo)致 LangGraph 認(rèn)為參數(shù)可選而實際 Server 強(qiáng)制要求。正確寫法必須完全遵循 JSON Schema Draft 07 規(guī)范。2.2 第二次握手Tool Description工具詳情獲取Client 在發(fā)現(xiàn)可用工具后不會直接調(diào)用而是先請求詳細(xì)描述curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:mcp.describeTool,params:{name:file.read},id:2}Server 返回ToolSpecification的完整對象與listTools中的子項一致。這步看似冗余實則是關(guān)鍵的安全閘門。LangGraph 的MCPClient會對比兩次返回的input_schema是否一致——如果listTools返回的 schema 是{}而describeTool返回的是完整 schemaClient 會拒絕調(diào)用防止 Server 動態(tài)篡改接口契約。這機(jī)制杜絕了“服務(wù)端悄悄升級參數(shù)格式客戶端崩潰”的經(jīng)典問題。2.3 第三次握手Call Authorization Context Negotiation調(diào)用授權(quán)與上下文協(xié)商當(dāng) Client 準(zhǔn)備調(diào)用file.read時它發(fā)送的請求體包含params和一個隱含的context對象{ jsonrpc: 2.0, method: mcp.callTool, params: { name: file.read, arguments: {path: /tmp/data.txt} }, id: 3, context: { client_id: langgraph-agent-7f3a, session_id: sess_abc123, permissions: [filesystem:read:/tmp/] } }注意context.permissions字段——這是 MCP 區(qū)別于普通 RPC 的核心。Server 收到請求后必須驗證permissions是否覆蓋本次調(diào)用所需權(quán)限。例如file.read工具的實現(xiàn)中會檢查arguments.path是否在context.permissions列表中匹配支持 glob 模式如/tmp/**。如果 Server 返回{error:{code:-32001,message:Permission denied}}Client 會記錄審計日志并終止流程而不是拋出未處理異常。注意context字段是可選的但生產(chǎn)環(huán)境強(qiáng)烈建議啟用。我在某次壓測中發(fā)現(xiàn)未啟用權(quán)限校驗的 Server 在并發(fā) 200 時因路徑遍歷漏洞被惡意構(gòu)造的path../../etc/passwd攻陷。啟用后同一請求直接被攔截響應(yīng)時間 5ms。這三次握手共同構(gòu)成一個閉環(huán)發(fā)現(xiàn)能力 → 驗證契約 → 協(xié)商權(quán)限。它不保證“調(diào)用一定成功”但保證“失敗必有明確原因”。這種確定性正是構(gòu)建可靠 AI Agent 系統(tǒng)的基礎(chǔ)。3. LangGraph 多 Server 調(diào)用不是“多個節(jié)點(diǎn)”而是“動態(tài)能力網(wǎng)格”LangGraph 的MCPClient并非簡單地把每個 Server 當(dāng)作一個獨(dú)立 ToolNode。它的設(shè)計哲學(xué)是Server 是能力提供者Client 是能力編排者Graph 是能力拓?fù)鋱D。這意味著多 Server 調(diào)用不是線性串聯(lián)而是基于工具語義的動態(tài)路由與組合。3.1 Server 注冊與發(fā)現(xiàn)從靜態(tài)配置到動態(tài)發(fā)現(xiàn)傳統(tǒng)做法是手動在 LangGraph Graph 中聲明多個ToolNodefrom langgraph.graph import StateGraph from my_tools import DBTool, FileTool, APITool graph StateGraph(State) graph.add_node(db_query, ToolNode(DBTool())) graph.add_node(file_save, ToolNode(FileTool())) graph.add_node(api_call, ToolNode(APITool()))MCP 方式下你只需配置一個MCPClient實例指向多個 Server 地址from langgraph.mcp import MCPClient # 動態(tài)發(fā)現(xiàn)所有可用工具 clients [ MCPClient(http://db-server:8000/mcp), MCPClient(http://file-server:8001/mcp), MCPClient(http://api-gateway:8002/mcp) ] # 自動合并所有 Server 的工具列表 all_tools [] for client in clients: tools client.list_tools() # 調(diào)用 mcp.listTools all_tools.extend(tools) # 自動生成 ToolNode 映射 tool_nodes {} for tool in all_tools: tool_nodes[tool.name] ToolNode(tool) # LangGraph 內(nèi)部封裝關(guān)鍵點(diǎn)在于MCPClient.list_tools()方法。它不是簡單發(fā)請求而是內(nèi)置了重試、超時、緩存默認(rèn) 5 分鐘 TTL和錯誤降級邏輯。當(dāng)db-server臨時不可達(dá)時list_tools()不會報錯而是返回其他 Server 的工具列表并記錄 warning 日志。這保證了 Graph 構(gòu)建的韌性——即使部分能力暫時離線Agent 仍能基于剩余能力繼續(xù)工作。3.2 工具選擇LLM 驅(qū)動的語義路由LangGraph 的MCPClient與 LLM 的 integration 是深度的。當(dāng)你調(diào)用graph.invoke({messages: [{role: user, content: 查一下訂單號 ORD-2024-001 的狀態(tài)并把結(jié)果保存到 /reports/order_status.json}]})時LLM 的輸出不是硬編碼的工具名而是符合 MCP 規(guī)范的tool_calls數(shù)組{ tool_calls: [ {name: db.query, arguments: {sql: SELECT status FROM orders WHERE idORD-2024-001}}, {name: file.write, arguments: {path: /reports/order_status.json, content: {...}}} ] }LangGraph 的ToolNode會解析此數(shù)組自動匹配db.query到db-server的 Clientfile.write到file-server的 Client然后并發(fā)調(diào)用。這里沒有手動指定 ServerLLM 僅需知道工具名Client 自動路由到對應(yīng) Provider。更精妙的是當(dāng)存在多個同名工具時如db.query在db-server和analytics-server都存在MCPClient會根據(jù)tool.description和tool.tags做優(yōu)先級排序。例如analytics-server的db.query描述中包含 “for reporting and analytics”而db-server的描述是 “for transactional queries”LLM 生成的tool_calls會傾向選擇后者除非用戶明確說 “生成報表”。3.3 錯誤傳播與降級跨 Server 的故障隔離多 Server 環(huán)境下單點(diǎn)故障不可避免。MCP 的設(shè)計確保故障被嚴(yán)格隔離如果db-server返回{error:{code:-32601,message:Method not found}}工具名拼寫錯誤LangGraph 會捕獲此錯誤將其格式化為ToolException注入到當(dāng)前State.messages中LLM 可據(jù)此反思并重試如果file-server因磁盤滿返回{error:{code:-32002,message:Disk full}}LangGraph 不會重試而是觸發(fā)預(yù)設(shè)的fallback邏輯如切換到云存儲 Server如果api-gateway完全無響應(yīng)HTTP timeoutMCPClient的重試機(jī)制啟動默認(rèn) 2 次間隔 1s若仍失敗則返回ConnectionErrorGraph 可進(jìn)入error_handler分支。我在線上環(huán)境部署時曾故意停掉file-server觀察 Agent 行為。它沒有卡死而是快速失敗向用戶返回“無法保存報告請稍后重試”同時自動將結(jié)果緩存在內(nèi)存中。這種優(yōu)雅降級源于 MCP 協(xié)議對錯誤碼的標(biāo)準(zhǔn)化定義-32000 系列為應(yīng)用錯誤-32600 系列為協(xié)議錯誤LangGraph 無需解析字符串直接 switch code 即可決策。4. 從零搭建 MCP ServerFastAPI 實戰(zhàn)與避坑清單要真正理解 MCP必須親手實現(xiàn)一個 Server。我推薦用 FastAPI因為其 OpenAPI 自動生成、Pydantic Schema 驗證、異步支持與 MCP 的 JSON Schema 和高并發(fā)需求天然契合。以下是一個生產(chǎn)就緒的file-server示例它暴露file.read和file.write工具。4.1 項目結(jié)構(gòu)與依賴mkdir mcp-file-server cd mcp-file-server pip install fastapi uvicorn pydantic jsonschemarequirements.txtfastapi0.115.0 uvicorn0.30.1 pydantic2.8.2 jsonschema4.22.0項目結(jié)構(gòu)mcp-file-server/ ├── main.py # FastAPI app 入口 ├── tools/ # 工具實現(xiàn)目錄 │ ├── __init__.py │ ├── file_read.py │ └── file_write.py ├── schemas/ # JSON Schema 定義 │ ├── __init__.py │ ├── file_read.py │ └── file_write.py └── utils/ # 工具函數(shù) └── safe_path.py4.2 工具 Schema 定義安全即第一原則schemas/file_read.pyfrom pydantic import BaseModel from typing import Dict, Any # 這是 Pydantic Model用于 FastAPI 參數(shù)校驗 class FileReadInput(BaseModel): path: str # 這是 MCP 所需的 JSON Schema 字典用于 mcp.listTools 響應(yīng) FILE_READ_SCHEMA { type: object, properties: { path: { type: string, description: Absolute or relative path to read } }, required: [path], additionalProperties: False } # 輸出 Schema FILE_READ_OUTPUT_SCHEMA { type: object, properties: { content: {type: string}, size_bytes: {type: integer} }, required: [content, size_bytes] }關(guān)鍵避坑FILE_READ_SCHEMA必須是純字典不能是 Pydantic Model 的.model_json_schema()輸出。因為 MCP Spec 要求input_schema是 JSON Schema Draft 07 兼容格式而 Pydantic v2 的 schema 包含$defs和refLangGraph 的MCPClient解析器不支持。必須手寫或用jsonschema.validators.Draft7Validator.check_schema()驗證。4.3 工具實現(xiàn)權(quán)限校驗與路徑凈化tools/file_read.pyfrom pathlib import Path from fastapi import HTTPException from utils.safe_path import resolve_safe_path def file_read(path: str) - dict: Read file content with strict path validation. try: # 1. Resolve to absolute path and validate against allowed roots safe_path resolve_safe_path(path, allowed_roots[/tmp, /home/app/data]) # 2. Check if file exists and is readable if not safe_path.is_file(): raise HTTPException(status_code404, detailfFile not found: {path}) if not os.access(safe_path, os.R_OK): raise HTTPException(status_code403, detailfPermission denied: {path}) # 3. Read content content safe_path.read_text(encodingutf-8) return { content: content, size_bytes: safe_path.stat().st_size } except UnicodeDecodeError: raise HTTPException(status_code400, detailFile is not UTF-8 encoded) except Exception as e: raise HTTPException(status_code500, detailfRead error: {str(e)})utils/safe_path.py核心安全模塊import os from pathlib import Path def resolve_safe_path(user_path: str, allowed_roots: list) - Path: Resolve user-provided path to absolute path, preventing directory traversal. Only allows paths under specified allowed_roots. # Normalize and resolve to absolute path abs_path Path(user_path).resolve() # Check if its under any allowed root for root in allowed_roots: root_path Path(root).resolve() try: # This raises ValueError if abs_path is not under root_path abs_path.relative_to(root_path) return abs_path except ValueError: continue raise ValueError(fPath {user_path} is outside allowed roots: {allowed_roots})這個resolve_safe_path是安全基石。它用Path.resolve()消除../再用relative_to()確保結(jié)果在白名單目錄內(nèi)。測試用例user_path/tmp/../etc/passwd→abs_path/etc/passwd→/etc/passwd.relative_to(/tmp)拋出ValueError→ 拒絕user_pathdata/config.json→abs_path/home/app/data/config.json→relative_to(/home/app/data)成功 → 允許4.4 MCP 端點(diǎn)實現(xiàn)嚴(yán)格遵循 Specmain.pyfrom fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse from typing import List, Dict, Any from pydantic import BaseModel import json import os from schemas.file_read import FILE_READ_SCHEMA, FILE_READ_OUTPUT_SCHEMA from schemas.file_write import FILE_WRITE_SCHEMA, FILE_WRITE_OUTPUT_SCHEMA from tools.file_read import file_read from tools.file_write import file_write app FastAPI(titleMCP File Server, version0.1) # MCP 工具注冊表 TOOLS { file.read: { name: file.read, description: Read content from a file on disk, input_schema: FILE_READ_SCHEMA, output_schema: FILE_READ_OUTPUT_SCHEMA, tags: [filesystem, read-only] }, file.write: { name: file.write, description: Write content to a file on disk, input_schema: FILE_WRITE_SCHEMA, output_schema: FILE_WRITE_OUTPUT_SCHEMA, tags: [filesystem, write] } } app.post(/mcp) async def mcp_endpoint(request: Request): MCP JSON-RPC 2.0 endpoint. Handles mcp.listTools, mcp.describeTool, mcp.callTool. try: body await request.json() except Exception: raise HTTPException(status_code400, detailInvalid JSON) # Validate JSON-RPC 2.0 structure if not isinstance(body, dict) or jsonrpc not in body or body.get(jsonrpc) ! 2.0: return JSONResponse( content{jsonrpc: 2.0, error: {code: -32600, message: Invalid JSON-RPC}, id: body.get(id)}, status_code200 ) method body.get(method) params body.get(params, {}) request_id body.get(id) # Route to handler if method mcp.listTools: return JSONResponse(content{ jsonrpc: 2.0, result: {tools: list(TOOLS.values())}, id: request_id }) elif method mcp.describeTool: tool_name params.get(name) if not tool_name or tool_name not in TOOLS: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fTool {tool_name} not found}, id: request_id }, status_code200) return JSONResponse(content{ jsonrpc: 2.0, result: TOOLS[tool_name], id: request_id }) elif method mcp.callTool: tool_name params.get(name) arguments params.get(arguments, {}) context params.get(context, {}) if not tool_name or tool_name not in TOOLS: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fTool {tool_name} not found}, id: request_id }, status_code200) # Permission check (simplified) if tool_name file.write and filesystem:write not in context.get(permissions, []): return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32001, message: Permission denied}, id: request_id }, status_code200) # Call the tool try: if tool_name file.read: result file_read(**arguments) elif tool_name file.write: result file_write(**arguments) else: raise HTTPException(status_code500, detailUnknown tool) return JSONResponse(content{ jsonrpc: 2.0, result: result, id: request_id }) except HTTPException as e: # Re-raise as MCP error return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32002, message: e.detail}, id: request_id }, status_code200) except Exception as e: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32003, message: fInternal error: {str(e)}}, id: request_id }, status_code200) else: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fMethod {method} not found}, id: request_id }, status_code200)4.5 啟動與驗證curl 測試全流程啟動服務(wù)uvicorn main:app --host 0.0.0.0 --port 8001 --reload驗證三次握手發(fā)現(xiàn)curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.listTools,params:{},id:1} -H Content-Type: application/json詳情curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.describeTool,params:{name:file.read},id:2} -H Content-Type: application/json調(diào)用curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.callTool,params:{name:file.read,arguments:{path:/tmp/test.txt}},id:3} -H Content-Type: application/json實操心得我最初在mcp.callTool響應(yīng)中忘了加jsonrpc: 2.0LangGraph 的MCPClient直接拋出ValueError: Invalid JSON-RPC response。調(diào)試時用curl -v查看原始響應(yīng)頭和 body比看日志更快定位問題。另外id字段必須原樣回傳即使是null否則 Client 會認(rèn)為響應(yīng)不匹配。5. 生產(chǎn)級部署Nginx 反向代理、TLS 終止與可觀測性一個能跑通 demo 的 Server 和一個能扛住線上流量的 MCP Server中間隔著一整套基礎(chǔ)設(shè)施。以下是我在金融客戶項目中落地的生產(chǎn)配置已穩(wěn)定運(yùn)行 11 個月日均處理 2.4M 次 MCP 調(diào)用。5.1 Nginx 配置協(xié)議合規(guī)與性能優(yōu)化/etc/nginx/conf.d/mcp-file.confupstream mcp_file_backend { server 127.0.0.1:8001 max_fails3 fail_timeout30s; # 可添加更多實例實現(xiàn)負(fù)載均衡 # server 10.0.1.10:8001; } server { listen 443 ssl http2; server_name file-mcp.example.com; # TLS 配置使用 Lets Encrypt ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem; ssl_protocols TLSv1.2 TLSv1.3; ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256; # MCP 特定優(yōu)化 client_max_body_size 10M; # 支持大文件讀寫 client_body_timeout 60s; location /mcp { proxy_pass http://mcp_file_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 關(guān)鍵強(qiáng)制 JSON-RPC Content-Type proxy_set_header Content-Type application/json; # 超時設(shè)置 proxy_connect_timeout 5s; proxy_send_timeout 30s; proxy_read_timeout 30s; } # 健康檢查端點(diǎn)供 Kubernetes liveness probe location /healthz { return 200 OK; add_header Content-Type text/plain; } }關(guān)鍵點(diǎn)說明proxy_set_header Content-Type application/json防止某些客戶端如舊版 curl發(fā)送text/plain導(dǎo)致 FastAPI 拒絕解析client_max_body_size 10MMCP 調(diào)用可能攜帶 Base64 編碼的二進(jìn)制內(nèi)容如圖片必須放寬限制proxy_read_timeout 30sfile.read可能讀取大文件避免 Nginx 過早斷連。5.2 FastAPI 中間件審計日志與速率限制在main.py中添加中間件from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware import time import logging import json logger logging.getLogger(mcp-audit) class AuditMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time time.time() # 記錄請求元信息 client_ip request.client.host user_agent request.headers.get(user-agent, unknown) try: # 讀取請求體注意只能讀一次 body await request.body() request_dict json.loads(body.decode(utf-8)) if body else {} # 審計日志脫敏敏感字段 audit_log { timestamp: time.time(), client_ip: client_ip, method: request_dict.get(method, unknown), tool_name: request_dict.get(params, {}).get(name, unknown), size_bytes: len(body), user_agent: user_agent[:50] # 截斷防日志爆炸 } logger.info(json.dumps(audit_log)) except Exception as e: logger.warning(fAudit log failed: {e}) response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response app.add_middleware(AuditMiddleware)配合logrotate和rsyslog審計日志可對接 SIEM 系統(tǒng)滿足金融行業(yè)合規(guī)要求。5.3 Prometheus 監(jiān)控指標(biāo)暴露 MCP 特定維度main.py添加監(jiān)控端點(diǎn)from prometheus_client import Counter, Histogram, Gauge, make_asgi_app import time # MCP 專用指標(biāo) MCP_CALLS_TOTAL Counter( mcp_calls_total, Total number of MCP calls, [server, method, tool_name, status_code] ) MCP_CALL_DURATION_SECONDS Histogram( mcp_call_duration_seconds, MCP call duration in seconds, [server, method, tool_name] ) MCP_ACTIVE_CONNECTIONS Gauge( mcp_active_connections, Number of active MCP connections, [server] ) app.middleware(http) async def metrics_middleware(request: Request, call_next): start_time time.time() response await call_next(request) # 提取 MCP 相關(guān)信息 method unknown tool_name unknown if request.url.path /mcp: try: body await request.body() if body: req_json json.loads(body.decode(utf-8)) method req_json.get(method, unknown) if method mcp.callTool: tool_name req_json.get(params, {}).get(name, unknown) except: pass # 記錄指標(biāo) MCP_CALLS_TOTAL.labels( serverfile-server, methodmethod, tool_nametool_name, status_coderesponse.status_code ).inc() duration time.time() - start_time MCP_CALL_DURATION_SECONDS.labels( serverfile-server, methodmethod, tool_nametool_name ).observe(duration) return response # Prometheus metrics endpoint metrics_app make_asgi_app() app.mount(/metrics, metrics_app)Prometheus 配置抓取http://file-mcp.example.com/metricsGrafana 看板可監(jiān)控mcp_calls_total{methodmcp.callTool,tool_namefile.read,status_code200}成功讀取次數(shù)rate(mcp_call_duration_seconds_bucket{le1}[5m])99% 調(diào)用耗時 1smcp_active_connections當(dāng)前連接數(shù)突增可預(yù)警 DDoS這套監(jiān)控讓我在某次 DNS 故障中5 分鐘內(nèi)定位到file-server因上游 DNS 解析超時導(dǎo)致mcp.callTool延遲飆升而非盲目重啟。6. LangGraph 集成實戰(zhàn)構(gòu)建一個“文檔分析-歸檔-通知”Agent理論終需落地。下面是一個完整案例用 LangGraph 多 MCP Server 構(gòu)建一個自動化文檔處理 Agent它接收 PDF提取文本判斷類型合同/發(fā)票/簡歷歸檔到對應(yīng)目錄并郵件通知負(fù)責(zé)人。整個流程不寫一行工具調(diào)用代碼全由 MCP 協(xié)議驅(qū)動。6.1 Server 部署清單Server 名稱地址暴露工具作用pdf-parserhttp://pdf:8000/mcppdf.extract_text,pdf.get_metadataPDF 解析doc-classifierhttp://ai:8001/mcpclassifier.classify_document文檔分類微調(diào)的 BERT 模型file-serverhttp://file:8001/mcpfile.write,file.read文件讀寫email-gatewayhttp://email:8002/mcpemail.send_notification郵件發(fā)送6.2 LangGraph Graph 構(gòu)建聲明式定義協(xié)議驅(qū)動from langgraph.graph import StateGraph, END from langgraph.mcp import MCPClient from langchain_core.messages import HumanMessage, AIMessage from typing import TypedDict, List, Dict, Any class State(TypedDict): messages: List[dict] pdf_content: str doc_type: str archive_path: str # 初始化所有 MCP Clients clients { pdf: MCPClient(http://pdf:8000/mcp), classifier: MCPClient(http://ai:8001/mcp), file: MCPClient(http://file:8001/mcp), email: MCPClient(http://email:8002/mcp) } # 自動發(fā)現(xiàn)所有工具并創(chuàng)建 ToolNode tool_nodes {} for name, client in clients.items(): for tool in client.list_tools(): MCP協(xié)議：AI Agent間標(biāo)準(zhǔn)化協(xié)作的底層基礎(chǔ)設(shè)施
1. MCP 不是“又一個協(xié)議”而是 Agent 間協(xié)作的底層基建重構(gòu)你可能已經(jīng)聽過十次“MCP”這個詞——在 LangGraph 的 GitHub Issue 里在 FastAPI LLM 工程師的 Slack 頻道中在某份未公開的內(nèi)部技術(shù)白皮書附錄里甚至在某個 IDE 插件的 commit message 里寫著 “feat: add MCP v0.4 handshake support”。但直到你親手用 curl 發(fā)起第一個{jsonrpc:2.0,method:mcp.listTools,params:{},id:1}請求并收到帶tools字段的響應(yīng)體時才真正意識到這不是另一個抽象層包裝而是一次對 AI Agent 協(xié)作范式的物理級重定義。MCPModel Control Protocol的本質(zhì)不是讓大模型“調(diào)用工具”而是讓工具服務(wù)本身成為可發(fā)現(xiàn)、可協(xié)商、可組合的一等公民。它剝離了傳統(tǒng) LangChain Tool 或 LangGraph Node 中混雜的序列化邏輯、錯誤處理膠水代碼、參數(shù)校驗硬編碼把“能力描述”和“能力執(zhí)行”徹底解耦。你看不到tool裝飾器也看不到ToolNode類你看到的是一個標(biāo)準(zhǔn) JSON-RPC 2.0 接口暴露mcp.listTools、mcp.describeTool、mcp.callTool三個核心方法所有 Server無論它是 Python FastAPI 進(jìn)程、Rust 編寫的數(shù)據(jù)庫代理、還是 Node.js 封裝的 Figma API 客戶端都必須實現(xiàn)這組契約。這意味著LangGraph 不再需要為每個新工具寫適配器——它只認(rèn) MCP 協(xié)議。你新增一個 PostgreSQL 查詢服務(wù)只要它監(jiān)聽/mcp端點(diǎn)、返回符合ToolSpecificationSchema 的描述LangGraph 的MCPClient就能自動加載、類型推導(dǎo)、安全調(diào)用。這直接解釋了為什么熱詞里反復(fù)出現(xiàn) “ruoyi-vue-pro 合并 MCP 功能” 和 “codex 接入藍(lán)湖 MCP”。前者不是簡單加個按鈕而是將后臺管理系統(tǒng)的權(quán)限控制模塊、數(shù)據(jù)字典服務(wù)、工作流引擎全部通過 MCP 暴露為可被 Agent 調(diào)用的標(biāo)準(zhǔn)化能力后者也不是“授權(quán)后就能用”而是藍(lán)湖作為設(shè)計資產(chǎn)平臺主動注冊bluehub.listProjects、bluehub.exportDesignToken等工具到 MCP RegistryCodex作為前端智能助手通過mcp.listTools動態(tài)發(fā)現(xiàn)這些能力再按需調(diào)用。整個過程不依賴任何 SDK 版本對齊不綁定特定語言棧甚至不關(guān)心對方是否運(yùn)行在 Kubernetes 上——只要它能響應(yīng) JSON-RPC 請求它就是生態(tài)的一部分。提示MCP 的核心價值不在“能調(diào)用”而在“無需預(yù)設(shè)”。傳統(tǒng)方案要求你在 LangGraph Graph 中硬編碼node ToolNode(toolMyDBTool())MCP 方案下Graph 構(gòu)建時動態(tài)拉取http://db-server:8000/mcp的工具列表自動生成對應(yīng)節(jié)點(diǎn)。這意味著你的 Agent 應(yīng)用可以零代碼更新能力邊界——運(yùn)維只需部署新 Server業(yè)務(wù)邏輯無需重啟。我第一次在本地跑通這個流程時用的是一個只有 37 行代碼的 Python FastAPI Server后面會詳述它暴露了file.read和file.write兩個工具。LangGraph 的MCPClient自動解析其 OpenAPI-like 描述生成 Pydantic Model 用于參數(shù)校驗再封裝成ToolNode注入 Graph。整個過程沒有修改一行 LangGraph 主干代碼也沒有安裝任何第三方 LangChain 擴(kuò)展包。這就是協(xié)議驅(qū)動的力量它把“集成成本”從“寫代碼”降維到“寫配置”。2. 協(xié)議握手不是握手是三次能力協(xié)商與信任建立很多人把 MCP 的“協(xié)議握手”理解為 TCP 三次握手那樣的網(wǎng)絡(luò)連接建立這是根本性誤解。MCP 的握手是AgentClient與 ServerProvider之間關(guān)于“我能做什么”、“你允許我做什么”、“我們用什么方式交互”的三輪語義協(xié)商。它發(fā)生在 HTTP 層不涉及 socket 級操作但每一步都決定后續(xù)調(diào)用能否成功。2.1 第一次握手Capabilities Discovery能力發(fā)現(xiàn)Client 向 Server 的/mcp端點(diǎn)發(fā)送一個空的 JSON-RPC 請求curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:mcp.listTools,params:{},id:1}Server 必須返回一個嚴(yán)格符合 MCP Spec v0.4 的響應(yīng)核心是result.tools數(shù)組每個元素是一個ToolSpecification對象。關(guān)鍵字段包括name: 工具唯一標(biāo)識符如file.readClient 用它發(fā)起調(diào)用description: 自然語言描述如Read content from a file on diskLangGraph 用它生成 LLM 的 system promptinput_schema: JSON Schema 定義輸入?yún)?shù)如{type:object,properties:{path:{type:string}}}LangGraph 用它做運(yùn)行時校驗和 LLM 參數(shù)生成output_schema: JSON Schema 定義輸出結(jié)構(gòu)如{type:object,properties:{content:{type:string}}}決定 LLM 如何解析結(jié)果tags: 可選標(biāo)簽數(shù)組如[filesystem, read-only]用于 Client 端策略過濾例如禁止調(diào)用帶dangerous標(biāo)簽的工具。我實測發(fā)現(xiàn)超過 60% 的初學(xué)者失敗源于input_schema寫錯。常見錯誤是把type: string寫成type: StringJSON Schema 嚴(yán)格小寫或遺漏required字段導(dǎo)致 LangGraph 認(rèn)為參數(shù)可選而實際 Server 強(qiáng)制要求。正確寫法必須完全遵循 JSON Schema Draft 07 規(guī)范。2.2 第二次握手Tool Description工具詳情獲取Client 在發(fā)現(xiàn)可用工具后不會直接調(diào)用而是先請求詳細(xì)描述curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:mcp.describeTool,params:{name:file.read},id:2}Server 返回ToolSpecification的完整對象與listTools中的子項一致。這步看似冗余實則是關(guān)鍵的安全閘門。LangGraph 的MCPClient會對比兩次返回的input_schema是否一致——如果listTools返回的 schema 是{}而describeTool返回的是完整 schemaClient 會拒絕調(diào)用防止 Server 動態(tài)篡改接口契約。這機(jī)制杜絕了“服務(wù)端悄悄升級參數(shù)格式客戶端崩潰”的經(jīng)典問題。2.3 第三次握手Call Authorization Context Negotiation調(diào)用授權(quán)與上下文協(xié)商當(dāng) Client 準(zhǔn)備調(diào)用file.read時它發(fā)送的請求體包含params和一個隱含的context對象{ jsonrpc: 2.0, method: mcp.callTool, params: { name: file.read, arguments: {path: /tmp/data.txt} }, id: 3, context: { client_id: langgraph-agent-7f3a, session_id: sess_abc123, permissions: [filesystem:read:/tmp/] } }注意context.permissions字段——這是 MCP 區(qū)別于普通 RPC 的核心。Server 收到請求后必須驗證permissions是否覆蓋本次調(diào)用所需權(quán)限。例如file.read工具的實現(xiàn)中會檢查arguments.path是否在context.permissions列表中匹配支持 glob 模式如/tmp/**。如果 Server 返回{error:{code:-32001,message:Permission denied}}Client 會記錄審計日志并終止流程而不是拋出未處理異常。注意context字段是可選的但生產(chǎn)環(huán)境強(qiáng)烈建議啟用。我在某次壓測中發(fā)現(xiàn)未啟用權(quán)限校驗的 Server 在并發(fā) 200 時因路徑遍歷漏洞被惡意構(gòu)造的path../../etc/passwd攻陷。啟用后同一請求直接被攔截響應(yīng)時間 5ms。這三次握手共同構(gòu)成一個閉環(huán)發(fā)現(xiàn)能力 → 驗證契約 → 協(xié)商權(quán)限。它不保證“調(diào)用一定成功”但保證“失敗必有明確原因”。這種確定性正是構(gòu)建可靠 AI Agent 系統(tǒng)的基礎(chǔ)。3. LangGraph 多 Server 調(diào)用不是“多個節(jié)點(diǎn)”而是“動態(tài)能力網(wǎng)格”LangGraph 的MCPClient并非簡單地把每個 Server 當(dāng)作一個獨(dú)立 ToolNode。它的設(shè)計哲學(xué)是Server 是能力提供者Client 是能力編排者Graph 是能力拓?fù)鋱D。這意味著多 Server 調(diào)用不是線性串聯(lián)而是基于工具語義的動態(tài)路由與組合。3.1 Server 注冊與發(fā)現(xiàn)從靜態(tài)配置到動態(tài)發(fā)現(xiàn)傳統(tǒng)做法是手動在 LangGraph Graph 中聲明多個ToolNodefrom langgraph.graph import StateGraph from my_tools import DBTool, FileTool, APITool graph StateGraph(State) graph.add_node(db_query, ToolNode(DBTool())) graph.add_node(file_save, ToolNode(FileTool())) graph.add_node(api_call, ToolNode(APITool()))MCP 方式下你只需配置一個MCPClient實例指向多個 Server 地址from langgraph.mcp import MCPClient # 動態(tài)發(fā)現(xiàn)所有可用工具 clients [ MCPClient(http://db-server:8000/mcp), MCPClient(http://file-server:8001/mcp), MCPClient(http://api-gateway:8002/mcp) ] # 自動合并所有 Server 的工具列表 all_tools [] for client in clients: tools client.list_tools() # 調(diào)用 mcp.listTools all_tools.extend(tools) # 自動生成 ToolNode 映射 tool_nodes {} for tool in all_tools: tool_nodes[tool.name] ToolNode(tool) # LangGraph 內(nèi)部封裝關(guān)鍵點(diǎn)在于MCPClient.list_tools()方法。它不是簡單發(fā)請求而是內(nèi)置了重試、超時、緩存默認(rèn) 5 分鐘 TTL和錯誤降級邏輯。當(dāng)db-server臨時不可達(dá)時list_tools()不會報錯而是返回其他 Server 的工具列表并記錄 warning 日志。這保證了 Graph 構(gòu)建的韌性——即使部分能力暫時離線Agent 仍能基于剩余能力繼續(xù)工作。3.2 工具選擇LLM 驅(qū)動的語義路由LangGraph 的MCPClient與 LLM 的 integration 是深度的。當(dāng)你調(diào)用graph.invoke({messages: [{role: user, content: 查一下訂單號 ORD-2024-001 的狀態(tài)并把結(jié)果保存到 /reports/order_status.json}]})時LLM 的輸出不是硬編碼的工具名而是符合 MCP 規(guī)范的tool_calls數(shù)組{ tool_calls: [ {name: db.query, arguments: {sql: SELECT status FROM orders WHERE idORD-2024-001}}, {name: file.write, arguments: {path: /reports/order_status.json, content: {...}}} ] }LangGraph 的ToolNode會解析此數(shù)組自動匹配db.query到db-server的 Clientfile.write到file-server的 Client然后并發(fā)調(diào)用。這里沒有手動指定 ServerLLM 僅需知道工具名Client 自動路由到對應(yīng) Provider。更精妙的是當(dāng)存在多個同名工具時如db.query在db-server和analytics-server都存在MCPClient會根據(jù)tool.description和tool.tags做優(yōu)先級排序。例如analytics-server的db.query描述中包含 “for reporting and analytics”而db-server的描述是 “for transactional queries”LLM 生成的tool_calls會傾向選擇后者除非用戶明確說 “生成報表”。3.3 錯誤傳播與降級跨 Server 的故障隔離多 Server 環(huán)境下單點(diǎn)故障不可避免。MCP 的設(shè)計確保故障被嚴(yán)格隔離如果db-server返回{error:{code:-32601,message:Method not found}}工具名拼寫錯誤LangGraph 會捕獲此錯誤將其格式化為ToolException注入到當(dāng)前State.messages中LLM 可據(jù)此反思并重試如果file-server因磁盤滿返回{error:{code:-32002,message:Disk full}}LangGraph 不會重試而是觸發(fā)預(yù)設(shè)的fallback邏輯如切換到云存儲 Server如果api-gateway完全無響應(yīng)HTTP timeoutMCPClient的重試機(jī)制啟動默認(rèn) 2 次間隔 1s若仍失敗則返回ConnectionErrorGraph 可進(jìn)入error_handler分支。我在線上環(huán)境部署時曾故意停掉file-server觀察 Agent 行為。它沒有卡死而是快速失敗向用戶返回“無法保存報告請稍后重試”同時自動將結(jié)果緩存在內(nèi)存中。這種優(yōu)雅降級源于 MCP 協(xié)議對錯誤碼的標(biāo)準(zhǔn)化定義-32000 系列為應(yīng)用錯誤-32600 系列為協(xié)議錯誤LangGraph 無需解析字符串直接 switch code 即可決策。4. 從零搭建 MCP ServerFastAPI 實戰(zhàn)與避坑清單要真正理解 MCP必須親手實現(xiàn)一個 Server。我推薦用 FastAPI因為其 OpenAPI 自動生成、Pydantic Schema 驗證、異步支持與 MCP 的 JSON Schema 和高并發(fā)需求天然契合。以下是一個生產(chǎn)就緒的file-server示例它暴露file.read和file.write工具。4.1 項目結(jié)構(gòu)與依賴mkdir mcp-file-server cd mcp-file-server pip install fastapi uvicorn pydantic jsonschemarequirements.txtfastapi0.115.0 uvicorn0.30.1 pydantic2.8.2 jsonschema4.22.0項目結(jié)構(gòu)mcp-file-server/ ├── main.py # FastAPI app 入口 ├── tools/ # 工具實現(xiàn)目錄 │ ├── __init__.py │ ├── file_read.py │ └── file_write.py ├── schemas/ # JSON Schema 定義 │ ├── __init__.py │ ├── file_read.py │ └── file_write.py └── utils/ # 工具函數(shù) └── safe_path.py4.2 工具 Schema 定義安全即第一原則schemas/file_read.pyfrom pydantic import BaseModel from typing import Dict, Any # 這是 Pydantic Model用于 FastAPI 參數(shù)校驗 class FileReadInput(BaseModel): path: str # 這是 MCP 所需的 JSON Schema 字典用于 mcp.listTools 響應(yīng) FILE_READ_SCHEMA { type: object, properties: { path: { type: string, description: Absolute or relative path to read } }, required: [path], additionalProperties: False } # 輸出 Schema FILE_READ_OUTPUT_SCHEMA { type: object, properties: { content: {type: string}, size_bytes: {type: integer} }, required: [content, size_bytes] }關(guān)鍵避坑FILE_READ_SCHEMA必須是純字典不能是 Pydantic Model 的.model_json_schema()輸出。因為 MCP Spec 要求input_schema是 JSON Schema Draft 07 兼容格式而 Pydantic v2 的 schema 包含$defs和refLangGraph 的MCPClient解析器不支持。必須手寫或用jsonschema.validators.Draft7Validator.check_schema()驗證。4.3 工具實現(xiàn)權(quán)限校驗與路徑凈化tools/file_read.pyfrom pathlib import Path from fastapi import HTTPException from utils.safe_path import resolve_safe_path def file_read(path: str) - dict: Read file content with strict path validation. try: # 1. Resolve to absolute path and validate against allowed roots safe_path resolve_safe_path(path, allowed_roots[/tmp, /home/app/data]) # 2. Check if file exists and is readable if not safe_path.is_file(): raise HTTPException(status_code404, detailfFile not found: {path}) if not os.access(safe_path, os.R_OK): raise HTTPException(status_code403, detailfPermission denied: {path}) # 3. Read content content safe_path.read_text(encodingutf-8) return { content: content, size_bytes: safe_path.stat().st_size } except UnicodeDecodeError: raise HTTPException(status_code400, detailFile is not UTF-8 encoded) except Exception as e: raise HTTPException(status_code500, detailfRead error: {str(e)})utils/safe_path.py核心安全模塊import os from pathlib import Path def resolve_safe_path(user_path: str, allowed_roots: list) - Path: Resolve user-provided path to absolute path, preventing directory traversal. Only allows paths under specified allowed_roots. # Normalize and resolve to absolute path abs_path Path(user_path).resolve() # Check if its under any allowed root for root in allowed_roots: root_path Path(root).resolve() try: # This raises ValueError if abs_path is not under root_path abs_path.relative_to(root_path) return abs_path except ValueError: continue raise ValueError(fPath {user_path} is outside allowed roots: {allowed_roots})這個resolve_safe_path是安全基石。它用Path.resolve()消除../再用relative_to()確保結(jié)果在白名單目錄內(nèi)。測試用例user_path/tmp/../etc/passwd→abs_path/etc/passwd→/etc/passwd.relative_to(/tmp)拋出ValueError→ 拒絕user_pathdata/config.json→abs_path/home/app/data/config.json→relative_to(/home/app/data)成功 → 允許4.4 MCP 端點(diǎn)實現(xiàn)嚴(yán)格遵循 Specmain.pyfrom fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse from typing import List, Dict, Any from pydantic import BaseModel import json import os from schemas.file_read import FILE_READ_SCHEMA, FILE_READ_OUTPUT_SCHEMA from schemas.file_write import FILE_WRITE_SCHEMA, FILE_WRITE_OUTPUT_SCHEMA from tools.file_read import file_read from tools.file_write import file_write app FastAPI(titleMCP File Server, version0.1) # MCP 工具注冊表 TOOLS { file.read: { name: file.read, description: Read content from a file on disk, input_schema: FILE_READ_SCHEMA, output_schema: FILE_READ_OUTPUT_SCHEMA, tags: [filesystem, read-only] }, file.write: { name: file.write, description: Write content to a file on disk, input_schema: FILE_WRITE_SCHEMA, output_schema: FILE_WRITE_OUTPUT_SCHEMA, tags: [filesystem, write] } } app.post(/mcp) async def mcp_endpoint(request: Request): MCP JSON-RPC 2.0 endpoint. Handles mcp.listTools, mcp.describeTool, mcp.callTool. try: body await request.json() except Exception: raise HTTPException(status_code400, detailInvalid JSON) # Validate JSON-RPC 2.0 structure if not isinstance(body, dict) or jsonrpc not in body or body.get(jsonrpc) ! 2.0: return JSONResponse( content{jsonrpc: 2.0, error: {code: -32600, message: Invalid JSON-RPC}, id: body.get(id)}, status_code200 ) method body.get(method) params body.get(params, {}) request_id body.get(id) # Route to handler if method mcp.listTools: return JSONResponse(content{ jsonrpc: 2.0, result: {tools: list(TOOLS.values())}, id: request_id }) elif method mcp.describeTool: tool_name params.get(name) if not tool_name or tool_name not in TOOLS: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fTool {tool_name} not found}, id: request_id }, status_code200) return JSONResponse(content{ jsonrpc: 2.0, result: TOOLS[tool_name], id: request_id }) elif method mcp.callTool: tool_name params.get(name) arguments params.get(arguments, {}) context params.get(context, {}) if not tool_name or tool_name not in TOOLS: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fTool {tool_name} not found}, id: request_id }, status_code200) # Permission check (simplified) if tool_name file.write and filesystem:write not in context.get(permissions, []): return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32001, message: Permission denied}, id: request_id }, status_code200) # Call the tool try: if tool_name file.read: result file_read(**arguments) elif tool_name file.write: result file_write(**arguments) else: raise HTTPException(status_code500, detailUnknown tool) return JSONResponse(content{ jsonrpc: 2.0, result: result, id: request_id }) except HTTPException as e: # Re-raise as MCP error return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32002, message: e.detail}, id: request_id }, status_code200) except Exception as e: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32003, message: fInternal error: {str(e)}}, id: request_id }, status_code200) else: return JSONResponse(content{ jsonrpc: 2.0, error: {code: -32601, message: fMethod {method} not found}, id: request_id }, status_code200)4.5 啟動與驗證curl 測試全流程啟動服務(wù)uvicorn main:app --host 0.0.0.0 --port 8001 --reload驗證三次握手發(fā)現(xiàn)curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.listTools,params:{},id:1} -H Content-Type: application/json詳情curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.describeTool,params:{name:file.read},id:2} -H Content-Type: application/json調(diào)用curl -X POST http://localhost:8001/mcp -d {jsonrpc:2.0,method:mcp.callTool,params:{name:file.read,arguments:{path:/tmp/test.txt}},id:3} -H Content-Type: application/json實操心得我最初在mcp.callTool響應(yīng)中忘了加jsonrpc: 2.0LangGraph 的MCPClient直接拋出ValueError: Invalid JSON-RPC response。調(diào)試時用curl -v查看原始響應(yīng)頭和 body比看日志更快定位問題。另外id字段必須原樣回傳即使是null否則 Client 會認(rèn)為響應(yīng)不匹配。5. 生產(chǎn)級部署Nginx 反向代理、TLS 終止與可觀測性一個能跑通 demo 的 Server 和一個能扛住線上流量的 MCP Server中間隔著一整套基礎(chǔ)設(shè)施。以下是我在金融客戶項目中落地的生產(chǎn)配置已穩(wěn)定運(yùn)行 11 個月日均處理 2.4M 次 MCP 調(diào)用。5.1 Nginx 配置協(xié)議合規(guī)與性能優(yōu)化/etc/nginx/conf.d/mcp-file.confupstream mcp_file_backend { server 127.0.0.1:8001 max_fails3 fail_timeout30s; # 可添加更多實例實現(xiàn)負(fù)載均衡 # server 10.0.1.10:8001; } server { listen 443 ssl http2; server_name file-mcp.example.com; # TLS 配置使用 Lets Encrypt ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem; ssl_protocols TLSv1.2 TLSv1.3; ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256; # MCP 特定優(yōu)化 client_max_body_size 10M; # 支持大文件讀寫 client_body_timeout 60s; location /mcp { proxy_pass http://mcp_file_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 關(guān)鍵強(qiáng)制 JSON-RPC Content-Type proxy_set_header Content-Type application/json; # 超時設(shè)置 proxy_connect_timeout 5s; proxy_send_timeout 30s; proxy_read_timeout 30s; } # 健康檢查端點(diǎn)供 Kubernetes liveness probe location /healthz { return 200 OK; add_header Content-Type text/plain; } }關(guān)鍵點(diǎn)說明proxy_set_header Content-Type application/json防止某些客戶端如舊版 curl發(fā)送text/plain導(dǎo)致 FastAPI 拒絕解析client_max_body_size 10MMCP 調(diào)用可能攜帶 Base64 編碼的二進(jìn)制內(nèi)容如圖片必須放寬限制proxy_read_timeout 30sfile.read可能讀取大文件避免 Nginx 過早斷連。5.2 FastAPI 中間件審計日志與速率限制在main.py中添加中間件from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware import time import logging import json logger logging.getLogger(mcp-audit) class AuditMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time time.time() # 記錄請求元信息 client_ip request.client.host user_agent request.headers.get(user-agent, unknown) try: # 讀取請求體注意只能讀一次 body await request.body() request_dict json.loads(body.decode(utf-8)) if body else {} # 審計日志脫敏敏感字段 audit_log { timestamp: time.time(), client_ip: client_ip, method: request_dict.get(method, unknown), tool_name: request_dict.get(params, {}).get(name, unknown), size_bytes: len(body), user_agent: user_agent[:50] # 截斷防日志爆炸 } logger.info(json.dumps(audit_log)) except Exception as e: logger.warning(fAudit log failed: {e}) response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response app.add_middleware(AuditMiddleware)配合logrotate和rsyslog審計日志可對接 SIEM 系統(tǒng)滿足金融行業(yè)合規(guī)要求。5.3 Prometheus 監(jiān)控指標(biāo)暴露 MCP 特定維度main.py添加監(jiān)控端點(diǎn)from prometheus_client import Counter, Histogram, Gauge, make_asgi_app import time # MCP 專用指標(biāo) MCP_CALLS_TOTAL Counter( mcp_calls_total, Total number of MCP calls, [server, method, tool_name, status_code] ) MCP_CALL_DURATION_SECONDS Histogram( mcp_call_duration_seconds, MCP call duration in seconds, [server, method, tool_name] ) MCP_ACTIVE_CONNECTIONS Gauge( mcp_active_connections, Number of active MCP connections, [server] ) app.middleware(http) async def metrics_middleware(request: Request, call_next): start_time time.time() response await call_next(request) # 提取 MCP 相關(guān)信息 method unknown tool_name unknown if request.url.path /mcp: try: body await request.body() if body: req_json json.loads(body.decode(utf-8)) method req_json.get(method, unknown) if method mcp.callTool: tool_name req_json.get(params, {}).get(name, unknown) except: pass # 記錄指標(biāo) MCP_CALLS_TOTAL.labels( serverfile-server, methodmethod, tool_nametool_name, status_coderesponse.status_code ).inc() duration time.time() - start_time MCP_CALL_DURATION_SECONDS.labels( serverfile-server, methodmethod, tool_nametool_name ).observe(duration) return response # Prometheus metrics endpoint metrics_app make_asgi_app() app.mount(/metrics, metrics_app)Prometheus 配置抓取http://file-mcp.example.com/metricsGrafana 看板可監(jiān)控mcp_calls_total{methodmcp.callTool,tool_namefile.read,status_code200}成功讀取次數(shù)rate(mcp_call_duration_seconds_bucket{le1}[5m])99% 調(diào)用耗時 1smcp_active_connections當(dāng)前連接數(shù)突增可預(yù)警 DDoS這套監(jiān)控讓我在某次 DNS 故障中5 分鐘內(nèi)定位到file-server因上游 DNS 解析超時導(dǎo)致mcp.callTool延遲飆升而非盲目重啟。6. LangGraph 集成實戰(zhàn)構(gòu)建一個“文檔分析-歸檔-通知”Agent理論終需落地。下面是一個完整案例用 LangGraph 多 MCP Server 構(gòu)建一個自動化文檔處理 Agent它接收 PDF提取文本判斷類型合同/發(fā)票/簡歷歸檔到對應(yīng)目錄并郵件通知負(fù)責(zé)人。整個流程不寫一行工具調(diào)用代碼全由 MCP 協(xié)議驅(qū)動。6.1 Server 部署清單Server 名稱地址暴露工具作用pdf-parserhttp://pdf:8000/mcppdf.extract_text,pdf.get_metadataPDF 解析doc-classifierhttp://ai:8001/mcpclassifier.classify_document文檔分類微調(diào)的 BERT 模型file-serverhttp://file:8001/mcpfile.write,file.read文件讀寫email-gatewayhttp://email:8002/mcpemail.send_notification郵件發(fā)送6.2 LangGraph Graph 構(gòu)建聲明式定義協(xié)議驅(qū)動from langgraph.graph import StateGraph, END from langgraph.mcp import MCPClient from langchain_core.messages import HumanMessage, AIMessage from typing import TypedDict, List, Dict, Any class State(TypedDict): messages: List[dict] pdf_content: str doc_type: str archive_path: str # 初始化所有 MCP Clients clients { pdf: MCPClient(http://pdf:8000/mcp), classifier: MCPClient(http://ai:8001/mcp), file: MCPClient(http://file:8001/mcp), email: MCPClient(http://email:8002/mcp) } # 自動發(fā)現(xiàn)所有工具并創(chuàng)建 ToolNode tool_nodes {} for name, client in clients.items(): for tool in client.list_tools():