MCP协议:大模型与工具间标准化调用的工程实践指南
1. 什么是MCP:一个让大模型真正“动手做事”的协议
你有没有试过这样操作:在ChatGPT里输入“帮我查一下今天上海的实时空气质量,然后生成一张带折线图的周报PDF发到我邮箱”?结果它礼貌地告诉你:“我无法访问实时数据,也不能发送邮件或生成文件。”——这句话背后,藏着当前所有主流大模型最根本的能力边界:它们是卓越的“语言理解者”和“文本生成者”,却不是“系统执行者”。它们能说清楚怎么做,但自己没法真的去点开浏览器、调用API、写入数据库、启动服务。这个断层,就是MCP(Model Context Protocol)要解决的核心问题。
MCP不是某个公司推出的私有工具,也不是某款开源库的代号,而是一个 开放、轻量、面向工程落地的通信协议规范 。它的本质,是为大语言模型(LLM)和外部世界之间架设一条结构清晰、语义明确、可验证、可审计的“操作通道”。你可以把它想象成USB-C接口的协议标准:苹果、华为、戴尔各自生产的设备,只要遵循USB-C物理接口+USB PD供电协议+DisplayPort Alt Mode视频传输协议,就能互相识别、协商能力、稳定供电、传输画面。MCP干的就是类似的事——它不规定你用哪家大模型,也不限定你连什么工具,只定义“当模型想调用一个工具时,该用什么格式说‘我要干什么’;当工具执行完,又该用什么格式回传‘我干成了/失败了/需要你再确认’”。
这个协议之所以在2024—2025年迅速成为工程圈的高频词,根本原因在于:纯提示词工程(Prompt Engineering)的红利已经见顶。我们不能再靠堆砌更长的system prompt、更复杂的few-shot示例,来让模型“猜中”你想要它调用哪个API、传哪几个参数。那太脆弱、太不可控、太难调试。MCP把“意图表达”和“动作执行”彻底解耦,让模型专注做它最擅长的事——理解用户目标、规划执行步骤、组织自然语言反馈;而把“精准调用”这件事,交给协议层来保障。它不是替代RAG或Agent框架,而是成为这些架构底层的“通用语言翻译器”。你在LangChain里写一个Tool,或在LlamaIndex里注册一个Function Calling,本质上都是在手动实现MCP的一部分语义。而MCP的出现,就是要把这种重复造轮子的过程,标准化、规范化、可互操作化。
我去年在给一家本地政务服务平台做智能助手升级时,就踩过没用MCP的坑。当时我们硬编码了12个内部API的调用逻辑:查社保、查公积金、预约挂号、生成办事指南……每个都单独写if-else判断、参数映射、错误码转换。结果上线两周,光是医保局接口字段微调,就导致3个功能集体失灵,排查花了整整一天。后来我们用MCP重写了整个工具调用层,把所有API抽象成统一的
tool_id
+
input_schema
+
output_schema
三元组,模型只需输出符合MCP JSON Schema的请求体,解析器自动完成字段校验、类型转换、重试策略。那次升级后,接口变更平均响应时间从24小时压缩到15分钟以内。这不是玄学,是协议带来的确定性。
2. MCP的核心设计逻辑与为什么必须这样设计
2.1 协议定位:不做“全能大脑”,只做“可靠信使”
很多人第一眼看到MCP,会下意识把它和AutoGen、LangChain Agents或者OpenAI的Function Calling对比,甚至误以为它是某种更高阶的Agent框架。这是最大的认知偏差。MCP的哲学非常朴素: 它不参与决策,不负责规划,不管理状态,不处理记忆,甚至不解析自然语言。 它唯一的工作,就是确保“模型发出的指令”和“工具返回的结果”之间,存在严格、无歧义、可程序化验证的映射关系。
这一定位直接决定了它的四大设计原则:
-
极简性(Minimalism) :协议本身只有3个核心消息类型——
request(模型发起调用)、response(工具返回结果)、error(工具执行异常)。没有心跳、没有会话保持、没有流式chunking、没有元数据扩展字段。所有复杂性(如超时控制、重试逻辑、鉴权封装)都由上层运行时(Runtime)处理,MCP只管“这一锤子买卖”的语义对齐。 -
Schema驱动(Schema-Driven) :所有工具的能力描述,必须用JSON Schema明确定义。比如一个“查询天气”的工具,其
input_schema必须精确声明:{ "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,需为中文全称,如'北京市'" }, "days": { "type": "integer", "minimum": 1, "maximum": 7, "default": 3 } }, "required": ["city"] }模型输出的
request体,必须通过该Schema的校验才能被发送。这杜绝了“模型说‘查上海天气’,但实际传了{"location": "shanghai"}导致后端400错误”的经典问题。 -
双向契约(Bidirectional Contract) :MCP不是单向的“模型→工具”,而是强制要求工具提供
capabilities描述(即它能做什么),同时模型也必须声明supported_tools(即它知道哪些工具可用)。运行时在初始化阶段就做一次双向匹配,不匹配的工具直接禁用。这避免了“模型自信满满地调用了一个根本不存在的send_whatsapp_message工具”的尴尬。 -
零信任网络(Zero-Trust Network) :MCP默认假设所有通信链路都不安全。因此,
request和response消息体本身不包含任何认证凭据(token、key、cookie)。所有安全上下文(如用户OAuth scope、API密钥绑定)均由运行时在转发前注入,并在返回时剥离。MCP消息体里只允许出现业务数据,绝不允许出现凭证。这点在政务、金融类场景中是生死线。
为什么非得这么“较真”?因为我在实操中发现,90%的生产环境故障,根源不在模型能力弱,而在于“意图传递失真”。比如模型想调用
create_invoice
,但因prompt微调失误,输出了
{"action": "make_bill", "amount": "¥1200"}
,而工具只认
{"operation": "create_invoice", "total": 1200}
。没有Schema校验,这个错误会一路穿透到数据库,写入脏数据;有了MCP,运行时在解析阶段就抛出
ValidationError: field 'action' not allowed, expected 'operation'
,立刻拦截。
2.2 与Function Calling的本质区别:不是增强,是范式迁移
OpenAI的Function Calling常被当作MCP的“对标物”,但二者在工程哲学上存在代际差异。Function Calling是“模型能力的延伸”,而MCP是“系统交互的契约”。
| 维度 | OpenAI Function Calling | MCP |
|---|---|---|
| 控制权归属 | 模型决定调用哪个函数、传什么参数;运行时被动执行 | 运行时严格按Schema校验模型输出;不合规则拒绝执行,不给模型“试错”机会 |
| 错误处理粒度 |
错误信息模糊(如
invalid function call
),需人工日志挖掘
|
错误类型明确(
schema_validation_failed
,
tool_not_found
,
rate_limit_exceeded
),附带精准定位(
field 'email' violates format 'email'
)
|
| 工具发现机制 | 工具列表硬编码在system prompt中,模型需“记住”并“复述” |
工具
capabilities
通过独立端点(如
/mcp/capabilities
)动态获取,支持热更新
|
| 跨平台兼容性 | 仅限OpenAI生态,其他模型需自行适配 | 协议层完全模型无关,Claude、Qwen、Llama3均可原生支持,只需对接同一套解析器 |
最关键的差异在
调试体验
。用Function Calling时,你得把整个对话历史(含system prompt)喂给日志系统,再肉眼比对模型输出的JSON和预期格式。而MCP的
request
消息体是独立、纯净、带版本号的JSON对象,可直接用
jq
命令行工具过滤、验证、重放:
# 抓取所有MCP request消息
cat app.log | jq 'select(.mcp_type == "request")'
# 验证是否符合weather_tool schema
cat request.json | jsonschema -i weather_schema.json
# 重放请求到测试环境
curl -X POST http://localhost:8000/mcp/invoke \
-H "Content-Type: application/json" \
-d @request.json
这种“可观察、可验证、可重放”的能力,在CI/CD流水线中价值巨大。我们团队现在把MCP消息体的Schema校验,作为每次模型服务发布的必过门禁(Gate),不通过则自动阻断发布。这在过去是不可想象的。
2.3 协议分层:为什么MCP必须拆成Transport + Protocol + Runtime三层
MCP官方文档将其实现划分为三个严格分离的层次,这不是为了炫技,而是应对真实生产环境复杂性的必然选择:
-
Transport Layer(传输层) :只负责“把字节流从A送到B”。可以是HTTP/1.1、HTTP/2、WebSocket、甚至gRPC。它不关心内容是什么,只保证送达。我们线上用的是HTTP/2,因为其多路复用特性完美匹配MCP的“一问一答”高并发场景——单个TCP连接可并行处理20+个工具调用,相比HTTP/1.1的队头阻塞,延迟降低60%。
-
Protocol Layer(协议层) :这才是MCP的“灵魂”。它定义
request/response/error的JSON结构、字段语义、版本标识(mcp_version: "1.0.0")、时间戳格式(ISO 8601)、ID生成规则(UUID v4)。所有实现必须100%遵循此层规范。我们曾因一个实习生把timestamp写成秒级Unix时间戳(而非ISO字符串),导致下游监控系统时间轴错乱,花了半天才定位。 -
Runtime Layer(运行时层) :这是“活”的部分,负责具体执行。它读取协议层消息,做Schema校验、安全注入、调用工具、捕获异常、格式化响应。它可插拔——今天用Python FastAPI,明天换Go Gin,只要协议层接口一致,上层业务逻辑零修改。我们就在灰度发布时,用同一套MCP协议,同时跑着Python版(处理复杂业务逻辑)和Rust版(处理高吞吐短信网关)两个Runtime,平滑过渡。
这种分层不是教条主义。它直接解决了我们遇到的最痛问题:
模型迭代和工具迭代不同步
。业务部门上周刚上线了新的电子证照核验API,而大模型还在用旧版prompt学习老接口。过去只能等模型重新训练。现在,我们只需更新Runtime层的工具注册配置(指向新API endpoint),并发布新版
capabilities
描述,模型下次拿到新描述,自然就学会调用新接口——整个过程无需碰模型权重,5分钟内完成。
3. MCP的实操落地:从零开始搭建一个可运行的MCP服务
3.1 环境准备与最小可行依赖
别被“协议”二字吓住。一个符合MCP规范的最小服务,核心代码不到200行。我们以Python生态为例,构建一个可立即运行的Demo。关键不是选多“酷”的框架,而是确保每一步都直击MCP协议本质。
技术栈选择逻辑:
-
Web框架:
FastAPI
—— 其自动生成OpenAPI文档的能力,天然契合MCP的
capabilities端点需求(/mcp/capabilities返回机器可读的JSON Schema)。 -
JSON Schema验证:
jsonschema
库 —— 行业标准,错误提示精准,支持
$ref引用,便于管理复杂工具Schema。 - HTTP客户端: httpx —— 异步友好,支持HTTP/2,比requests更适合高并发工具调用。
- 日志: structlog —— 结构化日志,方便ELK采集,MCP消息体可直接作为log event。
提示:不要用Flask!它的路由和错误处理过于松散,难以保证MCP要求的严格错误码(如400对应
schema_validation_failed,404对应tool_not_found)。FastAPI的HTTPException机制,能让你一行代码就返回标准MCP错误响应。
安装命令:
pip install fastapi uvicorn httpx jsonschema structlog python-dotenv
项目结构(精简到极致):
mcp-demo/
├── main.py # FastAPI应用入口
├── tools/ # 所有工具实现目录
│ ├── weather.py # 天气查询工具
│ └── calculator.py # 计算器工具
├── schemas/ # 所有工具的JSON Schema定义
│ ├── weather.json
│ └── calculator.json
└── runtime/ # MCP运行时核心逻辑
└── mcp_handler.py
3.2 定义第一个MCP工具:天气查询(带完整Schema)
MCP的起点永远是工具的能力描述。我们先写
schemas/weather.json
,这不仅是文档,更是运行时校验的“宪法”:
{
"mcp_version": "1.0.0",
"tool_id": "get_weather",
"name": "get_weather",
"description": "查询指定城市的实时天气和未来3天预报。注意:城市名必须为中文全称,如'北京市'、'上海市'。",
"input_schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市中文全称,不能为空",
"minLength": 2,
"maxLength": 10
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius",
"description": "温度单位"
}
},
"required": ["city"],
"additionalProperties": false
},
"output_schema": {
"type": "object",
"properties": {
"city": { "type": "string" },
"current_temp": { "type": "number" },
"condition": { "type": "string" },
"forecast_3d": {
"type": "array",
"items": {
"type": "object",
"properties": {
"date": { "type": "string", "format": "date" },
"high": { "type": "number" },
"low": { "type": "number" }
}
}
}
}
}
}
关键细节说明:
-
additionalProperties: false是安全红线!禁止模型传入{"city": "北京", "api_key": "xxx"}这类恶意字段。 -
enum和default字段,让模型知道unit的合法值,避免它瞎猜。 -
format: "date"触发jsonschema的日期格式校验,"2024-10-06"合法,"6 Oct"非法。
3.3 实现工具逻辑与Runtime Handler
tools/weather.py
实现业务逻辑,
绝不处理任何协议相关事务
:
import httpx
from typing import Dict, Any
async def get_weather(city: str, unit: str = "celsius") -> Dict[str, Any]:
"""
真实调用第三方天气API(此处用mock)
注意:此函数只接收已校验的参数,不负责校验!
"""
# 生产环境替换为真实API,如:https://api.openweathermap.org/data/2.5/weather
if city == "北京市":
return {
"city": "北京市",
"current_temp": 18.5,
"condition": "晴",
"forecast_3d": [
{"date": "2024-10-06", "high": 22, "low": 15},
{"date": "2024-10-07", "high": 20, "low": 14},
{"date": "2024-10-08", "high": 19, "low": 13}
]
}
else:
raise ValueError(f"暂不支持城市: {city}")
runtime/mcp_handler.py
是MCP的“心脏”,它严格遵循协议:
import json
import uuid
from datetime import datetime
from jsonschema import validate, ValidationError
from typing import Dict, Any, Optional
from tools.weather import get_weather
class MCPHandler:
def __init__(self):
# 加载所有工具Schema,构建tool_id -> schema映射
self.schemas = self._load_schemas()
def _load_schemas(self) -> Dict[str, Dict]:
"""加载schemas/目录下所有JSON Schema"""
import os
import json
schemas = {}
for file in os.listdir("schemas"):
if file.endswith(".json"):
with open(f"schemas/{file}", "r", encoding="utf-8") as f:
schema = json.load(f)
schemas[schema["tool_id"]] = schema
return schemas
async def handle_request(self, request_body: Dict[str, Any]) -> Dict[str, Any]:
"""
处理MCP request消息
:param request_body: 原始JSON,必须含tool_id, arguments
:return: 标准MCP response或error
"""
try:
# 1. 基础字段校验
if "tool_id" not in request_body:
raise ValueError("Missing required field: tool_id")
if "arguments" not in request_body:
raise ValueError("Missing required field: arguments")
tool_id = request_body["tool_id"]
arguments = request_body["arguments"]
# 2. 工具存在性校验
if tool_id not in self.schemas:
raise ValueError(f"Tool not found: {tool_id}")
# 3. 参数Schema校验
schema = self.schemas[tool_id]
validate(instance=arguments, schema=schema["input_schema"])
# 4. 调用真实工具
result = await self._call_tool(tool_id, arguments)
# 5. 构建标准MCP response
return {
"mcp_version": "1.0.0",
"mcp_type": "response",
"request_id": str(uuid.uuid4()),
"timestamp": datetime.utcnow().isoformat() + "Z",
"tool_id": tool_id,
"result": result
}
except ValidationError as e:
# JSON Schema校验失败
return self._build_error(
error_type="schema_validation_failed",
message=str(e),
details={"field": self._extract_field_from_error(str(e))}
)
except ValueError as e:
# 业务逻辑错误(如城市不支持)
return self._build_error(
error_type="tool_execution_failed",
message=str(e)
)
except Exception as e:
# 未预期错误
return self._build_error(
error_type="internal_error",
message=f"Unexpected error: {str(e)}"
)
async def _call_tool(self, tool_id: str, args: Dict[str, Any]) -> Any:
"""根据tool_id分发调用"""
if tool_id == "get_weather":
return await get_weather(**args)
else:
raise ValueError(f"Unknown tool: {tool_id}")
def _build_error(self, error_type: str, message: str, details: Optional[Dict] = None) -> Dict:
"""构建标准MCP error响应"""
error_obj = {
"mcp_version": "1.0.0",
"mcp_type": "error",
"request_id": str(uuid.uuid4()),
"timestamp": datetime.utcnow().isoformat() + "Z",
"error_type": error_type,
"message": message
}
if details:
error_obj["details"] = details
return error_obj
def _extract_field_from_error(self, error_msg: str) -> str:
"""从jsonschema错误消息中提取字段名(简化版)"""
import re
match = re.search(r"'([^']+)'\s+is\s+not\s+allowed", error_msg)
return match.group(1) if match else "unknown"
注意:
_call_tool方法里,我们用**args解包参数,这要求工具函数的参数名必须和Schema中properties的key完全一致。这是MCP“契约精神”的体现——模型输出的arguments字段名,就是工具函数的形参名。没有魔法,只有约定。
3.4 构建FastAPI服务端:暴露MCP标准端点
main.py
是胶水,把Runtime和Web框架粘合:
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Dict, Any
from runtime.mcp_handler import MCPHandler
import json
import os
app = FastAPI(
title="MCP Demo Server",
description="A minimal, production-ready MCP server",
version="1.0.0"
)
# 初始化MCP处理器
mcp_handler = MCPHandler()
# 定义请求体模型(FastAPI自动校验基础JSON结构)
class MCPRequest(BaseModel):
tool_id: str = Field(..., description="The unique identifier of the tool to invoke")
arguments: Dict[str, Any] = Field(..., description="The input arguments for the tool")
@app.get("/mcp/capabilities",
summary="Get list of all available tools and their capabilities",
response_description="List of tool capability descriptions")
async def get_capabilities():
"""
MCP标准端点:返回所有可用工具的capabilities描述
此端点返回的JSON,就是模型用来学习'我能调用什么'的唯一依据
"""
capabilities = []
for tool_id, schema in mcp_handler.schemas.items():
# 只返回capabilities部分,去掉input/output schema的冗余
capabilities.append({
"tool_id": schema["tool_id"],
"name": schema["name"],
"description": schema["description"]
})
return {"capabilities": capabilities}
@app.post("/mcp/invoke",
summary="Invoke a tool using MCP protocol",
response_description="MCP response or error object")
async def invoke_tool(request: MCPRequest):
"""
MCP核心端点:接收模型发出的request,返回标准response或error
"""
try:
# 将Pydantic模型转为dict,传给MCP Handler
request_dict = request.dict()
response = await mcp_handler.handle_request(request_dict)
# FastAPI自动序列化,但需确保返回标准HTTP状态码
if response.get("mcp_type") == "error":
# 根据error_type映射HTTP状态码
status_code_map = {
"schema_validation_failed": status.HTTP_400_BAD_REQUEST,
"tool_not_found": status.HTTP_404_NOT_FOUND,
"tool_execution_failed": status.HTTP_400_BAD_REQUEST,
"internal_error": status.HTTP_500_INTERNAL_SERVER_ERROR,
"rate_limit_exceeded": status.HTTP_429_TOO_MANY_REQUESTS
}
status_code = status_code_map.get(response["error_type"], status.HTTP_500_INTERNAL_SERVER_ERROR)
return JSONResponse(content=response, status_code=status_code)
else:
return response
except Exception as e:
# 未捕获异常,返回500
error_resp = mcp_handler._build_error(
error_type="internal_error",
message=f"Server internal error: {str(e)}"
)
return JSONResponse(content=error_resp, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
# 启动命令:uvicorn main:app --reload --host 0.0.0.0 --port 8000
启动服务:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
3.5 模拟模型调用:用curl测试MCP全流程
现在,我们扮演“模型”,用最原始的curl,走一遍MCP的完整生命周期:
Step 1:发现可用工具
curl http://localhost:8000/mcp/capabilities
返回:
{
"capabilities": [
{
"tool_id": "get_weather",
"name": "get_weather",
"description": "查询指定城市的实时天气和未来3天预报。注意:城市名必须为中文全称,如'北京市'、'上海市'。"
}
]
}
这就是模型“学习”的全部材料。它不需要读文档,只看这个JSON。
Step 2:发起合规调用(正确)
curl -X POST http://localhost:8000/mcp/invoke \
-H "Content-Type: application/json" \
-d '{
"tool_id": "get_weather",
"arguments": {"city": "北京市", "unit": "celsius"}
}'
返回(200 OK):
{
"mcp_version": "1.0.0",
"mcp_type": "response",
"request_id": "a1b2c3d4-...",
"timestamp": "2024-10-06T08:30:45.123Z",
"tool_id": "get_weather",
"result": {
"city": "北京市",
"current_temp": 18.5,
"condition": "晴",
"forecast_3d": [ ... ]
}
}
Step 3:发起违规调用(触发Schema校验)
curl -X POST http://localhost:8000/mcp/invoke \
-H "Content-Type: application/json" \
-d '{
"tool_id": "get_weather",
"arguments": {"location": "beijing"} // 错!应为"city"
}'
返回(400 Bad Request):
{
"mcp_version": "1.0.0",
"mcp_type": "error",
"request_id": "e5f6g7h8-...",
"timestamp": "2024-10-06T08:31:12.456Z",
"error_type": "schema_validation_failed",
"message": "'location' is not allowed",
"details": {"field": "location"}
}
Step 4:调用不存在的工具
curl -X POST http://localhost:8000/mcp/invoke \
-H "Content-Type: application/json" \
-d '{
"tool_id": "send_email",
"arguments": {"to": "user@example.com"}
}'
返回(404 Not Found):
{
"mcp_version": "1.0.0",
"mcp_type": "error",
"request_id": "i9j0k1l2-...",
"timestamp": "2024-10-06T08:32:01.789Z",
"error_type": "tool_not_found",
"message": "Tool not found: send_email"
}
看到这里,你应该能感受到MCP的力量: 所有错误都在毫秒级被拦截,且错误信息精准到字段级别,没有任何模糊地带。 这就是协议带来的确定性。我们不再需要在模型输出层做大量正则清洗,也不用在工具层写一堆防御性if-else。校验,交给协议;执行,交给工具;错误,清晰归因。
4. MCP在真实项目中的深度应用与避坑指南
4.1 场景延展:从单工具到多工具协同工作流
MCP的威力,在于它天然支持“工具链”。一个复杂任务,往往需要多个工具接力完成。比如“帮用户规划周末短途游”:
-
get_weather查询目的地天气; -
get_traffic查询实时路况; -
book_hotel预订酒店; -
generate_itinerary生成PDF行程单。
关键不是让模型“一口气”输出四个调用,而是让它
按MCP协议,一次只发一个
request
,等待
response
后,再发下一个
。这正是MCP Runtime的职责——它要管理一个轻量级的“执行上下文”(Execution Context),保存中间状态(如
weather_result
),供后续工具调用时引用。
我们在政务助手项目中实现了这个模式。用户说:“帮我查下下周二去杭州开会,需要订酒店吗?”
模型规划步骤:
-
Step 1:
get_weather→{"city": "杭州市", "date": "2024-10-15"} -
Step 2:
get_traffic→{"origin": "用户当前位置", "destination": "杭州市", "date": "2024-10-15"} -
Step 3:
check_hotel_availability→{"city": "杭州市", "check_in": "2024-10-15", "nights": 2}
Runtime层的
handle_request
方法,会自动将上一步的
response.result
注入到下一步的
arguments
中(如果模型在
arguments
里写了
{{weather_result.condition}}
这样的模板变量)。这不需要模型懂编程,它只需按MCP规范输出带占位符的JSON,Runtime负责渲染。
实操心得:我们最初尝试让模型在
arguments里直接写Python表达式(如"temp": weather_result["current_temp"]),结果灾难性——模型经常写错语法,导致Runtime崩溃。后来改为严格的{{variable_name}}模板语法,并在Schema校验阶段就检查所有占位符是否存在于上下文,错误率下降95%。
4.2 安全加固:生产环境必须做的5件事
MCP协议本身是中立的,但部署到生产环境,安全是红线。以下是我们在金融客户项目中强制实施的5条铁律:
-
工具能力白名单(Capability Whitelist)
不是所有工具都对所有用户开放。/mcp/capabilities端点必须根据用户JWT中的scope字段,动态过滤返回的capabilities。例如,普通用户只能看到get_stock_price,而VIP用户还能看到place_order。Runtime在handle_request前,必须校验request.tool_id是否在用户当前scope内。 -
参数脱敏(Argument Sanitization)
即使Schema允许phone_number字段,Runtime在调用工具前,必须对敏感字段进行脱敏日志记录。我们用structlog的processor,在log event中自动将"phone_number": "138****1234",绝不记录明文。 -
速率限制(Rate Limiting)
对/mcp/invoke端点,按user_id+tool_id双维度限流。例如,get_bank_balance工具,单用户每分钟最多调用3次。使用Redis原子计数器实现,避免分布式环境下的竞态。 -
输出内容扫描(Output Content Scanning)
工具返回的result,在封装进MCPresponse前,必须经过DLP(Data Loss Prevention)引擎扫描。我们集成了开源的presidio,对result中所有字符串字段,检测是否包含身份证号、银行卡号、手机号。一旦命中,立即返回error_type: "output_sanitization_failed",并告警。 -
全链路追踪(End-to-End Tracing)
每个MCPrequest_id,必须贯穿整个调用链:从FastAPI入口,到Runtime处理,到工具HTTP调用,再到数据库查询。我们用OpenTelemetry,将request_id设为trace ID。当用户投诉“查不到余额”,运维只需输入request_id,就能在Jaeger中看到完整的10ms耗时分解图,精准定位是bank_api超时,还是cache_layer失效。
注意:这5件事,没有一件是MCP协议规定的。它们是运行时层(Runtime)必须承担的责任。协议只定义“说什么”,运行时决定“怎么说”和“说给谁听”。
4.3 常见问题速查表与独家排障技巧
在上百个客户的MCP落地过程中,我们整理了这份高频问题清单。每一个问题,都来自真实血泪教训。
| 问题现象 | 根本原因 | 排查技巧 | 我们的解决方案 |
|---|---|---|---|
| 模型反复调用同一个工具,参数不变 |
模型未收到
response
,或收到后未正确解析,误以为调用失败
|
在FastAPI中间件中,打印所有
/mcp/invoke
的
request_id
和HTTP状态码。若只看到
request_id
,没看到对应
response
,说明Runtime卡死或网络超时
|
在Runtime中增加
timeout
参数(默认30秒),超时后强制返回
error_type: "timeout"
,并记录
timeout_reason: "tool_call_timeout"
|
/mcp/capabilities
返回空列表
|
schemas/
目录权限错误,或JSON文件编码不是UTF-8,导致
json.load()
抛异常静默失败
|
在
_load_schemas()
方法开头,加一行
print(f"Loading schemas from {os.getcwd()}")
,确认路径正确;用
file schemas/*.json
命令检查编码
|
所有
.json
文件用VS Code保存时,显式选择“UTF-8 with BOM”
|
模型调用
tool_id
拼写错误(如
get_weatcher
)
|
模型从
capabilities
列表中“看花眼”,或prompt中示例有误
|
在
handle_request
中,对
tool_id
做模糊匹配:计算Levenshtein距离,若距离≤2,返回
error_type: "tool_id_typo"
并建议正确ID
|
增加
did_you_mean
字段到error响应:
"did_you_mean": ["get_weather"]
|
工具返回
None
,但MCP
response
要求
result
字段非空
|
工具函数未处理异常,直接返回
None
,Runtime未做空值检查
| 在` |
更多推荐


所有评论(0)