工具集成与 MCP 协议
本文档系统阐述 AI Agent 的工具集成体系与 MCP（Model Context Protocol）协议，涵盖从协议架构、Server/Client 开发、Schema 设计、安全权限、错s误处理到传输层、版本管理与完整工具生态的全链路技术知识
1. MCP 概念与架构
1.1 定义
MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 提出的开放标准协议，用于规范 AI 模型（LLM）与外部工具、资源、数据源之间的连接方式。MCP 的目标是将"AI 如何调用工具"这一过程标准化，使任意 Host（宿主应用）能够通过统一的 Client-Server 架构接入任意第三方工具，避免为每个工具编写定制化集成代码
1.2 核心问题：为什么需要 MCP
在 MCP 出现之前，工具集成面临三大痛点：
痛点1: M x N 集成爆炸
M 个 AI 应用 x N 个工具 = M x N 套定制集成代码
每新增一个工具，所有应用都要适配
痛点2: 接口不统一
应用A 用 OpenAI Function Calling 格式
应用B 用自定义 JSON 格式
痛点3: 能力发现缺失
应用无法动态发现工具提供哪些能力
MCP 的解法: M + N 集成模式
M 个 Host 各实现 1 个 MCP Client
N 个工具各实现 1 个 MCP Server
集成复杂度从 O(M*N) 降为 O(M+N)1.3 三层架构：Host / Client / Server
┌─────────────────────────────────────────────────────────┐
│ Host（宿主） │
│ 运行 Agent 的应用程序 │
│ 例如: Claude Desktop, VS Code, 自研 Agent 平台 │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ MCP Client A│ │ MCP Client B│ │ MCP Client C│ │
│ │ (连接搜索 )│ │ (连接数据库 )│ │ (连接文件系统)│ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
└──────────┼────────────────┼────────────────┼───────────┘
│ JSON-RPC 2.0 │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│MCP Server│ │MCP Server│ │MCP Server│
│ Tools │ │ Tools │ │ Tools │
│ Resources│ │ Resources│ │ Resources│
│ Prompts │ │ Prompts │ │ Prompts │
└──────────┘ └──────────┘ └──────────┘
独立进程 独立进程 独立进程1.4 三种资源类型
┌──────────────┬───────────────────────────────────────────┐
│ Tools（工具）│ 可执行的函数,Client 调用,Server 执行返回 │
│ │ 例: search("query"), calculate("1+1") │
├──────────────┼───────────────────────────────────────────┤
│Resources(资源)│ 可读取的数据,Client 通过 URI 读取 │
│ │ 例: file:///data/report.pdf │
├──────────────┼───────────────────────────────────────────┤
│Prompts(提示) │ 预定义的提示模板,Client 获取并填充参数 │
│ │ 例: code_review(language, code) │
└──────────────┴───────────────────────────────────────────┘1.5 JSON-RPC 2.0 通信协议
请求消息：
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": { "name": "search", "arguments": { "query": "LangGraph" } }
}成功响应：
{
"jsonrpc": "2.0",
"id": 1,
"result": { "content": [{ "type": "text", "text": "LangGraph 是 Agent 编排框架" }] }
}错误响应：
{
"jsonrpc": "2.0",
"id": 1,
"error": { "code": -32602, "message": "Invalid params: missing 'query'" }
}核心方法：
初始化: initialize / initialized
工具: tools/list -> 列出工具; tools/call -> 调用工具
资源: resources/list -> 列出资源; resources/read -> 读取资源
提示: prompts/list -> 列出提示; prompts/get -> 获取提示1.6 通信时序
Client Server
│ ── initialize ──────────────────>│ 协商能力
│ ── initialized ─────────────────>│
│ ── tools/list ──────────────────>│ 发现工具
│ <── tools list ─────────────────│
│ ── tools/call(search) ──────────>│ 执行工具
│ <── search result ──────────────│
│ ── resources/read(uri) ────────>│ 读取资源
│ <── resource content ───────────│1.7 与其他概念的关联
-> MCP Server 开发：Server 是协议的提供方
-> MCP Client 集成：Client 是协议的消费方
-> 工具 Schema 设计：工具定义采用 JSON Schema
-> MCP 传输层：JSON-RPC 消息通过 stdio 或 HTTP 传输
2. MCP Server 开发
2.1 定义
MCP Server 是工具能力的提供方，作为独立进程运行，通过 MCP 协议向 Client 暴露 Tools、Resources、Prompts 三种能力。核心职责：注册工具定义、接收调用请求、执行业务逻辑、返回结构化结果。
对应 Demo: demos/05_工具集成与MCP/02_MCP_Server开发.py
2.2 Server 核心组件
┌─────────────────────────────────────────────────────┐
│ MCP Server │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 工具注册表 │ │ 资源注册表 │ │ 提示注册表 │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ 请求路由器 (Router) │ │
│ │ tools/list -> 返回工具列表 │ │
│ │ tools/call -> 路由到对应处理函数 │ │
│ └──────────────────────┬───────────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ JSON-RPC 传输层 (stdio / HTTP) │ │
│ └──────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘2.3 使用 MCP SDK 开发 Server
import asyncio
from mcp import types
from mcp.server import Server
# 创建 Server 实例
server = Server("knowledge-server")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""列出 Server 提供的所有工具"""
return [
types.Tool(
name="search_knowledge",
description="搜索知识库",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"],
},
),
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""处理工具调用请求"""
if name == "search_knowledge":
query = arguments.get("query", "")
result = search_knowledge(query)
return [types.TextContent(type="text", text=result)]
raise ValueError(f"Unknown tool: {name}")2.4 工具定义三要素
工具定义三要素:
name -> Client 用于路由调用 (唯一标识, snake_case)
description -> LLM 用于判断"是否该用这个工具" (至关重要!)
inputSchema -> LLM 用于构造正确的调用参数 (JSON Schema)
description 写得好坏,直接决定 LLM 能否正确选择工具!2.5 工具处理函数模式
async def tool_handler(arguments: dict) -> list[types.TextContent]:
"""工具处理函数标准模式: 提取 -> 校验 -> 执行 -> 封装 -> 错误处理"""
expression = arguments.get("expression")
if not expression:
raise ValueError("缺少必要参数: expression")
try:
result = eval(expression, {"__builtins__": {}}, {})
except Exception as e:
return [types.TextContent(type="text", text=f"计算错误: {e}")]
return [types.TextContent(type="text", text=f"计算结果: {result}")]2.6 Server 启动与运行
from mcp.server.stdio import stdio_server
async def main():
"""启动 MCP Server,通过 stdio 传输通信"""
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
if __name__ == "__main__":
asyncio.run(main())2.7 与其他概念的关联
<- MCP 概念与架构：Server 是协议架构中的提供方
-> MCP Client 集成：Client 连接 Server 发现并调用工具
-> 工具 Schema 设计：工具定义使用 JSON Schema
-> MCP 传输层：Server 通过传输层与 Client 通信
3. MCP Client 集成
3.1 定义
MCP Client 是 Host 内部的协议桥梁，负责连接一个或多个 MCP Server，发现各 Server 提供的工具能力，并聚合后暴露给 Agent。核心职责：管理连接、发现工具、路由调用、聚合结果。
对应 Demo: demos/05_工具集成与MCP/03_MCP_Client集成.py
3.2 多 Server 架构
┌──────────────────────────────────────────────────────────┐
│ Agent (LLM + 决策) │
└──────────────────────┬───────────────────────────────────┘
│ 统一工具列表
▼
┌──────────────────────────────────────────────────────────┐
│ AgentToolManager（工具管理器） │
│ - 聚合所有 Server 的工具 │
│ - 自动路由: 根据工具名找到对应 Server │
│ - 去重: 处理多 Server 同名工具冲突 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Client A │ │ Client B │ │ Client C │ │
│ └─────┬────┘ └─────┬────┘ └─────┬────┘ │
└─────────┼─────────────┼─────────────┼───────────────────┘
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│搜索工具集│ │数据库工具│ │文件工具集│
└─────────┘ └─────────┘ └─────────┘3.3 工具发现与自动路由
class AgentToolManager:
"""Agent 工具管理器:聚合多 Server 工具,自动路由调用"""
def __init__(self):
self.clients: dict[str, MCPClient] = {}
self.tool_registry: dict[str, str] = {} # tool_name -> server_name
def discover_tools(self):
"""发现所有已连接 Server 的工具"""
all_tools = []
for server_name, client in self.clients.items():
tools = client.list_tools()
for tool in tools:
self.tool_registry[tool.name] = server_name
all_tools.append(tool)
return all_tools
async def call_tool(self, tool_name: str, arguments: dict):
"""自动路由:根据工具名找到对应 Server 并调用"""
if tool_name not in self.tool_registry:
raise KeyError(f"工具 '{tool_name}' 未注册")
server_name = self.tool_registry[tool_name]
client = self.clients[server_name]
return await client.call_tool(tool_name, arguments)3.4 工具冲突处理
冲突场景: Server A "search"(网络) vs Server B "search"(本地)
处理策略:
┌────────────────┬──────────────────────────────────┐
│ 命名空间前缀 │ 工具名改为 "web_search"/"file_search"│
│ (推荐) │ 消除歧义 │
├────────────────┼──────────────────────────────────┤
│ 优先级覆盖 │ 高优先级 Server 的工具覆盖低优先级 │
├────────────────┼──────────────────────────────────┤
│ 显式指定 │ "serverA.search"/"serverB.search" │
└────────────────┴──────────────────────────────────┘3.5 与其他概念的关联
<- MCP Server 开发：Client 连接 Server 获取能力
-> 工具 Schema 设计：Client 聚合各 Server 的工具 Schema
-> 工具错误处理：Client 需处理 Server 调用失败的情况
-> 常用工具集成：多 Server 集成各种常用工具
4. 工具 Schema 设计
4.1 定义
工具 Schema 是描述工具接口的元数据，采用 JSON Schema 格式定义工具的名称、描述、参数类型、参数约束等信息。Schema 是 LLM 理解工具能力的唯一途径。
对应 Demo: demos/05_工具集成与MCP/04_工具Schema设计.py
4.2 好 Schema vs 坏 Schema 对比
=== 坏 Schema ===
{
"name": "search",
"description": "搜索", <-- 太模糊!
"inputSchema": {
"properties": {
"q": { "type": "string" }, <-- 参数名不清晰,无描述
"n": { "type": "number" } <-- n 是什么?
}
}
}
问题: LLM 不知道何时用、怎么用
=== 好 Schema ===
{
"name": "search_web",
"description": "在互联网上搜索信息,返回相关网页摘要。
适用于需要查找最新资讯、事实核查的场景。
不适用于搜索本地文件或数据库。",
"inputSchema": {
"properties": {
"query": {
"type": "string",
"description": "搜索查询词,如 'Python 异步编程'"
},
"max_results": {
"type": "integer",
"description": "返回结果数量,1-20",
"minimum": 1, "maximum": 20, "default": 5
}
},
"required": ["query"]
}
}
优势: LLM 能准确判断何时用、如何构造参数4.3 Schema 设计六原则
┌──────────────┬─────────────────────────────────────────────┐
│ 1. 名称清晰 │ 动词+名词: search_web, query_database │
│ 2. 描述详尽 │ 做什么 + 何时用 + 何时不适用 + 示例 │
│ 3. 参数自解释 │ 每个参数有 description, 参数名语义化 │
│ 4. 约束明确 │ min/max/enum 限定取值, 标注 required │
│ 5. 合理默认值 │ 可选参数提供 default, 减少 LLM 填写量 │
│ 6. 类型精确 │ enum 代替 string(取值有限时), format 指定 │
└──────────────┴─────────────────────────────────────────────┘4.4 参数校验
import jsonschema
def validate_tool_args(schema: dict, arguments: dict) -> tuple[bool, str]:
"""使用 JSON Schema 校验工具参数"""
try:
jsonschema.validate(arguments, schema)
return True, "校验通过"
except jsonschema.ValidationError as e:
return False, f"参数校验失败: {e.message}"
# 合法: validate_tool_args(schema, {"query": "LangGraph", "limit": 10}) -> True
# 缺少必填: validate_tool_args(schema, {"limit": 10}) -> False
# 类型错误: validate_tool_args(schema, {"query": "test", "limit": "ten"}) -> False4.5 与其他概念的关联
<- MCP Server 开发：Schema 是工具注册时的核心元数据
-> 工具安全与权限：Schema 中的约束是安全的第一道防线
-> 工具错误处理：参数校验失败需返回清晰的错误信息
5. 工具安全与权限
5.1 定义
工具安全确保 Agent 调用外部工具时不造成数据泄露、系统破坏、越权访问等风险。权限系统根据工具风险等级，实施不同级别的审批、约束和防护。
对应 Demo: demos/05_工具集成与MCP/05_工具安全与权限.py
5.2 五级风险分类
┌─────────┬──────────┬──────────────────────┬─────────────────┐
│ Level 0 │ 无风险 │ calculate, format │ 无需审批 │
│ Level 1 │ 低风险 │ search, query_db │ 记录日志,自动执行│
│ Level 2 │ 中风险 │ read_file, http_get │ 首次需用户确认 │
│ Level 3 │ 高风险 │ write_file, db_update│ 每次需用户确认 │
│ Level 4 │ 极高风险 │ exec_cmd, run_code │ 确认+沙箱+审计 │
└─────────┴──────────┴──────────────────────┴─────────────────┘5.3 路径控制（防穿越攻击）
import os
class PathGuard:
"""路径安全守卫:防止路径穿越攻击"""
def __init__(self, allowed_roots: list[str]):
self.allowed_roots = [os.path.realpath(r) for r in allowed_roots]
def validate_path(self, path: str) -> tuple[bool, str]:
"""验证路径是否在允许范围内"""
real_path = os.path.realpath(path) # 解析 ../ 和符号链接
for root in self.allowed_roots:
if real_path.startswith(root + os.sep) or real_path == root:
return True, real_path
return False, f"路径 '{path}' 超出允许范围"
# 攻击: /data/workspace/../../../etc/passwd -> 解析为 /etc/passwd -> 拒绝
# 防护: 始终用 os.path.realpath() 解析后再校验5.4 SQL 注入防护
class SQLGuard:
"""SQL 安全守卫"""
DANGEROUS = ["DROP", "DELETE", "TRUNCATE", "ALTER", "INSERT",
"UPDATE", "CREATE", "GRANT", "REVOKE", "--", ";"]
def validate_sql(self, sql: str) -> tuple[bool, str]:
"""验证 SQL 安全性"""
sql_upper = sql.upper()
if not sql_upper.strip().startswith("SELECT"):
return False, "仅允许 SELECT 查询"
for pattern in self.DANGEROUS:
if pattern in sql_upper:
return False, f"检测到危险操作: {pattern}"
if sql.count(";") > 1:
return False, "禁止多语句执行"
return True, "SQL 校验通过"
# 正确: cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))
# 危险: cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")5.5 安全检查流程
Agent 请求调用工具
-> 1. 参数校验 (Schema)
-> 2. 风险评估 (确定等级)
-> 3. 权限检查 (角色/路径/SQL)
-> 4. 审批确认 (高危需 HITL)
-> 5. 沙箱执行 (Critical 级别)
-> 6. 审计记录 (调用详情追溯)5.6 与其他概念的关联
<- 工具 Schema 设计：Schema 约束是安全的第一道防线
-> 工具错误处理：安全检查失败属于一种错误场景
-> 完整工具生态：安全策略是工具生态的核心组件
6. 工具错误处理
6.1 定义
工具错误处理对调用外部工具时发生的各类错误进行分类处理，通过重试、超时、降级、幂等性等策略保障系统鲁棒性。
对应 Demo: demos/05_工具集成与MCP/06_工具错误处理.py
6.2 错误类型分类
┌─────────────────┬──────────────────────┬──────────────────┐
│ TRANSIENT │ 网络抖动、暂时不可用 │ 指数退避重试 │
│ VALIDATION │ 参数错误、缺失必填 │ 不重试,返回错误 │
│ PERMISSION │ 权限不足、路径越权 │ 不重试,请求人工 │
│ TIMEOUT │ 工具执行超时 │ 超时后重试或降级 │
│ NOT_FOUND │ 工具不存在、资源已删 │ 不重试,告知 LLM │
│ RATE_LIMIT │ 请求频率超限 │ 等待后重试 │
└─────────────────┴──────────────────────┴──────────────────┘6.3 重试策略：指数退避
async def retry_with_backoff(func, args, config: RetryConfig):
"""带指数退避的重试执行"""
for attempt in range(config.max_retries + 1):
try:
return await func(*args)
except ToolError as e:
if e.error_type not in config.retry_on:
raise # 不可重试的错误
if attempt == config.max_retries:
raise # 重试次数用尽
delay = min(config.base_delay * (2 ** attempt), config.max_delay)
if config.jitter:
delay *= (0.5 + random.random() * 0.5) # 随机抖动
await asyncio.sleep(delay)指数退避时序:
调用1: 失败 -> 等待 1s
调用2: 失败 -> 等待 2s
调用3: 失败 -> 等待 4s
调用4: 成功! (或达到 max_retries)
抖动作用: 避免多客户端同时重试造成"重试风暴"6.4 降级策略
降级链:
主工具: GPT-4 API <- 失败(限流)
| 降级
备选1: GPT-3.5 API <- 失败(超时)
| 降级
备选2: 本地小模型 <- 成功
|
返回结果(质量降低,但服务不中断)
降级原则: 质量递减,可用性递增6.5 幂等性
非幂等操作(需特别处理):
POST /transfer <- 重复执行会多扣钱!
需要幂等键: idempotency-key = "txn_001"
相同 key 的重复请求 -> 返回首次结果
天然幂等操作(可安全重试):
GET /user/123 <- 读取无副作用
PUT /user/123 <- 全量更新,结果一致
DELETE /user/123 <- 已删除再删也无影响6.6 错误处理策略矩阵
┌─────────────┬──────┬──────┬──────┬──────┐
│ 错误类型 │ 重试 │ 超时 │ 降级 │ 幂等 │
│ TRANSIENT │ 是 │ - │ 可选 │ 检查 │
│ VALIDATION │ 否 │ - │ 否 │ - │
│ PERMISSION │ 否 │ - │ 否 │ - │
│ TIMEOUT │ 是 │ 是 │ 是 │ 必须 │
│ NOT_FOUND │ 否 │ - │ 是 │ - │
│ RATE_LIMIT │ 是 │ - │ 是 │ 检查 │
└─────────────┴──────┴──────┴──────┴──────┘6.7 与其他概念的关联
<- 工具安全与权限：权限拒绝是一种需要处理的错误
-> 常用工具集成：每种工具集成都需要错误处理
-> 完整工具生态：错误处理是工具生态的可靠性保障
7. 常用工具集成
7.1 定义
常用工具集成将搜索、数据库、文件操作、代码执行、浏览器自动化、通信通知等常见能力封装为 MCP 工具，供 Agent 按需调用。
对应 Demo: demos/05_工具集成与MCP/07_常用工具集成.py
7.2 六大类常用工具
┌──────────┬───────────────┬─────────────────────────────┐
│ 搜索类 │ web_search │ Low (只读) │
│ 数据库类 │ sql_query │ Low(读) / High(写) │
│ 文件类 │ read_file │ Medium / High(write) │
│ 代码类 │ run_python │ Critical (需沙箱) │
│ 浏览器类 │ navigate │ Medium │
│ 通信类 │ send_email │ High │
└──────────┴───────────────┴─────────────────────────────┘7.3 各类工具示例
# 搜索工具: 调用搜索 API,返回 JSON 格式结果
async def web_search(query: str, max_results: int = 5) -> str:
"""在互联网上搜索信息"""
results = await search_api.search(query, max_results)
return json.dumps([{"title": r.title, "url": r.url} for r in results])
# 数据库工具: SQL 安全检查 + 参数化执行
async def sql_query(connection: str, sql: str) -> str:
"""执行 SQL 查询(仅 SELECT)"""
is_safe, msg = sql_guard.validate_sql(sql)
if not is_safe:
return f"SQL 安全检查失败: {msg}"
rows = await get_connection(connection).execute(sql)
return json.dumps(rows)
# 文件工具: 路径检查 + 大小限制
async def read_file(path: str, encoding: str = "utf-8") -> str:
"""读取文件内容(路径必须在允许范围内)"""
is_safe, real_path = path_guard.validate_path(path)
if not is_safe:
return f"路径访问被拒绝: {real_path}"
if os.path.getsize(real_path) > 10 * 1024 * 1024:
return "文件过大,超过 10MB 限制"
with open(real_path, encoding=encoding) as f:
return f.read()
# 代码执行: 必须在沙箱中
async def run_python(code: str, timeout: float = 10.0) -> str:
"""在沙箱中执行 Python 代码"""
result = await sandbox.execute(code=code, timeout=timeout,
allowed_imports=["math", "json", "re", "datetime"], max_memory="256MB")
return result.output7.4 工具选择决策树
Agent 决策使用哪个工具?
需要外部信息? ──> web_search / kb_search
需要查询数据? ──> sql_query / nosql_query
需要读写文件? ──> read_file / write_file
需要执行计算? ──> run_python / calculate
需要浏览网页? ──> browser_navigate
需要发送通知? ──> send_email / send_message7.5 与其他概念的关联
<- MCP Client 集成：常用工具通过 MCP Server 提供给 Client
<- 工具 Schema 设计：每种工具都需要精心设计的 Schema
-> 工具安全与权限：不同工具有不同风险等级
-> 工具错误处理：每种工具需要针对性的错误处理
8. MCP 传输层
8.1 定义
MCP 传输层是 Client 与 Server 之间传递 JSON-RPC 消息的底层通信通道，定义了 stdio 和 Streamable HTTP 两种标准传输方式。
对应 Demo: demos/05_工具集成与MCP/08_MCP传输层.py
8.2 两种传输方式对比
┌────────────────┬─────────────────────┬─────────────────────────┐
│ 通信方式 │ stdin/stdout │ HTTP POST + SSE │
│ 部署模式 │ 本地(同机) │ 本地或远程 │
│ Server 形态 │ 子进程 │ 独立服务 │
│ 延迟 │ 极低(管道) │ 较低(HTTP 开销) │
│ 安全性 │ 进程隔离 │ 需 TLS + 认证 │
│ 多 Client │ 1:1 │ 1:N │
│ 典型场景 │ IDE 插件、本地工具 │ 云端服务、共享工具 │
└────────────────┴─────────────────────┴─────────────────────────┘8.3 stdio 传输架构
Host 进程
┌────────────────────────────────────┐
│ Agent -> MCP Client │
│ │ 写 stdin 读 stdout │
│ v ^ │
│ ┌──┐ stdin pipe ┌──┐ │
│ │ │─────────────>│ │ │
│ │ │<─────────────│ │ │
│ └──┘ stdout pipe └──┘ │
└─────────┬──────────────────────────┘
v fork/exec
┌────────────────────┐
│ MCP Server 进程 │
│ 读 stdin/write stdout│
└────────────────────┘
特点: 父子进程关系,管道通信,零网络开销8.4 Streamable HTTP 传输架构
Host 进程 远程 HTTP Server
┌──────────────┐ ┌──────────────────┐
│ MCP Client │ HTTP POST │ MCP HTTP Server │
│ │─────────┼───────────>│ │ 路由到处理函数 │
│ │<────────┼───────────│ │ │
└──────────────┘ +SSE响应 └──────────────────┘
流程: POST(JSON-RPC) -> SSE 流(多个响应) -> 连接复用8.5 传输层选择决策
工具需要远程访问?
├── 否 ──> stdio (零配置,零延迟,进程隔离安全)
└── 是 ──> Streamable HTTP
├── 多 Client 共享? ──> HTTP (1:N)
├── 负载均衡? ────────> HTTP (可加 LB)
└── 容器部署? ────────> HTTP (云原生友好)8.6 与其他概念的关联
<- MCP Server 开发：Server 通过传输层提供能力
<- MCP Client 集成：Client 通过传输层连接 Server
-> 完整工具生态：传输层是工具生态的通信基础
9. 工具版本管理
9.1 定义
工具版本管理对 MCP 工具的迭代演进进行系统化管理，采用 SemVer 标识变更，管理兼容性，并在废弃时提供平滑迁移路径。
对应 Demo: demos/05_工具集成与MCP/09_工具版本管理.py
9.2 语义化版本（SemVer）
版本号格式: MAJOR.MINOR.PATCH
1.4.2
│ │ └── PATCH: 补丁版本(向后兼容的 Bug 修复)
│ └──── MINOR: 次版本(向后兼容的功能新增)
└────── MAJOR: 主版本(不兼容的 API 变更)
变更类型与版本号:
Bug 修复 1.4.2 -> 1.4.3 (完全兼容)
新增可选参数 1.4.2 -> 1.5.0 (向后兼容)
删除参数 1.4.2 -> 2.0.0 (不兼容 Breaking)
修改参数类型 1.4.2 -> 2.0.0 (不兼容)9.3 版本兼容性规则
^1.0.0 (>=1.0.0,<2.0.0): 1.0.0~1.9.9 兼容, 2.0.0 不兼容
~1.4.0 (>=1.4.0,<1.5.0): 1.4.0~1.4.9 兼容, 1.5.0 不兼容
1.4.2 (精确匹配): 仅 1.4.2 兼容9.4 废弃流程
阶段1: Deprecation Warning (废弃警告)
- 标记 deprecated: true
- 调用时返回警告(但仍执行)
- 设定 sunset_date (如 6 个月后)
阶段2: Sunset (日落)
- 调用时返回错误"工具已废弃"
- 提供替代方案建议
- 不再执行逻辑
阶段3: Removal (移除)
- 从 tools/list 中完全移除9.5 版本变更类型
class VersionChange(Enum):
"""版本变更类型"""
PATCH = "patch" # Bug 修复,完全兼容
MINOR = "minor" # 新增功能,向后兼容
MAJOR = "major" # 破坏性变更,不兼容
DEPRECATION = "deprecation" # 废弃警告
REMOVAL = "removal" # 完全移除9.6 与其他概念的关联
<- 工具 Schema 设计：Schema 变更驱动版本号变更
-> 完整工具生态：版本管理是工具生态的治理能力
-> MCP 传输层：版本协商在连接初始化时进行
10. 完整工具生态
10.1 定义
完整工具生态（ToolEcosystem）是将 MCP 协议、Server/Client、Schema 设计、安全权限、错误处理、传输层、版本管理等所有组件有机集成的统一系统，提供从工具注册、发现、调用、监控到退役的全生命周期管理。
对应 Demo: demos/05_工具集成与MCP/10_完整工具生态.py
10.2 生态架构总览
┌──────────────────────────────────────────────────────────────┐
│ ToolEcosystem（工具生态） │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Agent │ │ 监控面板 │ │ 版本注册表 │ │
│ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 统一 API 层 (Unified API) │ │
│ └──────────────────────┬───────────────────────────────┘ │
│ ┌────────────────┼────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │安全策略层 │ │错误处理层 │ │传输管理层 │ │
│ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ MCP Client 池 (Multi-Server) │ │
│ └──────────────────────┬───────────────────────────────┘ │
└─────────────────────────┼────────────────────────────────────┘
│ JSON-RPC 2.0
▼
多个 MCP Server10.3 核心组件集成
class ToolEcosystem:
"""完整工具生态:集成所有工具管理能力"""
def __init__(self):
self.tool_manager = AgentToolManager() # 工具管理
self.security = SecurityPolicy() # 安全策略
self.error_handler = RobustToolExecutor() # 错误处理
self.version_registry = ToolVersionRegistry() # 版本管理
self.monitor = CallMonitor() # 调用监控
async def call_tool(self, tool_name: str, args: dict, context: dict):
"""统一的工具调用入口(集成所有能力)"""
# 1. 版本兼容性检查
self.version_registry.get_compatible(tool_name, "^1.0.0")
# 2. 安全检查
allowed, reason = self.security.check_permission(
tool_name, args, context.get("user_role", "user"))
if not allowed:
return CallResult(success=False, error=reason)
# 3. 带错误处理的调用
result = await self.error_handler.execute(
self.tool_manager.call_tool, (tool_name, args))
# 4. 记录调用(监控/审计)
self.monitor.record(CallRecord(tool=tool_name, args=args, result=result))
return result10.4 调用全流程
ToolEcosystem 工具调用全流程:
Agent 调用 call_tool("search", {"query": "MCP"})
-> 1. 版本检查: 是否有兼容版本?
-> 2. 安全检查: 风险等级? 权限? 路径? SQL?
-> 3. 参数校验: Schema 校验
-> 4. 路由到 Server: 找到提供此工具的 MCP Server
-> 5. 错误处理包装: 重试/超时/降级
-> 6. 传输层调用: stdio 或 HTTP 发送 JSON-RPC
-> 7. Server 执行: 运行业务逻辑
-> 8. 结果返回: 经过错误处理层检查
-> 9. 监控记录: 调用详情、延迟、成本
-> 10. 返回给 Agent10.5 工具全生命周期
工具生命周期:
注册 -> 发现 -> 调用 -> 监控 -> 版本升级 -> 废弃 -> 移除
Schema tools/ tools/ 记录 SemVer 警告 tools/
定义 list call 调用 版本号 日志 list10.6 生态健康度指标
┌────────────────┬──────────────────────────────────┐
│ 工具可用率 │ 可调用工具数 / 注册工具数 │
│ 调用成功率 │ 成功调用数 / 总调用数 │
│ 平均延迟 │ 所有工具调用的平均响应时间 │
│ 错误率 │ 按错误类型分布的错误占比 │
│ 版本覆盖率 │ 使用最新版本的 Agent 占比 │
│ 废弃工具使用率 │ 仍在调用已废弃工具的占比(应趋近 0) │
└────────────────┴──────────────────────────────────┘10.7 与其他概念的关联
<- MCP 概念与架构：生态基于 MCP 协议构建
<- MCP Server/Client：生态管理多 Server/Client
<- 工具 Schema 设计：生态使用 Schema 描述工具
<- 工具安全与权限：生态集成安全策略
<- 工具错误处理：生态提供鲁棒的错误处理
<- MCP 传输层：生态支持多种传输方式
<- 工具版本管理：生态治理工具版本演进
概念关系总览
┌──────────────────────────────────────────┐
│ MCP 工具集成技术体系 │
└──────────────────────────────────────────┘
协议层 能力层 治理层
│ │ │
▼ ▼ ▼
┌──────────┐ ┌────────────┐ ┌────────────┐
│MCP 概念 │──> Server ──>│ Schema 设计│──────────>│ 安全与权限 │
│与架构 │ 开发 │ │ │ │
└────┬─────┘ └────────────┘ └────────────┘
│ │ │
▼ ▼ ▼
┌──────────┐ ┌────────────┐ ┌────────────┐
│MCP 传输层 │ │常用工具集成 │<──────────│ 错误处理 │
│(stdio/ │ │(搜索/DB/ │ │(重试/超时/ │
│ HTTP) │ │ 文件/代码) │ │ 降级/幂等) │
└────┬─────┘ └────────────┘ └────────────┘
│ │ │
│ ▼ ▼
│ ┌──────────────────────────────────┐
└─────────────>│ 完整工具生态 (ToolEcosystem) │
│ 注册->发现->调用->监控-> │
│ 版本升级->废弃->移除 │
└──────────────────────────────────┘
▲
┌───────────┘
│ 工具版本管理 (SemVer/兼容性/废弃)
评论区