智能体协议 MCP

第一章：MCP 概述

1.1 什么是 MCP（Model Context Protocol）

概念名称	说明	注意事项
MCP（Model Context Protocol）	一种标准化协议，用于定义大语言模型（LLM）与外部工具、环境或服务之间的上下文交互格式。它规定了如何封装用户输入、系统状态、可用工具列表以及模型生成的动作指令，使 LLM 能在结构化上下文中进行工具调用和多轮推理。	MCP 并非具体实现，而是一套通信规范，类似 REST 或 gRPC，但专为 LLM 的”思考-行动”循环设计。
上下文驱动	MCP 强调”上下文即协议”，所有交互必须携带完整的会话上下文（包括历史消息、工具定义、当前目标等），确保模型决策具备充分依据。	上下文过大可能超出模型 token 限制，需合理裁剪或摘要。
动作导向输出	模型在 MCP 下的输出不是自由文本，而是结构化的动作（Action）对象，如 { "tool": "search", "args": { "query": "天气" } }，便于程序解析执行。	模型需经过微调或提示工程（prompt engineering）才能稳定输出合规动作格式。

1.2 MCP 的设计目标与核心思想

概念名称	说明	注意事项
统一工具调用接口	提供跨模型、跨平台的标准化工具调用方式，避免每个 LLM 项目重复定义工具交互逻辑。	工具注册需遵循 MCP 的 JSON Schema 规范，确保兼容性。
显式上下文管理	将对话历史、工具元数据、用户意图等显式编码到请求中，提升模型推理的可解释性与可控性。	开发者需主动维护上下文完整性，不能依赖模型”记住”状态。
解耦模型与执行环境	模型只负责”决策”（生成动作），不负责”执行”；执行由客户端或代理完成，结果再反馈回上下文。	此解耦提高了系统安全性（模型不直接访问敏感 API）和可测试性。
支持多轮自主推理	允许模型在单次请求中规划多个动作，或通过多轮交互逐步达成复杂目标（如”查天气→订机票→发邮件”）。	需设计合理的终止条件，防止无限循环或无效动作序列。

1.3 MCP 与其他协议（如 OpenAI API、LangChain 工具调用）的对比

协议/框架	所属类别	用途	与 MCP 的主要区别	注意事项
OpenAI API（Function Calling）	商业 LLM 接口	允许 GPT 模型调用预定义函数	1. 闭源、绑定 OpenAI 生态 2. 函数定义需在每次请求中传入 3. 输出格式由 OpenAI 控制，不可定制	无法用于本地开源模型；成本随调用次数增加
LangChain 工具调用机制	LLM 应用开发框架	提供统一接口集成多种工具（如搜索、计算、数据库）	1. 是代码框架而非通信协议 2. 工具调用逻辑内嵌在 Python/JS 代码中 3. 不强制结构化上下文传输	适合快速原型，但跨语言/跨服务部署困难
MCP（Model Context Protocol）	开放协议标准	定义 LLM 与外部世界的通用交互格式	1. 语言无关、平台无关 2. 强调上下文完整性和动作标准化 3. 可用于任何支持 JSON 输入/输出的 LLM	需自行实现客户端/服务端；生态尚在早期阶段
Toolformer / ReAct 等研究范式	学术方法	探索 LLM 使用工具的能力	属于训练或推理策略，非运行时协议	通常需特定 prompt 模板，缺乏工程化接口

第二章：MCP 基础概念

2.1 上下文（Context）的定义与结构

概念名称	说明	注意事项
上下文（Context）	MCP 中的核心数据单元，包含当前会话所需的所有信息，用于指导模型生成合理的动作。上下文是 JSON 对象，作为每次 MCP 请求的主体。	上下文必须完整、自包含，不能依赖外部状态。
用户消息（user_messages）	数组，记录用户输入的历史消息，每条包含 role（如 “user”）和 content（文本）。	消息应按时间顺序排列；避免冗余或重复内容以节省 token。
系统指令（system_prompt）	可选字符串，提供全局行为指导（如”你是一个天气助手”）。	不应频繁变更；过长会影响有效上下文长度。
可用工具列表（tools）	数组，描述当前可调用的工具，每个工具含 name、description、parameters（JSON Schema）。	工具定义需符合 MCP 工具规范；未列出的工具不可被调用。
动作历史（action_history）	数组，记录已执行的动作及其结果，每项包含 action（动作对象）和 result（返回值或错误）。	用于支持多轮推理；结果应结构化，避免纯自由文本。
会话元数据（metadata）	可选字段，包含 session_id、user_id、timestamp 等，用于追踪和调试。	不参与模型推理，但对日志和审计至关重要。

示例上下文结构（简化）：

{
  "user_messages": [
    { "role": "user", "content": "今天北京天气如何？" }
  ],
  "tools": [
    {
      "name": "get_weather",
      "description": "查询指定城市的当前天气",
      "parameters": {
        "type": "object",
        "properties": {
          "city": { "type": "string" }
        },
        "required": ["city"]
      }
    }
  ],
  "action_history": [],
  "metadata": { "session_id": "sess_001" }
}

2.2 工具（Tool）与动作（Action）的抽象

概念名称	说明	注意事项
工具（Tool）	一个可被 LLM 调用的外部功能单元，具有明确输入输出契约。在 MCP 中以 JSON Schema 描述其接口。	工具应幂等、无副作用（或副作用可控）；避免高延迟操作。
工具名称（name）	字符串，唯一标识一个工具，如 “search_web” 或 “send_email”。	命名应清晰、无歧义；建议使用 snake_case。
工具描述（description）	自然语言说明工具用途，帮助模型理解何时调用。	描述需具体，避免模糊词汇如”处理数据”。
参数规范（parameters）	使用 JSON Schema 定义输入参数结构，包括类型、必填项、枚举等。	必须严格符合 JSON Schema Draft 7+；不支持自定义类型。
动作（Action）	模型生成的结构化调用指令，包含 tool 名称和 args 参数对象。	动作必须对应上下文中声明的某个工具；参数需符合其 schema。
动作格式	{ "tool": "tool_name", "args": { ... } }	模型输出必须能被 JSON.parse() 解析；禁止自由文本混合。

示例工具定义：

{
  "name": "calculate",
  "description": "执行简单数学计算",
  "parameters": {
    "type": "object",
    "properties": {
      "expression": { "type": "string", "description": "如 '2 + 3 * 4'" }
    },
    "required": ["expression"]
  }
}

示例动作：

{ "tool": "calculate", "args": { "expression": "sqrt(16)" } }

2.3 请求（Request）与响应（Response）格式规范

消息类型	字段	说明	注意事项
Request（请求）	context	完整的上下文对象（见 2.1 节）	必须包含 user_messages 和 tools；其他字段可选
Request	model	可选，指定使用的 LLM 名称（如 “llama3-8b”）	仅当服务端支持多模型路由时使用
Request	max_actions	可选整数，限制单次响应最多生成的动作数	默认为 1；设为 0 表示仅生成文本回复
Response（响应）	actions	动作对象数组，每个含 tool 和 args	若为空数组，表示模型选择不调用工具，仅回复文本
Response	reply	可选字符串，模型生成的自然语言回复	当 actions 非空时，reply 通常为空或为中间思考
Response	error	可选，若解析失败则包含错误信息	仅在协议层错误时出现（如 JSON 无效）

典型请求示例：

{
  "context": { },
  "max_actions": 1
}

典型响应示例（工具调用）：

{
  "actions": [
    { "tool": "get_weather", "args": { "city": "北京" } }
  ]
}

典型响应示例（纯文本）：

{
  "actions": [],
  "reply": "您好！请问您想查询哪个城市的天气？"
}

2.4 会话状态（Session State）管理

概念名称	说明	注意事项
会话（Session）	一次用户与 LLM 代理的完整交互过程，由多个 Request/Response 轮次组成。	会话应有唯一 ID（session_id），用于关联上下文。
状态持久化	客户端或服务端需保存完整的上下文（含 action_history）以支持多轮对话。	禁止仅保存部分消息；否则模型可能”遗忘”已执行动作。
上下文裁剪策略	当上下文接近模型 token 限制时，需移除早期无关消息或摘要历史。	裁剪时应保留 tool definitions 和最近 action_history；避免破坏工具调用链。
状态重置	用户发起新任务时，应创建全新上下文，清空 action_history 和 user_messages。	重置不等于重启服务，仅初始化会话数据结构。
并发会话支持	单个服务可同时处理多个 session_id 对应的会话，彼此隔离。	需确保 session_id 全局唯一；建议使用 UUID。

操作建议：

• 每次收到模型响应后，将执行结果追加到 action_history；
• 下一轮请求前，更新 user_messages（加入用户新输入）；
• 始终携带完整的 tools 列表（即使未变化），因模型可能依赖其描述做决策。

第三章：MCP 协议规范详解

3.1 MCP 消息结构（JSON Schema）

字段路径	类型	必填	说明	示例值	注意事项
$.context	object	是	完整会话上下文（见第二章）	{ "user_messages": [...], "tools": [...] }	必须包含 user_messages 和 tools
$.context.user_messages	array	是	用户与助手的历史消息列表	[ { "role": "user", "content": "查天气" } ]	role 通常为 “user” 或 “assistant”；不建议混入 tool 结果
$.context.tools	array	是	当前可用工具定义列表	[ { "name": "get_weather", ... } ]	每个工具必须含 name、description、parameters
$.context.action_history	array	否	已执行动作及结果记录	[ { "action": { ... }, "result": { "temp": 25 } } ]	result 应为 JSON 对象或字符串；避免二进制数据
$.context.metadata	object	否	会话元信息	{ "session_id": "abc123" }	不参与模型推理，仅用于追踪
$.model	string	否	指定 LLM 模型名称	”llama3-8b-instruct”	服务端可忽略此字段
$.max_actions	integer	否	单次响应最大动作数	2	默认为 1；0 表示禁止工具调用
$.response_format	string	否	响应格式偏好（预留）	“mcp_v1”	当前版本固定为 MCP v1，可省略

完整请求 JSON Schema（简化版）：

{
  "type": "object",
  "properties": {
    "context": {
      "type": "object",
      "properties": {
        "user_messages": { "type": "array", "items": { "type": "object" } },
        "tools": { "type": "array", "items": { "type": "object" } },
        "action_history": { "type": "array", "items": { "type": "object" } },
        "metadata": { "type": "object" }
      },
      "required": ["user_messages", "tools"]
    },
    "model": { "type": "string" },
    "max_actions": { "type": "integer", "minimum": 0 }
  },
  "required": ["context"]
}

3.2 工具注册与发现机制

操作名称	操作细节	注意事项
静态注册（推荐）	工具在每次 MCP 请求的 context.tools 字段中显式声明。服务端无需维护全局工具池。	所有工具必须随上下文传递；适用于无状态服务架构。
动态注册（扩展）	客户端先调用 /register-tools 接口注册工具集，后续请求通过 tool_set_id 引用。	需服务端支持状态管理；增加系统复杂度，仅用于高频复用场景。
工具发现	模型仅能调用 context.tools 中列出的工具；未声明的工具不可见。	禁止模型”猜测”工具名；确保安全性与可控性。
工具版本控制	工具可通过 name 包含版本（如 “search_v2”），或在 parameters 中声明兼容性。	避免在生产环境频繁变更工具接口；建议向后兼容。
工具描述优化	description 应包含使用场景、参数含义、返回示例。	例如：“查询实时天气，city 为中文城市名，返回温度与天气状况。”

工具注册示例（context.tools 片段）：

[
  {
    "name": "send_email",
    "description": "向指定邮箱发送邮件，主题和正文需明确",
    "parameters": {
      "type": "object",
      "properties": {
        "to": { "type": "string", "format": "email" },
        "subject": { "type": "string" },
        "body": { "type": "string" }
      },
      "required": ["to", "subject", "body"]
    }
  }
]

3.3 同步与异步调用模式

调用模式	操作细节	注意事项
同步调用	客户端发送请求后阻塞等待，直到服务端返回完整响应（含 actions 或 reply）。	适用于低延迟、短时任务（如计算、简单查询）；超时时间建议 ≤30 秒。
异步调用	客户端提交请求后立即获得 task_id，通过轮询 /task/{id}/status 获取结果。	适用于长耗时操作（如视频处理、批量分析）；需实现任务队列与状态存储。
流式响应（预留）	服务端逐步返回部分动作或思考过程（类似 SSE）。	当前 MCP v1 不强制支持；若实现需定义流格式（如 NDJSON）。
混合模式	单次响应中部分动作为同步（立即执行），部分标记为 async: true（后台执行）。	需在 action 中增加 metadata 字段标识；客户端需分别处理。
超时处理	同步调用超时时，服务端应返回 error；异步任务应记录失败状态。	超时不应导致资源泄漏；建议设置最大执行时间（如 5 分钟）。

同步请求流程：

客户端 → 发送完整 context → 等待 → 收到 actions → 执行工具 → 更新上下文 → 下一轮请求

异步请求流程：

客户端 → POST /tasks → 获得 task_id → 轮询 GET /tasks/{id} → 获取最终 context 更新

3.4 错误处理与状态码

错误类型	HTTP 状态码	错误代码（code）	说明	响应体示例	注意事项
无效 JSON	400	invalid_json	请求体无法解析为 JSON	{ "error": { "code": "invalid_json", "message": "Unexpected end of input" } }	客户端应校验 JSON 有效性后再发送
上下文缺失字段	400	missing_field	context 缺少 user_messages 或 tools	{ "error": { "code": "missing_field", "field": "context.tools" } }	必填字段必须存在且类型正确
工具参数无效	400	invalid_tool_args	模型生成的 action.args 不符合工具 schema	{ "error": { "code": "invalid_tool_args", "tool": "get_weather", "detail": "city is required" } }	此错误通常由模型输出错误引起，需优化 prompt 或微调
模型推理失败	500	model_error	LLM 推理过程中崩溃或超时	{ "error": { "code": "model_error", "message": "Inference timeout" } }	服务端应记录日志；客户端可重试
工具执行异常	200（业务错误）	tool_execution_failed	工具调用成功但执行失败（如 API 限流）	{ "actions": [ { "tool": "...", "args": {...} } ], "action_results": [ { "error": "Rate limit exceeded" } ] }	工具错误应在 action_history.result 中返回，非协议层错误
未知错误	500	internal_error	服务端内部未预期异常	{ "error": { "code": "internal_error" } }	应包含 request_id 便于追踪

通用错误响应格式：

{
  "error": {
    "code": "string",
    "message": "string",
    "request_id": "uuid"
  }
}

注意事项：

• 协议层错误（4xx/5xx）表示请求无法被处理；
• 工具执行失败属于业务逻辑，应通过 action_history 返回，保持 200 状态；
• 所有错误应包含可读 message 和唯一 request_id 用于调试。

第四章：MCP 客户端开发

4.1 构建 MCP 兼容的客户端

步骤名称	操作细节	注意事项
选择编程语言	可使用任意支持 HTTP 和 JSON 的语言（如 Python、JavaScript、Go、Rust）。	推荐使用具备成熟 HTTP 客户端库的语言（如 Python 的 requests、JS 的 fetch）。
实现上下文管理器	设计 Context 类或结构体，用于维护 user_messages、tools、action_history 等字段。	上下文应可序列化为 JSON；避免在内存中丢失历史状态。
集成工具注册机制	提供 register_tool(name, description, schema) 方法，将工具加入 context.tools。	工具参数必须符合 JSON Schema Draft 7+；建议运行时校验 schema 合法性。
封装 MCP 请求方法	实现 send_mcp_request(context, endpoint, model=None, max_actions=1) 函数。	自动处理 JSON 序列化、HTTP 头（Content-Type: application/json）、超时设置。
支持会话隔离	每个用户会话应有独立的上下文实例，通过 session_id 区分。	多线程/异步环境下需确保上下文线程安全。
日志与调试支持	记录每次发送的请求和接收的响应（可脱敏）。	调试模式下可打印完整 JSON；生产环境应控制日志体积。

最小可运行客户端骨架（Python 伪代码）：

class MCPClient:
    def __init__(self, endpoint):
        self.endpoint = endpoint
        self.context = {
            "user_messages": [],
            "tools": [],
            "action_history": []
        }

    def register_tool(self, name, desc, params_schema):
        self.context["tools"].append({
            "name": name,
            "description": desc,
            "parameters": params_schema
        })

    def add_user_message(self, content):
        self.context["user_messages"].append({"role": "user", "content": content})

4.2 发送上下文请求

方法名称	语法	用途	代码示例	注意事项
send_request	POST /mcp/v1/invoke	向 MCP 服务端发送完整上下文以获取模型决策	import requests resp = requests.post( "http://localhost:8000/mcp/v1/invoke", json={ "context": client.context, "max_actions": 1 }, timeout=30 )	设置合理超时（建议 10–30 秒）；检查 resp.status_code == 200
构建请求体	构造符合 3.1 节规范的 JSON 对象	确保字段完整、类型正确	request_body = { "context": current_context, "model": "llama3-8b", "max_actions": 2 }	不要遗漏 tools 字段；即使未变化也需重新发送
处理网络异常	捕获连接错误、超时、DNS 失败等	提升客户端健壮性	try: resp = requests.post(...) except requests.Timeout: print("Request timed out")	应实现重试机制（如指数退避），但限制最大重试次数（≤3）
验证响应格式	检查返回 JSON 是否含 actions 或 error	防止后续解析崩溃	data = resp.json() if "error" in data: raise Exception(data["error"]["message"])	即使 HTTP 200，也可能含业务错误（如模型输出无效）

4.3 解析模型返回的动作指令

方法名称	语法	用途	代码示例	注意事项
parse_response	输入：HTTP 响应 JSON	提取模型生成的结构化动作	response = resp.json() actions = response.get("actions", []) reply = response.get("reply", "")	若 actions 非空，通常忽略 reply；反之则展示 reply 给用户
验证动作合法性	检查每个 action.tool 是否在 context.tools 中定义	防止模型”幻觉”调用未注册工具	tool_names = {t["name"] for t in context["tools"]} for a in actions: if a["tool"] not in tool_names: raise ValueError(f"Unknown tool: {a['tool']}")	必须校验；否则可能执行恶意或不存在的函数
校验参数合规性	使用 jsonschema 库验证 args 是否符合工具 parameters	确保参数类型、必填项正确	from jsonschema import validate tool_def = get_tool_by_name(a["tool"]) validate(instance=a["args"], schema=tool_def["parameters"])	参数校验失败应记录日志并终止执行，避免工具崩溃
处理空动作	actions 为空且 reply 存在	表示模型选择直接回复，无需工具调用	if not actions and reply: print("Assistant:", reply)	此为正常流程，常见于澄清问题或最终回答

4.4 执行本地工具并回传结果

方法名称	语法	用途	代码示例	注意事项
execute_tool	输入：action 对象	调用本地注册的函数并捕获结果	def execute_tool(action): func = TOOL_REGISTRY[action["tool"]] try: result = func(**action["args"]) return {"result": result} except Exception as e: return {"error": str(e)}	工具函数应返回可序列化对象；禁止返回文件句柄等非 JSON 类型
注册工具函数	将 Python/JS 函数映射到工具名	建立 tool name → callable 的映射	TOOL_REGISTRY = {} def register_impl(name, func): TOOL_REGISTRY[name] = func	工具实现与 MCP 客户端解耦；便于单元测试
更新上下文	将 action 和执行结果追加到 action_history	为下一轮推理提供完整历史	context["action_history"].append({ "action": action, "result": execution_result })	result 字段应包含成功值或 error 字符串，不可为空
回传至下一轮	在下次请求前，确保上下文已更新	支持多轮工具链式调用	client.add_user_message("继续") next_response = client.send_request()	不要清空 action_history；否则模型无法知道前序动作结果
安全沙箱（可选）	在受限环境中执行工具（如 Docker、subprocess）	防止恶意或危险操作	subprocess.run([...], timeout=10, capture_output=True)	对涉及系统命令、网络请求的工具强烈建议沙箱化

工具执行与上下文更新示例：

action = { "tool": "get_weather", "args": { "city": "上海" } }
result = execute_tool(action)  # 返回 { "result": { "temp": 18, "desc": "多云" } }
client.context["action_history"].append({
    "action": action,
    "result": result["result"] if "result" in result else result
})

第五章：MCP 服务端实现

5.1 搭建 MCP 服务端接口

方法/组件	语法/结构	用途	代码示例（FastAPI 伪代码）	注意事项
HTTP 路由定义	POST /mcp/v1/invoke	接收符合 MCP 规范的请求	from fastapi import FastAPI, Request app = FastAPI() @app.post("/mcp/v1/invoke") async def invoke_mcp(request: Request): body = await request.json() return process_mcp_request(body)	路径应版本化（如 /v1/）；支持 CORS 若需浏览器调用
请求验证中间件	使用 Pydantic 或 jsonschema 校验输入	确保请求符合 3.1 节 JSON Schema	class MCPRequest(BaseModel): context: dict max_actions: int = 1 model: Optional[str] = None	必须校验 context.user_messages 和 context.tools 是否存在且为数组
响应格式封装	返回标准 MCP 响应对象	统一输出结构，便于客户端解析	return { "actions": generated_actions, "reply": fallback_reply }	即使无动作也应返回 { "actions": [] }，不可省略字段
超时与限流	设置请求超时（如 30s）、并发限制	防止资源耗尽	使用 uvicorn workers + timeout-keep-alive；或集成 slowapi 限流	推理耗时长时，建议改用异步任务模式（见 3.3 节）
健康检查端点	GET /health	用于监控与部署就绪探针	@app.get("/health") def health(): return {"status": "ok"}	Kubernetes 或 Docker 部署时必需

最小可运行服务端骨架：

from fastapi import FastAPI
import json

app = FastAPI()

@app.post("/mcp/v1/invoke")
async def mcp_invoke(request: dict):
    # 此处应校验 request 结构
    context = request["context"]
    # 调用 LLM 引擎生成动作（见 5.4）
    actions = call_llm_engine(context)
    return {"actions": actions}

5.2 注册可用工具集

概念/操作	说明	注意事项
工具集来源	MCP 服务端不主动维护全局工具池；工具由客户端在 context.tools 中声明	服务端仅解析和传递工具定义，不存储；符合无状态设计原则
工具元数据缓存（可选）	若需优化性能，可对重复工具定义做哈希缓存（如 name + schema hash）	仅用于日志或审计，不影响核心逻辑；避免内存泄漏
工具合法性校验	服务端应验证每个工具的 parameters 是否为合法 JSON Schema	使用 jsonschema.Draft7Validator.check_schema(schema) 提前校验
工具描述增强（可选）	服务端可注入额外元信息（如权限标签），但不得修改原始定义	例如：tool["_internal"] = {"requires_auth": True}
动态工具注册（高级）	通过 /tools/register 接口预注册工具，后续请求引用 tool_set_id	仅适用于有状态服务；增加复杂度，不推荐初学者使用

工具校验示例（服务端）：

from jsonschema import Draft7Validator

def validate_tool(tool_def):
    if not isinstance(tool_def, dict):
        raise ValueError("Tool must be object")
    if "name" not in tool_def or not isinstance(tool_def["name"], str):
        raise ValueError("Missing valid 'name'")
    try:
        Draft7Validator.check_schema(tool_def.get("parameters", {}))
    except Exception as e:
        raise ValueError(f"Invalid parameters schema: {e}")

5.3 处理多轮对话上下文

操作名称	操作细节	注意事项
上下文透传	服务端不修改 context，仅将其传递给 LLM 引擎	严禁服务端”记忆”历史；所有状态由客户端维护
上下文长度监控	在送入 LLM 前估算 token 数，必要时触发裁剪	使用 tiktoken 或 llama-cpp tokenizer 计算长度
自动摘要（可选）	对过长 action_history 或 user_messages 生成摘要	调用小型 LLM 生成：“此前已查询北京、上海天气…”
工具结果注入	将客户端回传的 action_results 视为普通消息追加？	否！action_history 已由客户端完整提供，服务端无需处理
会话隔离保障	利用无状态设计天然隔离；无需 session store	每个请求自包含上下文，天然支持并发多用户

上下文处理流程：

a. 接收完整 context（含完整 history）
b. （可选）校验长度 → 裁剪或报错
c. 构造 prompt（见 5.4）
d. 调用 LLM
e. 解析输出为 actions
f. 返回 { "actions": [...] }

5.4 与 LLM 推理引擎集成（如 vLLM、Ollama、Llama.cpp）

推理引擎	集成方式	用途	代码示例（调用逻辑）	注意事项
vLLM	通过 OpenAI 兼容 API 或 direct Python API	高吞吐、低延迟的生产级推理	from vllm import LLM, SamplingParams llm = LLM(model="meta-llama/Llama-3-8B-Instruct") outputs = llm.generate(prompts, sampling_params)	需 GPU；支持张量并行；prompt 需按 chat template 格式化
Ollama	调用本地 ollama serve 的 REST API	快速本地原型开发	resp = requests.post("http://localhost:11434/api/generate", json={"model": "llama3", "prompt": full_prompt})	默认非结构化输出；需用 regex 或 JSON mode 提取动作
Llama.cpp	通过 llama-cpp-python 绑定	CPU 推理，轻量部署	from llama_cpp import Llama llm = Llama(model_path="llama-3-8b.Q4_K_M.gguf") output = llm(full_prompt, max_tokens=200, stop=["\n"])	启用 —grammar 可强制 JSON 输出；需预编译 grammar 文件
Transformers (HuggingFace)	直接加载 AutoModelForCausalLM	灵活微调与实验	from transformers import pipeline pipe = pipeline("text-generation", model="...") result = pipe(prompt, max_new_tokens=150)	推理速度慢；适合研究，不适合高并发
启用结构化输出	使用 JSON mode / Grammar / Guided Decoding	确保模型输出合规动作	vLLM: SamplingParams(guided_decoding={"type": "json", "schema": ACTION_SCHEMA})	必须启用！否则模型输出自由文本无法解析为 actions

Prompt 构造示例（Llama-3 Instruct 格式）：

full_prompt = (
    "<|begin_of_text|>"
    "<|start_header_id|>system<|end_header_id|>\n\n"
    "你是一个智能代理，可调用以下工具：\n"
    + json.dumps(context["tools"], ensure_ascii=False) +
    "\n请根据上下文生成一个动作，格式为 JSON：{ tool: ..., args: {...} }"
    "<|eot_id|>"
    + "".join([
        f"<|start_header_id|>{msg['role']}<|end_header_id|>\n\n{msg['content']}<|eot_id|>"
        for msg in context["user_messages"]
    ])
    + "<|start_header_id|>assistant<|end_header_id|>\n\n"
)

第六章：MCP 实战案例

6.1 案例一：智能客服中的工具调用

要素	说明	注意事项
应用目标	用户询问订单状态、退换货政策等，模型调用内部 API 获取实时信息并生成自然语言回复	需保护用户隐私，禁止暴露原始 API 响应
工具定义	- get_order_status(order_id: str) → { status: str, delivery_date: str } - check_return_policy(product_type: str) → { allowed: bool, days: int }	参数需校验格式（如 order_id 为 12 位数字）；返回值应脱敏
上下文初始化	user_messages: [ { "role": "user", "content": "我的订单 ABC123 到哪了？" } ] tools: [ get_order_status, check_return_policy ]	客户端在首次请求前注册所需工具
模型动作输出	{ "tool": "get_order_status", "args": { "order_id": "ABC123" } }	若订单不存在，工具返回 { "error": "Order not found" }
客户端执行与回传	执行 get_order_status → 返回 { "status": "已发货", "delivery_date": "2026-03-15" } 追加到 action_history 发起第二轮请求（可选）	第二轮可让模型生成最终回复：“您的订单已发货，预计 3 月 15 日送达。“
最终响应	actions: [] reply: "您的订单 ABC123 已发货，预计 2026-03-15 送达。"	reply 由模型在第二轮生成；也可在客户端拼接
安全控制	工具函数验证用户身份（如从 metadata.user_id 匹配订单归属）	禁止未授权查询他人订单；建议在工具实现层校验

关键客户端逻辑（伪代码）：

client.register_tool("get_order_status", "...", {
    "type": "object",
    "properties": { "order_id": { "type": "string" } },
    "required": ["order_id"]
})
client.add_user_message("我的订单 ABC123 到哪了？")
resp1 = client.send_request()
if resp1["actions"]:
    result = execute_tool(resp1["actions"][0])
    client.context["action_history"].append({
        "action": resp1["actions"][0],
        "result": result
    })
    # 可选：自动发起第二轮获取自然语言回复
    client.add_user_message("请根据以上信息回答用户")
    resp2 = client.send_request()
    print(resp2.get("reply", ""))

6.2 案例二：自动化数据查询代理

要素	说明	注意事项
应用目标	用户用自然语言提问（如”上季度销售额最高的产品？”），模型调用 SQL 查询工具并返回结构化结果	需防止 SQL 注入；限制查询范围
工具定义	run_sql_query(query: str) → { columns: […], rows: […] }	禁止直接暴露 SQL 工具！应封装为语义化工具（如 get_sales_top_n）
推荐工具设计	- get_sales_top_n(n: int, period: str) - get_customer_count(region: str)	参数枚举化（period ∈ [“last_month”, “last_quarter”]）
上下文示例	user: “上季度销售额最高的产品？“ tools: [ get_sales_top_n ]	模型应选择 get_sales_top_n(n=1, period="last_quarter")
模型动作	{ "tool": "get_sales_top_n", "args": { "n": 1, "period": "last_quarter" } }	若用户问”前五名”，则 n=5
工具执行	调用内部 BI 系统，返回 { "products": [ { "name": "手机X", "sales": 1200000 } ] }	结果应限制字段数量，避免泄露敏感维度
最终回复生成	模型第二轮输入包含 action_history.result，输出 reply	可在单轮完成（若 max_actions=0 且上下文含结果），但推荐两轮更可靠
性能优化	对高频查询结果缓存（如 Redis），避免重复查数仓	缓存键 = tool_name + sorted(args.items())

安全工具实现示例（Python）：

def get_sales_top_n(n: int, period: str):
    if period not in ["last_month", "last_quarter"]:
        raise ValueError("Invalid period")
    if n > 10:
        n = 10  # 限制最大返回条数
    # 构造预定义 SQL，非拼接
    sql = f"SELECT product, SUM(sales) ... WHERE period = '{period}' ORDER BY sales DESC LIMIT {n}"
    return db.query(sql)

6.3 案例三：多智能体协作系统

要素	说明	注意事项
应用目标	多个 MCP 客户端（智能体）协作完成复杂任务，如”策划一场线上活动”	各智能体职责分离（策划、设计、通知）
智能体角色	- PlannerAgent：分解任务 - DesignerAgent：生成海报 - NotifierAgent：发送邮件/消息	每个智能体运行独立 MCP 客户端，共享全局 session_id
协作机制	1. 用户请求发给 Planner 2. Planner 输出子任务（如 { "assign": "design", "task": "做海报" }） 3. 主控器路由到 DesignerAgent 4. 结果汇总后交 Notifier	需主控协调器（Orchestrator）管理 agent 路由与上下文合并
工具设计	每个 agent 有专属工具集： - Planner: decompose_task() - Designer: generate_poster(prompt) - Notifier: send_email(to, subject, body)	工具不可跨 agent 调用；通过上下文传递中间结果
上下文共享	全局 context.action_history 记录所有 agent 的动作与结果	主控器在调用下一 agent 前，将前序结果注入其 user_messages
示例流程	user: “为新品发布会做宣传” → Planner: assign design task → Designer: generate_poster(“AI 新品…”) → 返回 image_url → Planner: assign notify task with image_url → Notifier: send_email(…, body=“见海报: image_url”)	每步需验证前序结果有效性（如 image_url 是否生成成功）
错误传播	若 Designer 失败，Planner 应收到 error 并重试或降级	主控器需捕获异常并更新 action_history.result.error
通信协议	所有 agent 与主控器之间仍使用标准 MCP 协议	不引入新协议；保持兼容性

主控器伪代码：

global_context = init_context(user_request)
planner = MCPClient(planner_endpoint)
planner.context = global_context
resp = planner.send_request()
for action in resp["actions"]:
    if action["tool"] == "assign":
        target_agent = action["args"]["agent"]
        sub_task = action["args"]["task"]
        # 构造新上下文给目标 agent
        agent_context = {
            "user_messages": [{"role": "user", "content": sub_task}],
            "tools": AGENT_TOOLS[target_agent],
            "action_history": global_context["action_history"],  # 继承历史
            "metadata": global_context["metadata"]
        }
        agent_client = MCPClient(ENDPOINTS[target_agent])
        agent_client.context = agent_context
        agent_resp = agent_client.send_request()
        # 执行本地工具（如调用 Designer 服务）
        result = execute_local_tool(agent_resp)
        # 更新全局历史
        global_context["action_history"].append({
            "action": action,
            "result": result
        })

第七章：MCP 生态与扩展

7.1 与 LangChain / LlamaIndex 的集成

集成目标	操作细节	注意事项
将 MCP 作为 LangChain 的 Tool 执行后端	在 LangChain 中定义一个 MCPTool 类，其 _run 方法封装 MCP 客户端调用	LangChain 负责链式编排，MCP 负责结构化工具调用；避免双重工具注册
使用 LangChain 构造上下文并触发 MCP 请求	将 LangChain 的中间步骤（如 Agent scratchpad）转换为 MCP context.user_messages	需将 LangChain 的”思考-行动”历史映射为 MCP 的 action_history
利用 LlamaIndex 查询引擎作为 MCP 工具	将 LlamaIndex 的 query_engine.query() 封装为一个 MCP 工具（如 “rag_search”）	工具参数应包含 query 字符串；返回结果需结构化（如 { "answer": "...", "sources": [...] }）
双向桥接模式	LangChain → MCP：LangChain Agent 决策后调用 MCP 工具 MCP → LangChain：MCP 动作结果作为 LangChain 的 Observation 输入	推荐单向集成（MCP 为主或 LangChain 为主），双向易导致状态混乱

示例：LangChain 调用 MCP 工具：

from langchain.tools import BaseTool

class MCPWeatherTool(BaseTool):
    name = "get_weather"
    description = "查询城市天气"

    def _run(self, city: str):
        mcp_client = get_global_mcp_client()
        mcp_client.add_user_message(f"查{city}天气")
        resp = mcp_client.send_request()
        # 执行动作并返回结果
        result = execute_action(resp["actions"][0])
        return result["temp"]

关键原则：

• MCP 提供标准化协议，LangChain/LlamaIndex 提供高级编排
• 不重复实现工具逻辑；优先将现有 LangChain 工具”MCP 化”

7.2 自定义工具开发规范

规范项	要求	注意事项
工具命名	使用 snake_case，语义明确，如 send_email、calculate_tax	避免缩写（如 calc）；禁止使用保留字
参数设计	必须使用 JSON Schema 描述；参数应原子化（避免嵌套过深）	例如：用 start_date 和 end_date 而非 date_range: { start, end }
输入校验	工具函数内部必须校验参数合法性（类型、范围、格式）	如 email 字段需正则校验；数值需检查边界
输出结构	返回值必须为 JSON 序列化对象，禁止异常抛出到 MCP 层	错误应捕获并返回 { "error": "message" }，而非 raise Exception
幂等性	相同输入应产生相同输出（或可预期副作用）	如 create_user 非幂等，应避免；改用 get_or_create_user
副作用控制	高风险操作（如删除、支付）需二次确认或权限检查	建议在工具描述中注明：“此操作不可逆”
文档完整性	description 必须包含：功能说明、参数含义、返回示例、限制条件	示例：“查询股票价格。参数 symbol: 股票代码（如 AAPL）。返回 { price: float, currency: str }。仅支持美股。“
测试覆盖	每个工具需有单元测试，覆盖正常、边界、错误场景	使用 pytest 或 unittest；模拟外部依赖

工具模板（Python）：

def my_tool(param1: str, param2: int):
    # 1. 校验输入
    if not isinstance(param1, str) or len(param1) == 0:
        return {"error": "param1 must be non-empty string"}
    if param2 < 0:
        return {"error": "param2 must be >= 0"}

    # 2. 执行逻辑（模拟）
    try:
        result = external_api.call(param1, param2)
        return {
            "status": "success",
            "data": result
        }
    except Exception as e:
        return {"error": f"API failed: {str(e)}"}

对应的 MCP 工具定义：

{
  "name": "my_tool",
  "description": "调用外部 API 获取数据。param1: 标识符；param2: 数量（≥0）。返回成功状态或错误信息。",
  "parameters": {
    "type": "object",
    "properties": {
      "param1": { "type": "string", "minLength": 1 },
      "param2": { "type": "integer", "minimum": 0 }
    },
    "required": ["param1", "param2"]
  }
}

7.3 安全性与权限控制

安全维度	控制措施	注意事项
工具调用授权	在工具执行前检查 metadata.user_id 是否有权限	权限策略可硬编码或对接 IAM 系统（如 OAuth scopes）
输入净化	对所有工具参数进行白名单校验，拒绝特殊字符（如 SQL/Shell 元字符）	即使使用 ORM，也应做前端校验；防御 SSRF、命令注入
上下文隔离	确保不同用户的 context 不交叉（无状态服务天然满足）	若使用缓存，key 必须包含 session_id 或 user_id
敏感信息过滤	工具返回结果中移除密码、token、身份证等字段	可在 execute_tool 后增加 sanitize_result(result) 步骤
速率限制	对单个 session_id 或 IP 限制 MCP 请求频率（如 10 次/分钟）	使用令牌桶算法；通过 Redis 实现分布式限流
审计日志	记录每次工具调用：user_id、tool_name、args（脱敏）、timestamp	日志应不可篡改；保留至少 180 天以满足合规
沙箱执行	高风险工具在 Docker 容器或 subprocess 中运行	设置资源限制（CPU、内存、网络）；禁用文件系统写入
模型输出防护	即使模型生成非法动作（如调用未注册工具），客户端也应拒绝执行	客户端必须校验 tool name 是否在注册列表中

权限检查示例（工具层）：

def delete_file(file_path: str, user_id: str):
    # 1. 检查用户是否有权访问该路径
    if not is_authorized(user_id, file_path):
        return {"error": "Permission denied"}
    # 2. 限制路径范围（防止 ../ 攻击）
    if not file_path.startswith("/user_data/"):
        return {"error": "Invalid path"}
    # 3. 执行删除
    os.remove(file_path)
    return {"status": "deleted"}

7.4 性能优化与批处理支持

优化方向	方法	注意事项
上下文压缩	对长 history 使用滑动窗口或摘要（如保留最近 5 轮 + 工具定义）	摘要由小型 LLM 生成；避免丢失关键 action 结果
工具缓存	对幂等工具（如 get_weather）缓存结果，键 = tool_name + sorted(args.items())	设置 TTL（如 5 分钟）；高变数据（股价）不缓存
并发工具执行	若 actions 包含多个独立工具，客户端并行执行（asyncio/goroutine）	依赖顺序的动作（如先登录再查询）不能并行
批处理请求	客户端聚合多个用户请求，一次性发送给服务端（需服务端支持）	仅适用于无状态、独立会话；需修改服务端为 batch 模式
LLM 推理优化	使用 vLLM 的 continuous batching 或 PagedAttention	启用 —enable-chunked-prefill 提升吞吐
流式响应（未来）	服务端逐步返回动作（如 NDJSON 流），客户端渐进执行	当前 MCP v1 未强制支持；若实现需定义流格式
减少 JSON 开销	使用 msgpack 或 protobuf 替代 JSON（需双方协商）	牺牲可读性；仅在高吞吐场景考虑
预热与连接池	客户端复用 HTTP 连接；服务端预加载模型	避免每次请求新建 TCP 连接；使用 keep-alive

批处理示例（服务端扩展）：

请求格式（批量）：

{
  "batch": [
    { "context": ctx1, "max_actions": 1 },
    { "context": ctx2, "max_actions": 1 }
  ]
}

响应格式：

{
  "results": [
    { "actions": [...], "reply": "..." },
    { "actions": [...], "reply": "..." }
  ]
}

注意事项：

• 批处理会增加单次延迟，但提升整体吞吐
• 不同用户的上下文不得混合在同一 batch 中（除非明确允许）

第八章：MCP 最佳实践与调试技巧

8.1 上下文长度控制策略

策略名称	操作细节	注意事项
Token 预估	在发送请求前，使用 tokenizer（如 tiktoken、llama-cpp tokenizer）估算上下文总 token 数	必须包含 system prompt、user_messages、tools、action_history
滑动窗口裁剪	保留最近 N 轮消息，移除最早 user_message 和对应 action_history 条目	优先保留工具定义（tools）和最近 2–3 轮动作结果
关键信息摘要	对早期 action_history 生成摘要（如”此前已查询北京、上海天气”），替换原始条目	使用小型 LLM 或规则模板生成；避免丢失决策依据
工具定义复用标识	若 tools 未变化，服务端可支持 tool_set_id 引用，减少重复传输	客户端需缓存 tool_set_id 与 hash；仅适用于有状态扩展
分段推理（Plan-and-Execute）	将复杂任务拆分为多个独立子会话，避免单次上下文过长	子任务间通过主控器传递关键结果（如最终答案）
模型最大上下文对齐	确保客户端预估的 max_context_length ≤ 模型实际支持长度（如 Llama-3-8B 为 8192）	预留 500–1000 token 给模型生成动作，避免截断输出

上下文裁剪伪代码（Python）：

def trim_context(context, max_tokens=7000):
    tokenizer = get_tokenizer("llama3")
    current_tokens = count_tokens(context, tokenizer)
    if current_tokens <= max_tokens:
        return context

    # 优先保留 tools 和最近消息
    messages = context["user_messages"]
    history = context["action_history"]

    while current_tokens > max_tokens and len(messages) > 2:
        # 移除最早的一对消息和动作（假设一一对应）
        messages.pop(0)
        if history:
            history.pop(0)
        current_tokens = count_tokens({
            "user_messages": messages,
            "tools": context["tools"],
            "action_history": history
        }, tokenizer)

    return {
        "user_messages": messages,
        "tools": context["tools"],
        "action_history": history,
        "metadata": context.get("metadata", {})
    }

8.2 工具调用失败重试机制

重试场景	重试策略	注意事项
网络瞬时错误（如超时、5xx）	指数退避重试（1s → 2s → 4s），最多 3 次	仅对幂等工具重试；非幂等操作（如支付）禁止重试
工具参数校验失败	不重试；返回错误并记录日志	表明模型输出不符合 schema，需优化提示或微调
模型输出无法解析为动作	触发”澄清”流程：追加系统消息”请按 JSON 格式输出动作”并重试	限制澄清次数（≤2），防止死循环
外部 API 限流（429）	按 Retry-After 头延迟后重试；否则等待 5–10 秒	需全局限流计数器，避免雪崩
工具执行逻辑异常	记录错误，向模型反馈 result.error，由模型决定下一步	例如：模型收到”数据库连接失败”后可建议”稍后再试”
重试上下文更新	每次重试需将前次错误加入 action_history.result	确保模型知晓失败原因，避免重复相同动作

重试逻辑示例（客户端）：

def execute_with_retry(action, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            result = actual_tool_impl(action["args"])
            return {"result": result}
        except NetworkError as e:
            if attempt == max_retries:
                return {"error": f"Max retries exceeded: {e}"}
            wait_time = 2 ** attempt
            time.sleep(wait_time)
        except ValidationError as e:
            return {"error": f"Invalid args: {e}"}  # 不重试
        except Exception as e:
            return {"error": str(e)}  # 一般异常不重试

8.3 日志记录与可观测性

日志类型	记录内容	注意事项
请求/响应日志	完整 MCP 请求体、响应体（脱敏后）、timestamp、session_id、request_id	敏感字段（如密码、身份证）必须过滤；建议采样存储（如 10%）
工具执行日志	tool_name、args（脱敏）、执行耗时、结果状态（success/error）	用于性能分析和故障定位；结构化日志便于 ELK 查询
模型决策日志	模型原始输出、是否成功解析为动作、解析失败原因	用于评估模型行为；可结合 LLM tracing 工具（如 LangSmith）
错误追踪	所有异常堆栈、HTTP 状态码、用户上下文快照	关联 request_id 实现全链路追踪
指标监控	QPS、平均延迟、工具调用成功率、上下文平均长度	通过 Prometheus 暴露指标；设置告警阈值（如错误率 > 5%）
用户行为分析	匿名化统计高频工具、常见失败模式、多轮对话深度	用于产品优化；需用户同意（GDPR/CCPA 合规）

日志结构示例（JSONL）：

{
  "timestamp": "2026-03-10T17:20:00Z",
  "request_id": "req_abc123",
  "session_id": "sess_xyz789",
  "event": "mcp_request",
  "context_user_messages_count": 2,
  "tools_count": 3,
  "max_actions": 1
}

{
  "timestamp": "2026-03-10T17:20:02Z",
  "request_id": "req_abc123",
  "event": "tool_executed",
  "tool_name": "get_weather",
  "duration_ms": 245,
  "status": "success"
}

8.4 单元测试与模拟服务器

测试目标	测试方法	注意事项
工具函数逻辑	使用 pytest/unittest 直接调用工具函数，验证输入/输出	模拟外部依赖（如 mock.patch("requests.post")）
MCP 客户端行为	Mock HTTP 响应，验证 send_request、parse_response 逻辑	使用 responses（Python）或 fetch-mock（JS）
模型动作解析	提供典型模型输出（合规/非法），测试解析器鲁棒性	覆盖边界情况：空动作、多动作、schema 不匹配
上下文管理	验证 add_user_message、update_action_history 是否正确维护状态	检查序列化/反序列化后一致性
模拟 MCP 服务端	启动本地 FastAPI 测试服务器，返回预设响应	用于集成测试；支持动态响应（如根据 tool name 返回不同结果）
端到端流程测试	模拟完整用户对话：提问 → 工具调用 → 回复	使用 pytest-asyncio 测试异步客户端

模拟服务器示例（FastAPI）：

from fastapi import FastAPI

app = FastAPI()

@app.post("/mcp/v1/invoke")
async def mock_mcp(request: dict):
    tool_name = request["context"]["tools"][0]["name"]
    if tool_name == "get_weather":
        return {
            "actions": [
                { "tool": "get_weather", "args": { "city": "北京" } }
            ]
        }
    elif tool_name == "calculate":
        return { "actions": [], "reply": "结果是 42" }
    else:
        return { "actions": [] }

单元测试示例（Python）：

def test_weather_tool_call():
    client = MCPClient("http://localhost:8001")
    client.register_tool("get_weather", "...", {...})
    client.add_user_message("北京天气？")

    with responses.RequestsMock() as rsps:
        rsps.add(
            rsps.POST,
            "http://localhost:8001/mcp/v1/invoke",
            json={ "actions": [ { "tool": "get_weather", "args": { "city": "北京" } } ] }
        )
        resp = client.send_request()
        assert len(resp["actions"]) == 1
        assert resp["actions"][0]["tool"] == "get_weather"