工具系统设计——工具注册表、工具规格定义、工具发现与调用、MCP协议集成
覆盖范围：工具注册、参数校验、执行调度、MCP适配、工具发现、权限安全、版本管理
目录
- 一、工具系统架构总览
- 二、ToolSpec 工具规格详细设计
- 三、工具注册表深度设计
- 四、工具发现机制
- 五、工具选择策略
- 六、工具参数生成与验证
- 七、工具执行引擎
- 八、工具结果处理
- 九、MCP协议集成
- 十、内置工具库设计
- 十一、自定义工具开发指南
- 十二、工具权限与安全
- 十三、工具版本管理
- 十四、与LangChain/CrewAI工具系统的对比
- 十五、通俗实例：智能运维Agent
- 十六、完整代码实现
- 十七、Q&A常见问题
一、工具系统架构总览
1.1 工具系统在Agent架构中的定位
工具系统（Tool System）是Agent与外部世界交互的桥梁。如果说LLM是Agent的"大脑"，那么工具系统就是Agent的"双手"——大脑负责思考决策，双手负责执行落地。没有工具系统的Agent只能"纸上谈兵"，有了工具系统才能"真刀真枪"地完成实际任务。
通俗类比：想象你是一个公司经理（LLM），你手下有一整个工具箱（工具注册表），里面有螺丝刀、扳手、电钻等工具。你要修一台机器，首先需要知道有哪些工具可用（工具发现），然后选择合适的工具（工具选择），检查工具是否匹配（参数验证），使用工具操作（工具执行），最后检查操作结果（结果处理）。
1.2 工具系统五大核心组件
工具系统由五大核心组件构成，每个组件职责明确、边界清晰：
| 组件 | 职责 | 核心问题 |
|---|---|---|
| 工具注册表 | 管理所有工具的注册与查询 | “有哪些工具可用？” |
| 工具规格 | 定义工具的元数据与接口契约 | “这个工具怎么用？” |
| 工具发现 | 根据需求找到合适的工具 | “哪个工具能做这件事？” |
| 工具执行引擎 | 调用工具并管理执行过程 | “如何安全高效地执行？” |
| 工具结果处理 | 处理工具返回的结果 | “执行结果如何呈现？” |
1.3 工具系统数据流
1.4 工具系统设计原则
设计一个优秀的工具系统，需要遵循以下原则：
- 统一接口：所有工具遵循统一的注册、调用、返回接口，降低使用心智负担
- 可扩展性：支持动态注册、热插拔，新增工具不需要重启系统
- 安全隔离：工具执行在受控环境中进行，防止恶意工具破坏系统
- 可观测性：每次工具调用都有完整日志和追踪，便于调试和审计
- 容错性：单个工具失败不影响整体系统，有完善的错误处理和重试机制
- 性能可控：支持超时控制、资源限制、并发管理，防止工具执行耗尽资源
1.5 工具系统分层架构
二、ToolSpec 工具规格详细设计
2.1 ToolSpec 的角色
ToolSpec（工具规格定义）是工具的"身份证"和"使用说明书"。它描述了一个工具的全部元信息：叫什么名字、做什么用、需要什么参数、返回什么结果、有什么权限限制。没有ToolSpec的工具就像没有说明书的设备——LLM不知道它存在、不知道它怎么用、更不敢调用它。
通俗类比：ToolSpec 就像是餐馆的菜单。菜单上写着每道菜的名字（name）、描述（description）、配料（parameters）、价格（权限级别）、上菜时间（超时设置）。顾客（LLM）看菜单点菜，厨师（执行引擎）按菜单做菜。
2.2 ToolSpec 完整字段定义
from typing import Callable, Any, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
import time
class PermissionLevel(Enum):
"""工具权限级别"""
READONLY = "readonly" # 只读操作，如搜索、查询
WRITE = "write" # 写入操作，如创建文件、发送消息
DESTRUCTIVE = "destructive" # 破坏性操作，如删除文件、格式化磁盘
SYSTEM = "system" # 系统级操作，如执行命令、修改配置
class ToolStatus(Enum):
"""工具状态"""
ACTIVE = "active" # 活跃可用
DISABLED = "disabled" # 已禁用
ERROR = "error" # 异常状态
LOADING = "loading" # 加载中
@dataclass
class ToolSpec:
"""工具规格定义 - 工具的完整元信息"""
# === 基础信息 ===
name: str # 工具名称，全局唯一
description: str # 工具描述，供LLM理解工具用途
func: Callable # 工具的可调用函数
# === 参数与返回值 ===
parameters: dict = field(default_factory=dict) # 参数JSON Schema
returns: dict = field(default_factory=dict) # 返回值JSON Schema
# === 权限与安全 ===
permission: PermissionLevel = PermissionLevel.READONLY # 权限级别
requires_confirmation: bool = False # 是否需要用户确认
allowed_in_sandbox: bool = True # 是否允许在沙箱中执行
# === 执行控制 ===
timeout: float = 30.0 # 超时时间（秒）
max_retries: int = 0 # 最大重试次数
rate_limit: Optional[int] = None # 每分钟最大调用次数
# === 元数据 ===
version: str = "1.0.0" # 工具版本
author: str = "" # 作者
tags: list = field(default_factory=list) # 标签，用于分类和搜索
examples: list = field(default_factory=list) # 使用示例
# === 运行时状态 ===
status: ToolStatus = ToolStatus.ACTIVE # 当前状态
call_count: int = 0 # 累计调用次数
last_called: Optional[float] = None # 最后调用时间戳
avg_latency: float = 0.0 # 平均延迟（秒）
# === 缓存 ===
cacheable: bool = False # 结果是否可缓存
cache_ttl: int = 300 # 缓存有效期（秒）
def to_schema(self) -> dict:
"""转换为LLM可理解的工具描述格式"""
return {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
"returns": self.returns,
"permission": self.permission.value,
"timeout": self.timeout,
"requires_confirmation": self.requires_confirmation,
"tags": self.tags,
"version": self.version,
}
def to_public_schema(self) -> dict:
"""转换为公开的工具描述（不含内部实现细节）"""
return {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
2.3 ToolSpec 字段说明表
| 字段分类 | 字段名 | 类型 | 说明 |
|---|---|---|---|
| 基础信息 | name | str | 工具名称，全局唯一标识 |
| 基础信息 | description | str | 工具描述，LLM据此理解工具用途 |
| 基础信息 | func | Callable | 工具的实际执行函数 |
| 参数返回 | parameters | dict | 参数的JSON Schema定义 |
| 参数返回 | returns | dict | 返回值的JSON Schema定义 |
| 权限安全 | permission | PermissionLevel | 权限级别枚举 |
| 权限安全 | requires_confirmation | bool | 是否需要用户确认才执行 |
| 权限安全 | allowed_in_sandbox | bool | 是否允许在沙箱环境中执行 |
| 执行控制 | timeout | float | 执行超时时间（秒） |
| 执行控制 | max_retries | int | 失败后最大重试次数 |
| 执行控制 | rate_limit | int/None | 每分钟最大调用次数限制 |
| 元数据 | version | str | 工具版本号 |
| 元数据 | tags | list | 分类标签，用于搜索和过滤 |
| 元数据 | examples | list | 使用示例 |
| 运行时状态 | status | ToolStatus | 工具当前状态 |
| 运行时状态 | call_count | int | 累计调用次数 |
| 运行时状态 | avg_latency | float | 平均执行延迟 |
| 缓存 | cacheable | bool | 结果是否可缓存 |
| 缓存 | cache_ttl | int | 缓存有效期（秒） |
2.4 参数JSON Schema设计
参数Schema是ToolSpec中最关键的部分，它定义了工具需要哪些参数、每个参数的类型和约束。LLM根据这个Schema生成调用参数。
# 搜索工具的参数Schema示例
search_tool_params = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词",
"minLength": 1,
"maxLength": 500
},
"num_results": {
"type": "integer",
"description": "返回结果数量",
"default": 10,
"minimum": 1,
"maximum": 50
},
"language": {
"type": "string",
"description": "搜索语言",
"enum": ["zh", "en", "ja", "ko"],
"default": "zh"
},
"time_range": {
"type": "string",
"description": "时间范围过滤",
"enum": ["day", "week", "month", "year", "all"],
"default": "all"
}
},
"required": ["query"],
"additionalProperties": False
}
# 文件写入工具的参数Schema示例
file_write_params = {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径（绝对路径）",
"pattern": "^/[^<>:|?*]*$"
},
"content": {
"type": "string",
"description": "写入内容"
},
"mode": {
"type": "string",
"description": "写入模式",
"enum": ["write", "append"],
"default": "write"
},
"encoding": {
"type": "string",
"description": "文件编码",
"default": "utf-8"
}
},
"required": ["path", "content"],
"additionalProperties": False
}
2.5 权限级别设计
通俗类比：权限级别就像是银行账户的操作等级。READONLY是查看余额，WRITE是转账，DESTRUCTIVE是销户，SYSTEM是修改银行系统配置。级别越高，需要的确认和审计就越严格。
2.6 ToolSpec 创建示例
# 创建一个完整的搜索工具规格
search_spec = ToolSpec(
name="web_search",
description="在互联网上搜索信息，返回相关网页摘要和链接。适用于查找最新资讯、技术文档、产品信息等。",
func=search_function,
parameters=search_tool_params,
returns={
"type": "object",
"properties": {
"results": {
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"url": {"type": "string"},
"snippet": {"type": "string"}
}
}
},
"total": {"type": "integer"}
}
},
permission=PermissionLevel.READONLY,
requires_confirmation=False,
timeout=15.0,
max_retries=2,
rate_limit=60,
version="2.1.0",
author="system",
tags=["search", "web", "information"],
examples=[
{"query": "Python asyncio教程", "num_results": 5},
{"query": "最新AI新闻", "time_range": "day"}
],
cacheable=True,
cache_ttl=600
)
三、工具注册表深度设计
3.1 工具注册表的角色
工具注册表（ToolRegistry）是工具系统的"中央仓库"，管理Agent可调用的全部工具。它负责工具的注册、注销、查询、列表展示等核心功能。
标准注册方式为装饰器模式：
from typing import Callable, Any
from dataclasses import dataclass
@dataclass
class ToolSpec:
name: str
description: str
func: Callable
parameters: dict
class ToolRegistry:
def __init__(self):
self._tools = {}
def register(self, name, desc="", params=None):
def wrapper(fn):
self._tools[name] = ToolSpec(name, desc, fn, params or {})
return fn
return wrapper
def schemas(self) -> list[dict]:
return [{"name": t.name, "description": t.description, "parameters": t.parameters} for t in self._tools.values()]
registry = ToolRegistry()
@registry.register("web_search", "搜索互联网", {"type":"object","properties":{"query":{"type":"string"}}})
async def search(query: str):
return f"模拟搜索:{query}"
3.2 注册表完整实现
import asyncio
import threading
import importlib
import inspect
from typing import Callable, Any, Optional
from pathlib import Path
class ToolRegistry:
"""工具注册表 - 管理所有工具的生命周期"""
def __init__(self):
self._tools: dict[str, ToolSpec] = {}
self._lock = threading.RLock() # 线程安全锁
self._hooks: dict[str, list[Callable]] = {
"before_register": [],
"after_register": [],
"before_unregister": [],
"after_unregister": [],
}
self._tool_groups: dict[str, list[str]] = {} # 工具分组
# ========== 注册 ==========
def register(self, name: str, description: str = "", parameters: dict = None,
**kwargs) -> Callable:
"""
装饰器模式注册工具
用法:
@registry.register("my_tool", "我的工具", {"type": "object", ...})
async def my_tool(param: str):
return result
"""
def wrapper(fn: Callable) -> Callable:
spec = ToolSpec(
name=name,
description=description,
func=fn,
parameters=parameters or {},
**kwargs
)
self._register_spec(spec)
return fn
return wrapper
def register_spec(self, spec: ToolSpec) -> None:
"""直接注册一个ToolSpec对象"""
self._register_spec(spec)
def register_function(self, name: str, func: Callable, description: str = "",
parameters: dict = None, **kwargs) -> None:
"""直接注册一个函数（非装饰器方式）"""
spec = ToolSpec(
name=name,
description=description,
func=func,
parameters=parameters or {},
**kwargs
)
self._register_spec(spec)
def _register_spec(self, spec: ToolSpec) -> None:
"""内部注册逻辑"""
self._run_hooks("before_register", spec)
with self._lock:
if spec.name in self._tools:
# 版本检查：允许高版本覆盖低版本
existing = self._tools[spec.name]
if self._version_compare(spec.version, existing.version) <= 0:
raise ValueError(
f"工具 '{spec.name}' 已存在且版本不低于当前版本 "
f"(existing: {existing.version}, new: {spec.version})"
)
self._tools[spec.name] = spec
self._run_hooks("after_register", spec)
# ========== 注销 ==========
def unregister(self, name: str) -> Optional[ToolSpec]:
"""注销工具"""
with self._lock:
if name not in self._tools:
return None
spec = self._tools[name]
self._run_hooks("before_unregister", spec)
del self._tools[name]
# 从分组中移除
for group_tools in self._tool_groups.values():
if name in group_tools:
group_tools.remove(name)
self._run_hooks("after_unregister", spec)
return spec
# ========== 查询 ==========
def get(self, name: str) -> Optional[ToolSpec]:
"""按名称获取工具"""
with self._lock:
return self._tools.get(name)
def has(self, name: str) -> bool:
"""检查工具是否存在"""
with self._lock:
return name in self._tools
def list_tools(self, tag: str = None, permission: PermissionLevel = None,
status: ToolStatus = ToolStatus.ACTIVE) -> list[ToolSpec]:
"""列出工具，支持按标签、权限、状态过滤"""
with self._lock:
tools = list(self._tools.values())
if tag:
tools = [t for t in tools if tag in t.tags]
if permission:
tools = [t for t in tools if t.permission == permission]
if status:
tools = [t for t in tools if t.status == status]
return tools
def schemas(self) -> list[dict]:
"""返回所有工具的Schema（供LLM使用）"""
with self._lock:
return [t.to_schema() for t in self._tools.values()]
def public_schemas(self) -> list[dict]:
"""返回公开Schema（精简版）"""
with self._lock:
return [t.to_public_schema() for t in self._tools.values()]
# ========== 分组管理 ==========
def create_group(self, group_name: str, tool_names: list[str]) -> None:
"""创建工具分组"""
with self._lock:
self._tool_groups[group_name] = list(tool_names)
def get_group(self, group_name: str) -> list[ToolSpec]:
"""获取分组中的工具"""
with self._lock:
names = self._tool_groups.get(group_name, [])
return [self._tools[n] for n in names if n in self._tools]
# ========== 动态加载 ==========
def load_from_module(self, module_path: str) -> int:
"""
从Python模块动态加载工具
模块中所有用@registry.register装饰的函数会自动注册
返回注册的工具数量
"""
module = importlib.import_module(module_path)
count = 0
for name, obj in inspect.getmembers(module):
if callable(obj) and hasattr(obj, '_tool_registered'):
count += 1
return count
def load_from_directory(self, dir_path: str) -> int:
"""从目录批量加载工具模块"""
path = Path(dir_path)
if not path.is_dir():
raise FileNotFoundError(f"目录不存在: {dir_path}")
count = 0
for py_file in path.glob("*.py"):
if py_file.name.startswith("_"):
continue
module_name = py_file.stem
try:
count += self.load_from_module(module_name)
except Exception as e:
print(f"加载模块 {module_name} 失败: {e}")
return count
# ========== 钩子系统 ==========
def add_hook(self, event: str, callback: Callable) -> None:
"""添加钩子函数"""
if event in self._hooks:
self._hooks[event].append(callback)
def _run_hooks(self, event: str, *args, **kwargs) -> None:
"""执行钩子"""
for callback in self._hooks.get(event, []):
try:
callback(*args, **kwargs)
except Exception as e:
print(f"钩子 {event} 执行失败: {e}")
# ========== 统计 ==========
def stats(self) -> dict:
"""返回注册表统计信息"""
with self._lock:
total = len(self._tools)
by_permission = {}
by_status = {}
total_calls = 0
for tool in self._tools.values():
perm = tool.permission.value
by_permission[perm] = by_permission.get(perm, 0) + 1
status = tool.status.value
by_status[status] = by_status.get(status, 0) + 1
total_calls += tool.call_count
return {
"total_tools": total,
"by_permission": by_permission,
"by_status": by_status,
"total_calls": total_calls,
"groups": len(self._tool_groups),
}
@staticmethod
def _version_compare(v1: str, v2: str) -> int:
"""比较版本号，返回1(v1>v2)、0(相等)、-1(v1<v2)"""
parts1 = [int(x) for x in v1.split(".")]
parts2 = [int(x) for x in v2.split(".")]
for p1, p2 in zip(parts1, parts2):
if p1 > p2:
return 1
elif p1 < p2:
return -1
return 0
3.3 装饰器注册模式详解
装饰器模式是最优雅的工具注册方式。开发者只需在函数上方添加装饰器，工具就自动注册到注册表中：
# 创建全局注册表实例
registry = ToolRegistry()
# 方式1：基础装饰器注册
@registry.register(
"web_search",
"搜索互联网，返回相关网页信息",
{
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
)
async def web_search(query: str):
return f"模拟搜索结果: {query}"
# 方式2：带完整参数的装饰器注册
@registry.register(
"file_read",
"读取文件内容",
{
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"},
"encoding": {"type": "string", "default": "utf-8"}
},
"required": ["path"]
},
permission=PermissionLevel.READONLY,
timeout=10.0,
tags=["file", "read"],
version="1.2.0"
)
async def file_read(path: str, encoding: str = "utf-8"):
with open(path, "r", encoding=encoding) as f:
return f.read()
# 方式3：非装饰器直接注册
def calculate(expression: str):
"""计算数学表达式"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return {"expression": expression, "result": result}
except Exception as e:
return {"error": str(e)}
registry.register_function(
name="calculator",
func=calculate,
description="计算数学表达式",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
permission=PermissionLevel.READONLY,
tags=["math", "calculate"]
)
3.4 动态加载与热插拔
# 动态加载示例
class ToolHotSwap:
"""工具热插拔管理器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self._watched_dirs: dict[str, float] = {} # 目录 -> 上次修改时间
def watch_directory(self, dir_path: str) -> None:
"""监控目录变化，自动加载新工具"""
path = Path(dir_path)
if not path.is_dir():
return
self._watched_dirs[dir_path] = self._get_dir_mtime(path)
def check_updates(self) -> list[str]:
"""检查是否有更新，返回变化的工具列表"""
changed = []
for dir_path, last_mtime in self._watched_dirs.items():
path = Path(dir_path)
current_mtime = self._get_dir_mtime(path)
if current_mtime > last_mtime:
# 重新加载该目录
new_count = self.registry.load_from_directory(dir_path)
changed.append(f"{dir_path}: {new_count} tools reloaded")
self._watched_dirs[dir_path] = current_mtime
return changed
@staticmethod
def _get_dir_mtime(path: Path) -> float:
"""获取目录最后修改时间"""
return max(f.stat().st_mtime for f in path.glob("*.py"))
3.5 注册表与LLM的交互
class ToolRegistryLLMAdapter:
"""工具注册表与LLM之间的适配器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
def get_tools_for_llm(self, filter_tags: list[str] = None) -> list[dict]:
"""
获取LLM可理解的工具列表
可按标签过滤，只暴露相关工具给LLM
"""
tools = self.registry.list_tools()
if filter_tags:
tools = [
t for t in tools
if any(tag in t.tags for tag in filter_tags)
]
return [t.to_schema() for t in tools]
def get_tool_descriptions(self) -> str:
"""生成工具描述文本（用于System Prompt）"""
tools = self.registry.list_tools()
lines = ["可用工具列表:"]
for tool in tools:
lines.append(f"\n- {tool.name}: {tool.description}")
if tool.parameters.get("properties"):
lines.append(" 参数:")
for param_name, param_info in tool.parameters["properties"].items():
required = param_name in tool.parameters.get("required", [])
req_str = "必填" if required else "可选"
lines.append(
f" - {param_name} ({param_info.get('type', 'any')}): "
f"{param_info.get('description', '')} [{req_str}]"
)
return "\n".join(lines)
通俗类比：注册表与LLM的交互就像图书馆的目录系统。注册表是图书管理员，LLM是来借书的读者。管理员不会把所有书都搬到读者面前，而是根据读者的需求（filter_tags），提供一份精选的书目清单（get_tools_for_llm），并附上每本书的简介和借阅要求（参数说明）。
四、工具发现机制
4.1 工具发现的核心问题
工具发现（Tool Discovery）解决的核心问题是：“给定一个任务需求，如何找到最合适的工具？” 这听起来简单，但当工具数量从几个增长到几百个甚至上千个时，这就变成了一个极具挑战性的问题。
4.2 静态注册发现
静态注册发现是最简单直接的方式：工具在系统启动时注册到注册表，使用时直接按名称或标签查找。
class StaticToolDiscovery:
"""静态工具发现 - 基于预注册的工具表"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
def discover_by_name(self, name: str) -> Optional[ToolSpec]:
"""按名称精确查找"""
return self.registry.get(name)
def discover_by_tag(self, tag: str) -> list[ToolSpec]:
"""按标签查找"""
return self.registry.list_tools(tag=tag)
def discover_by_permission(self, permission: PermissionLevel) -> list[ToolSpec]:
"""按权限级别查找"""
return self.registry.list_tools(permission=permission)
def discover_all(self) -> list[ToolSpec]:
"""列出所有可用工具"""
return self.registry.list_tools()
def discover_for_task(self, task_description: str) -> list[ToolSpec]:
"""
根据任务描述发现工具（基于关键词匹配）
这是最简单的匹配方式，适用于工具数量较少的场景
"""
task_lower = task_description.lower()
scored_tools = []
for tool in self.registry.list_tools():
score = 0
# 名称匹配
if tool.name.lower() in task_lower:
score += 10
# 描述匹配
desc_words = tool.description.lower().split()
for word in desc_words:
if len(word) > 2 and word in task_lower:
score += 3
# 标签匹配
for tag in tool.tags:
if tag.lower() in task_lower:
score += 5
if score > 0:
scored_tools.append((score, tool))
scored_tools.sort(key=lambda x: x[0], reverse=True)
return [tool for _, tool in scored_tools]
通俗类比：静态发现就像在自家工具箱里找工具。你知道扳手在哪个格子、螺丝刀在哪个抽屉，直接去拿就行。但如果工具箱有上千件工具，靠记忆找就太慢了。
4.3 MCP动态发现
MCP（Model Context Protocol）是一种标准化的工具发现协议。通过MCP，Agent可以在运行时动态发现远程工具服务器上提供的工具，无需预先注册。
import json
from typing import Any
class MCPDiscoveryClient:
"""MCP动态发现客户端"""
def __init__(self):
self._servers: dict[str, dict] = {} # 服务器名 -> 服务器信息
self._discovered_tools: dict[str, ToolSpec] = {}
async def connect_server(self, server_name: str, server_config: dict) -> bool:
"""
连接MCP工具服务器
server_config 示例:
{
"transport": "stdio", # 或 "http"
"command": "python",
"args": ["-m", "mcp_server_search"],
"env": {"API_KEY": "..."}
}
"""
try:
# 模拟MCP握手
print(f"[MCP] 连接服务器: {server_name}")
print(f"[MCP] 传输方式: {server_config.get('transport', 'stdio')}")
# 发送initialize请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "agent-tool-system",
"version": "1.0.0"
}
}
}
# 模拟响应
init_response = {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": server_name,
"version": "1.0.0"
}
}
self._servers[server_name] = {
"config": server_config,
"info": init_response,
"connected": True
}
return True
except Exception as e:
print(f"[MCP] 连接失败: {e}")
return False
async def discover_tools(self, server_name: str) -> list[dict]:
"""从MCP服务器发现工具"""
if server_name not in self._servers:
raise ValueError(f"服务器 {server_name} 未连接")
# 模拟发送 tools/list 请求
list_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
# 模拟响应
mock_response = {
"tools": [
{
"name": "mcp_search",
"description": "MCP搜索工具",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
},
{
"name": "mcp_fetch",
"description": "MCP网页抓取工具",
"inputSchema": {
"type": "object",
"properties": {
"url": {"type": "string"}
},
"required": ["url"]
}
}
]
}
discovered = []
for tool_info in mock_response["tools"]:
# 转换为ToolSpec
spec = ToolSpec(
name=f"{server_name}_{tool_info['name']}",
description=tool_info["description"],
func=self._create_mcp_proxy(server_name, tool_info["name"]),
parameters=tool_info.get("inputSchema", {}),
tags=["mcp", server_name],
version="1.0.0"
)
self._discovered_tools[spec.name] = spec
discovered.append(spec)
return discovered
def _create_mcp_proxy(self, server_name: str, tool_name: str) -> Callable:
"""创建MCP工具代理函数"""
async def proxy(**kwargs):
# 模拟发送 tools/call 请求
call_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": kwargs
}
}
# 模拟响应
return {
"content": [
{"type": "text", "text": f"[MCP] {server_name}.{tool_name} 执行结果"}
]
}
return proxy
async def disconnect_server(self, server_name: str) -> None:
"""断开MCP服务器连接"""
if server_name in self._servers:
self._servers[server_name]["connected"] = False
# 移除该服务器的工具
to_remove = [
name for name, spec in self._discovered_tools.items()
if server_name in spec.tags
]
for name in to_remove:
del self._discovered_tools[name]
del self._servers[server_name]
def get_discovered_tools(self) -> list[ToolSpec]:
"""获取所有已发现的MCP工具"""
return list(self._discovered_tools.values())
4.4 运行时语义搜索发现
当工具数量很多时，关键词匹配不够精确。运行时语义搜索使用向量嵌入和相似度计算来发现最相关的工具。
class SemanticToolDiscovery:
"""基于语义相似度的工具发现"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self._embeddings: dict[str, list[float]] = {} # 工具名 -> 嵌入向量
self._initialized = False
def _get_embedding(self, text: str) -> list[float]:
"""
获取文本的嵌入向量
实际实现中会调用embedding模型API
这里使用简化的哈希模拟
"""
# 模拟128维向量
import hashlib
hash_val = hashlib.md5(text.encode()).hexdigest()
vector = [int(hash_val[i:i+2], 16) / 255 for i in range(0, 32, 1)]
# 扩展到128维
return (vector * 4)[:128]
def build_index(self) -> None:
"""构建工具嵌入索引"""
tools = self.registry.list_tools()
for tool in tools:
# 将名称、描述、标签组合成索引文本
index_text = f"{tool.name} {tool.description} {' '.join(tool.tags)}"
self._embeddings[tool.name] = self._get_embedding(index_text)
self._initialized = True
print(f"[SemanticDiscovery] 索引构建完成，共 {len(tools)} 个工具")
def discover(self, query: str, top_k: int = 5) -> list[tuple[float, ToolSpec]]:
"""
语义搜索发现工具
返回 (相似度分数, 工具规格) 列表
"""
if not self._initialized:
self.build_index()
query_embedding = self._get_embedding(query)
results = []
for tool_name, tool_embedding in self._embeddings.items():
similarity = self._cosine_similarity(query_embedding, tool_embedding)
tool = self.registry.get(tool_name)
if tool:
results.append((similarity, tool))
results.sort(key=lambda x: x[0], reverse=True)
return results[:top_k]
@staticmethod
def _cosine_similarity(v1: list[float], v2: list[float]) -> float:
"""计算余弦相似度"""
if len(v1) != len(v2):
return 0.0
dot = sum(a * b for a, b in zip(v1, v2))
norm1 = sum(a * a for a in v1) ** 0.5
norm2 = sum(b * b for b in v2) ** 0.5
if norm1 == 0 or norm2 == 0:
return 0.0
return dot / (norm1 * norm2)
4.5 三种发现模式对比
| 维度 | 静态注册发现 | MCP动态发现 | 运行时语义搜索 |
|---|---|---|---|
| 发现时机 | 启动时 | 运行时按需 | 运行时按需 |
| 匹配方式 | 名称/标签精确匹配 | 协议声明 | 语义相似度 |
| 适用规模 | 小规模（<50工具） | 中等规模 | 大规模（>100工具） |
| 延迟 | 极低 | 中等（需网络通信） | 较高（向量计算） |
| 准确性 | 高（精确匹配） | 高（协议保证） | 中（依赖嵌入质量） |
| 动态性 | 低（需重启） | 高（运行时发现） | 中（需重建索引） |
| 实现复杂度 | 低 | 中 | 高 |
通俗类比：
- 静态发现 = 在通讯录里找电话号码（你认识的人）
- MCP动态发现 = 在黄页上查找商家电话（你不认识但知道有这类服务）
- 语义搜索 = 用搜索引擎搜索"附近修水管的人"（模糊需求，语义匹配）
五、工具选择策略
5.1 工具选择的核心挑战
工具发现找到了候选工具，工具选择则决定"用哪个"。当多个工具都能完成类似任务时，如何选择最优工具？
5.2 LLM自主选择
LLM自主选择是最灵活的方式：将所有可用工具的描述提供给LLM，让LLM根据任务上下文自主决策使用哪个工具。
class LLMToolSelector:
"""LLM自主工具选择器"""
def __init__(self, registry: ToolRegistry, llm_client=None):
self.registry = registry
self.llm_client = llm_client
def build_selection_prompt(self, task: str, available_tools: list[ToolSpec]) -> str:
"""构建工具选择提示词"""
tool_descriptions = []
for i, tool in enumerate(available_tools):
params_str = ""
if tool.parameters.get("properties"):
params_str = "\n 参数: " + json.dumps(
tool.parameters.get("properties", {}),
ensure_ascii=False
)
tool_descriptions.append(
f" {i+1}. {tool.name}: {tool.description}{params_str}"
)
prompt = f"""你是一个智能助手，需要从以下工具中选择合适的工具来完成任务。
任务: {task}
可用工具:
{chr(10).join(tool_descriptions)}
请分析任务需求，选择最合适的工具，并以JSON格式输出:
{{
"tool_name": "工具名称",
"reason": "选择理由",
"parameters": {{
"参数名": "参数值"
}}
}}
"""
return prompt
async def select(self, task: str, candidate_tools: list[ToolSpec] = None) -> dict:
"""
LLM选择工具
返回: {"tool_name": ..., "reason": ..., "parameters": ...}
"""
if candidate_tools is None:
candidate_tools = self.registry.list_tools()
prompt = self.build_selection_prompt(task, candidate_tools)
if self.llm_client:
response = await self.llm_client.chat(prompt)
# 解析LLM返回的JSON
try:
return json.loads(response)
except json.JSONDecodeError:
return {"error": "LLM返回格式错误", "raw": response}
else:
# 无LLM时的降级：返回所有工具供人工选择
return {
"candidates": [t.name for t in candidate_tools],
"note": "无LLM可用，请手动选择"
}
5.3 规则匹配选择
规则匹配使用预定义的规则将任务类型映射到工具。速度快、可预测，但灵活性差。
class RuleBasedToolSelector:
"""基于规则的工具选择器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self._rules: list[dict] = []
def add_rule(self, keywords: list[str], tool_name: str,
priority: int = 0, params_template: dict = None) -> None:
"""添加匹配规则"""
self._rules.append({
"keywords": keywords,
"tool_name": tool_name,
"priority": priority,
"params_template": params_template or {}
})
# 按优先级排序
self._rules.sort(key=lambda r: r["priority"], reverse=True)
def select(self, task: str) -> Optional[dict]:
"""
根据任务匹配工具
返回: {"tool_name": ..., "parameters": ...}
"""
task_lower = task.lower()
for rule in self._rules:
for keyword in rule["keywords"]:
if keyword.lower() in task_lower:
return {
"tool_name": rule["tool_name"],
"parameters": rule["params_template"],
"matched_keyword": keyword
}
return None
def setup_default_rules(self) -> None:
"""设置默认规则集"""
self.add_rule(
keywords=["搜索", "查找", "查询信息", "search"],
tool_name="web_search",
priority=10
)
self.add_rule(
keywords=["计算", "算数", "calculator"],
tool_name="calculator",
priority=10
)
self.add_rule(
keywords=["读文件", "读取文件", "查看文件", "read file"],
tool_name="file_read",
priority=10
)
self.add_rule(
keywords=["写文件", "保存文件", "write file"],
tool_name="file_write",
priority=10
)
self.add_rule(
keywords=["执行代码", "运行代码", "run code", "execute"],
tool_name="code_execute",
priority=5
)
5.4 向量检索选择
向量检索选择将任务描述和工具描述都转化为向量，通过相似度计算选择最匹配的工具。
class VectorToolSelector:
"""基于向量检索的工具选择器"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.semantic_discovery = SemanticToolDiscovery(registry)
async def select(self, task: str, top_k: int = 3) -> list[dict]:
"""
向量检索选择工具
返回排序后的工具候选列表
"""
results = self.semantic_discovery.discover(task, top_k=top_k)
candidates = []
for score, tool in results:
candidates.append({
"tool_name": tool.name,
"description": tool.description,
"similarity_score": round(score, 4),
"parameters": tool.parameters
})
return candidates
5.5 混合选择策略
实际系统中，通常会组合多种策略以获得最佳效果：
class HybridToolSelector:
"""混合工具选择器 - 融合多种策略"""
def __init__(self, registry: ToolRegistry, llm_client=None):
self.registry = registry
self.rule_selector = RuleBasedToolSelector(registry)
self.rule_selector.setup_default_rules()
self.vector_selector = VectorToolSelector(registry)
self.llm_selector = LLMToolSelector(registry, llm_client)
async def select(self, task: str) -> dict:
"""
混合选择策略:
1. 先用规则匹配（快速路径）
2. 规则未命中时用向量检索缩小范围
3. 用LLM从候选中做最终决策
"""
# 步骤1: 规则匹配
rule_result = self.rule_selector.select(task)
if rule_result:
return {
"strategy": "rule_match",
**rule_result
}
# 步骤2: 向量检索
vector_candidates = await self.vector_selector.select(task, top_k=5)
if not vector_candidates:
return {"error": "未找到匹配工具", "task": task}
# 步骤3: LLM决策
candidate_tools = []
for candidate in vector_candidates:
tool = self.registry.get(candidate["tool_name"])
if tool:
candidate_tools.append(tool)
llm_result = await self.llm_selector.select(task, candidate_tools)
return {
"strategy": "hybrid",
"vector_candidates": vector_candidates,
"llm_decision": llm_result
}
5.6 三种选择策略对比
| 维度 | LLM自主选择 | 规则匹配 | 向量检索 |
|---|---|---|---|
| 灵活性 | 最高 | 低 | 中 |
| 响应速度 | 慢（需LLM推理） | 极快 | 较快 |
| 准确性 | 高（理解上下文） | 高（精确匹配时） | 中（依赖嵌入质量） |
| 可解释性 | 中 | 高 | 低 |
| 维护成本 | 低 | 高（需维护规则） | 中（需重建索引） |
| 适用场景 | 复杂任务 | 明确意图 | 模糊需求 |
通俗类比：
- LLM选择 = 让一个有经验的员工自主决定用什么工具（灵活但慢）
- 规则匹配 = 查操作手册"遇到A情况用B工具"（快但不灵活）
- 向量检索 = 在搜索引擎输入需求，看搜索结果推荐什么（折中方案）
六、工具参数生成与验证
6.1 参数生成的挑战
LLM决定使用某个工具后，需要生成该工具的调用参数。参数生成是工具调用中最容易出错的环节——LLM可能生成错误类型的参数、遗漏必需参数、或包含多余参数。
每工具执行前必须参数校验。使用JSON Schema或Pydantic：
def validate(schema: dict, args: dict):
required = schema.get("required", [])
for r in required:
if r not in args:
raise ValueError(f"缺少必需参数:{r}")
return {k: v for k, v in args.items() if k in schema.get("properties", {})}
6.2 完整参数验证器
import re
from typing import Any, Union
class ParameterValidator:
"""工具参数验证器 - 基于JSON Schema"""
def __init__(self):
self._type_validators = {
"string": self._validate_string,
"integer": self._validate_integer,
"number": self._validate_number,
"boolean": self._validate_boolean,
"array": self._validate_array,
"object": self._validate_object,
}
def validate(self, schema: dict, args: dict) -> dict:
"""
验证参数是否符合Schema
返回验证后的参数（经过类型转换和默认值填充）
异常时抛出 ValidationError
"""
errors = []
validated = {}
properties = schema.get("properties", {})
required = schema.get("required", [])
# 检查必需参数
for req in required:
if req not in args:
errors.append(f"缺少必需参数: {req}")
# 验证每个参数
for param_name, param_value in args.items():
if param_name not in properties:
if schema.get("additionalProperties", True) is False:
errors.append(f"未知参数: {param_name}")
continue
param_schema = properties[param_name]
try:
validated_value = self._validate_value(param_value, param_schema)
validated[param_name] = validated_value
except ValueError as e:
errors.append(f"参数 {param_name} 验证失败: {e}")
# 填充默认值
for param_name, param_schema in properties.items():
if param_name not in validated and "default" in param_schema:
validated[param_name] = param_schema["default"]
if errors:
raise ValueError("; ".join(errors))
return validated
def _validate_value(self, value: Any, schema: dict) -> Any:
"""验证单个值"""
expected_type = schema.get("type")
if expected_type and expected_type in self._type_validators:
return self._type_validators[expected_type](value, schema)
return value
def _validate_string(self, value: Any, schema: dict) -> str:
"""验证字符串"""
if not isinstance(value, str):
# 尝试转换
value = str(value)
# 最小长度
min_len = schema.get("minLength")
if min_len is not None and len(value) < min_len:
raise ValueError(f"字符串长度不能小于 {min_len}")
# 最大长度
max_len = schema.get("maxLength")
if max_len is not None and len(value) > max_len:
raise ValueError(f"字符串长度不能超过 {max_len}")
# 正则模式
pattern = schema.get("pattern")
if pattern and not re.match(pattern, value):
raise ValueError(f"字符串不匹配模式: {pattern}")
# 枚举值
enum_values = schema.get("enum")
if enum_values and value not in enum_values:
raise ValueError(f"值必须是 {enum_values} 之一")
return value
def _validate_integer(self, value: Any, schema: dict) -> int:
"""验证整数"""
try:
value = int(value)
except (ValueError, TypeError):
raise ValueError(f"无法转换为整数: {value}")
minimum = schema.get("minimum")
if minimum is not None and value < minimum:
raise ValueError(f"值不能小于 {minimum}")
maximum = schema.get("maximum")
if maximum is not None and value > maximum:
raise ValueError(f"值不能大于 {maximum}")
return value
def _validate_number(self, value: Any, schema: dict) -> float:
"""验证数字"""
try:
value = float(value)
except (ValueError, TypeError):
raise ValueError(f"无法转换为数字: {value}")
minimum = schema.get("minimum")
if minimum is not None and value < minimum:
raise ValueError(f"值不能小于 {minimum}")
maximum = schema.get("maximum")
if maximum is not None and value > maximum:
raise ValueError(f"值不能大于 {maximum}")
return value
def _validate_boolean(self, value: Any, schema: dict) -> bool:
"""验证布尔值"""
if isinstance(value, bool):
return value
if isinstance(value, str):
if value.lower() in ("true", "1", "yes"):
return True
elif value.lower() in ("false", "0", "no"):
return False
if isinstance(value, (int, float)):
return bool(value)
raise ValueError(f"无法转换为布尔值: {value}")
def _validate_array(self, value: Any, schema: dict) -> list:
"""验证数组"""
if not isinstance(value, list):
raise ValueError(f"期望数组类型，得到: {type(value)}")
min_items = schema.get("minItems")
if min_items is not None and len(value) < min_items:
raise ValueError(f"数组至少需要 {min_items} 个元素")
max_items = schema.get("maxItems")
if max_items is not None and len(value) > max_items:
raise ValueError(f"数组最多 {max_items} 个元素")
item_schema = schema.get("items")
if item_schema:
validated_items = []
for i, item in enumerate(value):
try:
validated_items.append(self._validate_value(item, item_schema))
except ValueError as e:
raise ValueError(f"数组第 {i} 个元素验证失败: {e}")
return validated_items
return value
def _validate_object(self, value: Any, schema: dict) -> dict:
"""验证对象"""
if not isinstance(value, dict):
raise ValueError(f"期望对象类型，得到: {type(value)}")
# 递归验证嵌套对象
return self.validate(schema, value)
6.3 参数生成流程
6.4 Pydantic 验证方案
除了手写验证器，还可以使用Pydantic进行更优雅的参数验证：
from pydantic import BaseModel, Field, validator
from typing import Optional
class SearchParams(BaseModel):
"""搜索工具参数（Pydantic模型）"""
query: str = Field(..., min_length=1, max_length=500, description="搜索关键词")
num_results: int = Field(default=10, ge=1, le=50, description="返回结果数量")
language: str = Field(default="zh", description="搜索语言")
time_range: str = Field(default="all", description="时间范围")
@validator("language")
def validate_language(cls, v):
allowed = ["zh", "en", "ja", "ko"]
if v not in allowed:
raise ValueError(f"语言必须是 {allowed} 之一")
return v
@validator("time_range")
def validate_time_range(cls, v):
allowed = ["day", "week", "month", "year", "all"]
if v not in allowed:
raise ValueError(f"时间范围必须是 {allowed} 之一")
return v
# 使用示例
try:
params = SearchParams(query="Python教程", num_results=5)
print(f"验证通过: {params.dict()}")
except Exception as e:
print(f"验证失败: {e}")
通俗类比：参数验证就像机场安检。你的行李（参数）必须通过尺寸检查（类型验证）、重量检查（范围验证）、禁带物品检查（枚举验证），通过后才能登机（执行工具）。安检人员（验证器）会告诉你哪里不合格，你需要修正后重新通过。
七、工具执行引擎
7.1 执行引擎的职责
工具调度层根据LLM输出解析工具名称与参数，走校验后执行：
async def route(tool_name: str, args: dict, registry: ToolRegistry):
tool = registry._tools.get(tool_name)
if not tool:
return f"错误:工具{tool_name}不存在"
validated = validate(tool.parameters, args)
return await tool.function(**validated)
执行引擎不仅要调用工具函数，还要管理整个执行过程：超时控制、资源限制、错误处理、并发管理、沙箱隔离。
7.2 执行引擎完整实现
import asyncio
import signal
import resource
import traceback
from typing import Any
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from enum import Enum
class ExecutionStatus(Enum):
"""执行状态"""
SUCCESS = "success"
TIMEOUT = "timeout"
ERROR = "error"
CANCELLED = "cancelled"
RATE_LIMITED = "rate_limited"
@dataclass
class ExecutionResult:
"""工具执行结果"""
status: ExecutionStatus
result: Any = None
error: str = ""
duration: float = 0.0
tool_name: str = ""
parameters: dict = None
def to_dict(self) -> dict:
return {
"status": self.status.value,
"result": self.result,
"error": self.error,
"duration": round(self.duration, 3),
"tool_name": self.tool_name,
}
class ToolExecutor:
"""工具执行引擎"""
def __init__(self, registry: ToolRegistry, validator: ParameterValidator = None):
self.registry = registry
self.validator = validator or ParameterValidator()
self._executor = ThreadPoolExecutor(max_workers=10)
self._call_history: dict[str, list[float]] = {} # 工具名 -> 调用时间戳列表
self._result_cache: dict[str, tuple[Any, float]] = {} # 缓存key -> (结果, 过期时间)
async def execute(self, tool_name: str, args: dict,
context: dict = None) -> ExecutionResult:
"""
执行工具
这是执行引擎的主入口
"""
import time
start_time = time.time()
# 1. 查找工具
tool = self.registry.get(tool_name)
if not tool:
return ExecutionResult(
status=ExecutionStatus.ERROR,
error=f"工具不存在: {tool_name}",
tool_name=tool_name,
duration=time.time() - start_time
)
# 2. 检查工具状态
if tool.status != ToolStatus.ACTIVE:
return ExecutionResult(
status=ExecutionStatus.ERROR,
error=f"工具不可用（状态: {tool.status.value}）",
tool_name=tool_name,
duration=time.time() - start_time
)
# 3. 限流检查
if tool.rate_limit:
if not self._check_rate_limit(tool_name, tool.rate_limit):
return ExecutionResult(
status=ExecutionStatus.RATE_LIMITED,
error=f"工具调用频率超限（限制: {tool.rate_limit}/分钟）",
tool_name=tool_name,
duration=time.time() - start_time
)
# 4. 参数验证
try:
validated_args = self.validator.validate(tool.parameters, args)
except ValueError as e:
return ExecutionResult(
status=ExecutionStatus.ERROR,
error=f"参数验证失败: {e}",
tool_name=tool_name,
parameters=args,
duration=time.time() - start_time
)
# 5. 权限检查
if tool.requires_confirmation:
if context and not context.get("confirmed", False):
return ExecutionResult(
status=ExecutionStatus.ERROR,
error="此工具需要用户确认才能执行",
tool_name=tool_name,
duration=time.time() - start_time
)
# 6. 缓存检查
cache_key = self._make_cache_key(tool_name, validated_args)
if tool.cacheable and cache_key in self._result_cache:
cached_result, expire_time = self._result_cache[cache_key]
if time.time() < expire_time:
return ExecutionResult(
status=ExecutionStatus.SUCCESS,
result=cached_result,
tool_name=tool_name,
duration=0.0,
)
# 7. 执行工具（带超时控制）
try:
result = await self._execute_with_timeout(
tool.func, validated_args, tool.timeout
)
# 更新工具统计
duration = time.time() - start_time
self._update_tool_stats(tool, duration)
# 缓存结果
if tool.cacheable:
self._result_cache[cache_key] = (result, time.time() + tool.cache_ttl)
return ExecutionResult(
status=ExecutionStatus.SUCCESS,
result=result,
tool_name=tool_name,
parameters=validated_args,
duration=duration
)
except asyncio.TimeoutError:
return ExecutionResult(
status=ExecutionStatus.TIMEOUT,
error=f"工具执行超时（{tool.timeout}秒）",
tool_name=tool_name,
duration=time.time() - start_time
)
except asyncio.CancelledError:
return ExecutionResult(
status=ExecutionStatus.CANCELLED,
error="工具执行被取消",
tool_name=tool_name,
duration=time.time() - start_time
)
except Exception as e:
return ExecutionResult(
status=ExecutionStatus.ERROR,
error=f"{type(e).__name__}: {str(e)}",
tool_name=tool_name,
duration=time.time() - start_time
)
async def _execute_with_timeout(self, func: Callable, args: dict,
timeout: float) -> Any:
"""带超时控制执行工具"""
if asyncio.iscoroutinefunction(func):
# 异步函数直接等待
return await asyncio.wait_for(func(**args), timeout=timeout)
else:
# 同步函数在线程池中执行
loop = asyncio.get_event_loop()
future = loop.run_in_executor(self._executor, func, **args)
return await asyncio.wait_for(future, timeout=timeout)
def _check_rate_limit(self, tool_name: str, rate_limit: int) -> bool:
"""检查调用频率"""
import time
now = time.time()
window = 60.0 # 1分钟窗口
if tool_name not in self._call_history:
self._call_history[tool_name] = []
# 清理过期记录
self._call_history[tool_name] = [
t for t in self._call_history[tool_name] if now - t < window
]
if len(self._call_history[tool_name]) >= rate_limit:
return False
self._call_history[tool_name].append(now)
return True
@staticmethod
def _make_cache_key(tool_name: str, args: dict) -> str:
"""生成缓存键"""
return f"{tool_name}:{json.dumps(args, sort_keys=True)}"
@staticmethod
def _update_tool_stats(tool: ToolSpec, duration: float) -> None:
"""更新工具统计数据"""
tool.call_count += 1
tool.last_called = time.time()
# 移动平均
if tool.avg_latency == 0:
tool.avg_latency = duration
else:
tool.avg_latency = (tool.avg_latency * 0.9 + duration * 0.1)
async def execute_batch(self, calls: list[dict]) -> list[ExecutionResult]:
"""批量并行执行工具调用"""
tasks = [
self.execute(call["tool_name"], call.get("args", {}), call.get("context"))
for call in calls
]
return await asyncio.gather(*tasks, return_exceptions=True)
def clear_cache(self) -> None:
"""清空结果缓存"""
self._result_cache.clear()
def get_cache_stats(self) -> dict:
"""获取缓存统计"""
import time
now = time.time()
active = sum(1 for _, exp in self._result_cache.values() if exp > now)
return {
"total_entries": len(self._result_cache),
"active_entries": active,
"expired_entries": len(self._result_cache) - active,
}
7.3 执行引擎架构
7.4 沙箱隔离执行
对于不可信的工具或高权限操作，执行引擎应支持沙箱隔离：
class SandboxedExecutor:
"""沙箱隔离执行器"""
def __init__(self, max_memory_mb: int = 256, max_cpu_seconds: int = 30):
self.max_memory_mb = max_memory_mb
self.max_cpu_seconds = max_cpu_seconds
def _set_resource_limits(self):
"""设置进程资源限制"""
try:
# 内存限制
mem_bytes = self.max_memory_mb * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
# CPU时间限制
resource.setrlimit(resource.RLIMIT_CPU, (self.max_cpu_seconds, self.max_cpu_seconds))
except (resource.error, AttributeError):
# Windows不支持resource模块，跳过
pass
async def execute_sandboxed(self, func: Callable, args: dict,
timeout: float = 30.0) -> Any:
"""在沙箱中执行函数"""
loop = asyncio.get_event_loop()
def run_in_sandbox():
self._set_resource_limits()
return func(**args)
try:
future = loop.run_in_executor(None, run_in_sandbox)
return await asyncio.wait_for(future, timeout=timeout)
except asyncio.TimeoutError:
raise TimeoutError(f"沙箱执行超时（{timeout}秒）")
except Exception as e:
raise RuntimeError(f"沙箱执行失败: {e}")
通俗类比：执行引擎就像工厂的生产车间。工具函数是生产线上的机器，执行引擎是车间主任。主任负责：确认机器可用（状态检查）、确认工人有操作资格（权限检查）、控制生产节奏不要太快（限流）、检查原材料合格（参数验证）、设置安全护栏（超时控制）、记录生产数据（统计更新）。
八、工具结果处理
8.1 结果处理的重要性
工具执行完成后，原始结果可能格式不统一、过长、包含敏感信息或需要进一步加工。结果处理器负责将原始结果转化为LLM和用户可理解的形式。
8.2 结果处理器实现
import json
from typing import Any, Optional
class ResultProcessor:
"""工具结果处理器"""
def __init__(self, max_length: int = 4000, max_array_items: int = 20):
self.max_length = max_length
self.max_array_items = max_array_items
self._formatters: dict[str, Callable] = {}
self._filters: list[Callable] = []
def register_formatter(self, tool_name: str, formatter: Callable) -> None:
"""为特定工具注册自定义格式化器"""
self._formatters[tool_name] = formatter
def add_filter(self, filter_func: Callable) -> None:
"""添加结果过滤器（如脱敏）"""
self._filters.append(filter_func)
def process(self, result: Any, tool_name: str = "",
context: dict = None) -> dict:
"""
处理工具执行结果
返回标准化的结果字典
"""
context = context or {}
# 1. 应用过滤器（脱敏等）
for filter_func in self._filters:
result = filter_func(result, tool_name)
# 2. 自定义格式化
if tool_name in self._formatters:
result = self._formatters[tool_name](result)
else:
# 3. 默认格式化
result = self._default_format(result)
# 4. 截断处理
result = self._truncate(result)
# 5. 包装为标准格式
return {
"tool_name": tool_name,
"result": result,
"truncated": len(str(result)) > self.max_length,
"timestamp": time.time(),
}
def _default_format(self, result: Any) -> str:
"""默认格式化"""
if isinstance(result, str):
return result
elif isinstance(result, (dict, list)):
# 截断数组
if isinstance(result, list) and len(result) > self.max_array_items:
result = result[:self.max_array_items]
result.append({"_note": f"... 已截断，共 {len(result)} 条"})
return json.dumps(result, ensure_ascii=False, indent=2)
else:
return str(result)
def _truncate(self, result: str) -> str:
"""截断过长结果"""
if len(result) <= self.max_length:
return result
return result[:self.max_length] + f"\n... [结果已截断，原始长度: {len(result)}]"
def format_for_llm(self, processed: dict) -> str:
"""将处理后的结果格式化为LLM可读文本"""
lines = [
f"[工具: {processed['tool_name']}]",
f"结果:",
processed["result"],
]
if processed.get("truncated"):
lines.append("(注意: 结果已截断)")
return "\n".join(lines)
def format_for_user(self, processed: dict) -> str:
"""将处理后的结果格式化为用户可读文本"""
return processed["result"]
8.3 结果缓存
class ResultCache:
"""工具结果缓存"""
def __init__(self, max_size: int = 1000, default_ttl: int = 300):
self._cache: dict[str, dict] = {}
self._max_size = max_size
self._default_ttl = default_ttl
self._hits = 0
self._misses = 0
def get(self, tool_name: str, args: dict) -> Optional[Any]:
"""获取缓存结果"""
key = self._make_key(tool_name, args)
if key in self._cache:
entry = self._cache[key]
if time.time() < entry["expire"]:
self._hits += 1
return entry["result"]
else:
del self._cache[key]
self._misses += 1
return None
def set(self, tool_name: str, args: dict, result: Any, ttl: int = None) -> None:
"""设置缓存"""
if len(self._cache) >= self._max_size:
self._evict()
key = self._make_key(tool_name, args)
self._cache[key] = {
"result": result,
"expire": time.time() + (ttl or self._default_ttl),
"created": time.time(),
}
def invalidate(self, tool_name: str) -> int:
"""使某工具的所有缓存失效"""
keys_to_remove = [k for k in self._cache if k.startswith(f"{tool_name}:")]
for k in keys_to_remove:
del self._cache[k]
return len(keys_to_remove)
def _evict(self):
"""LRU淘汰"""
oldest_key = min(self._cache, key=lambda k: self._cache[k]["created"])
del self._cache[oldest_key]
@staticmethod
def _make_key(tool_name: str, args: dict) -> str:
return f"{tool_name}:{json.dumps(args, sort_keys=True)}"
def stats(self) -> dict:
"""缓存统计"""
total = self._hits + self._misses
hit_rate = self._hits / total if total > 0 else 0
return {
"size": len(self._cache),
"max_size": self._max_size,
"hits": self._hits,
"misses": self._misses,
"hit_rate": round(hit_rate, 4),
}
8.4 结果处理流程
通俗类比：结果处理就像新闻编辑的工作。记者（工具函数）发回原始素材（原始结果），编辑要做的：检查有无敏感信息需要打码（过滤脱敏）、把素材整理成文章（格式化）、太长的内容删减到版面能放下（截断）、最后排版成读者爱看的样式（用户可读格式）。
九、MCP协议集成
9.1 MCP协议概述
MCP（Model Context Protocol）是一种标准化的工具发现与调用协议，旨在解决"Agent如何发现和使用外部工具"的通用问题。MCP定义了工具服务器（Tool Server）和工具客户端（Tool Client）之间的通信规范，使任何Agent都能通过统一协议接入任何工具服务。
MCP是工具标准层(第7章深入)。
9.2 MCP架构
9.3 MCP通信流程
9.4 MCP工具服务器实现
class MCPToolServer:
"""MCP工具服务器 - 提供工具给Agent使用"""
def __init__(self, server_name: str, version: str = "1.0.0"):
self.server_name = server_name
self.version = version
self._tools: dict[str, dict] = {}
self._handlers: dict[str, Callable] = {}
def register_tool(self, name: str, description: str,
input_schema: dict, handler: Callable) -> None:
"""注册MCP工具"""
self._tools[name] = {
"name": name,
"description": description,
"inputSchema": input_schema
}
self._handlers[name] = handler
def handle_request(self, request: dict) -> dict:
"""处理MCP请求"""
method = request.get("method")
req_id = request.get("id")
params = request.get("params", {})
if method == "initialize":
return self._handle_initialize(req_id, params)
elif method == "tools/list":
return self._handle_list(req_id, params)
elif method == "tools/call":
return self._handle_call(req_id, params)
else:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"未知方法: {method}"}
}
def _handle_initialize(self, req_id: int, params: dict) -> dict:
"""处理初始化请求"""
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": self.server_name,
"version": self.version
}
}
}
def _handle_list(self, req_id: int, params: dict) -> dict:
"""处理工具列表请求"""
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"tools": list(self._tools.values())
}
}
def _handle_call(self, req_id: int, params: dict) -> dict:
"""处理工具调用请求"""
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name not in self._handlers:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32602, "message": f"工具不存在: {tool_name}"}
}
try:
handler = self._handlers[tool_name]
result = handler(**arguments)
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"content": [
{"type": "text", "text": str(result)}
]
}
}
except Exception as e:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32603, "message": str(e)}
}
# 创建MCP服务器示例
mcp_server = MCPToolServer("search-server", "1.0.0")
def search_handler(query: str, num: int = 10):
"""搜索处理器"""
return f"搜索 '{query}' 的前 {num} 条结果"
mcp_server.register_tool(
name="search",
description="搜索互联网内容",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num": {"type": "integer", "default": 10}
},
"required": ["query"]
},
handler=search_handler
)
9.5 MCP客户端实现
class MCPClient:
"""MCP客户端 - 连接MCP服务器并使用其工具"""
def __init__(self):
self._connections: dict[str, dict] = {}
self._tools: dict[str, dict] = {}
self._tool_to_server: dict[str, str] = {}
async def connect(self, server_name: str, transport: dict) -> bool:
"""
连接MCP服务器
transport: {"type": "stdio", "command": "...", "args": [...]}
{"type": "http", "url": "http://..."}
"""
# 模拟连接和初始化
init_response = {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": server_name, "version": "1.0.0"}
}
self._connections[server_name] = {
"transport": transport,
"info": init_response,
"connected": True
}
# 自动发现工具
await self._discover_tools(server_name)
return True
async def _discover_tools(self, server_name: str) -> None:
"""发现服务器上的工具"""
# 模拟 tools/list 请求
mock_tools = [
{
"name": "search",
"description": "搜索互联网",
"inputSchema": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"]
}
}
]
for tool in mock_tools:
full_name = f"{server_name}.{tool['name']}"
self._tools[full_name] = tool
self._tool_to_server[full_name] = server_name
async def call_tool(self, tool_name: str, arguments: dict) -> Any:
"""调用MCP工具"""
if tool_name not in self._tools:
raise ValueError(f"工具不存在: {tool_name}")
server_name = self._tool_to_server[tool_name]
if server_name not in self._connections:
raise RuntimeError(f"服务器未连接: {server_name}")
# 模拟 tools/call 请求
short_name = tool_name.split(".")[-1]
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": short_name, "arguments": arguments}
}
# 模拟响应
mock_result = {
"content": [
{"type": "text", "text": f"工具 {tool_name} 执行结果"}
]
}
return mock_result
async def disconnect(self, server_name: str) -> None:
"""断开服务器连接"""
if server_name in self._connections:
# 移除该服务器的工具
to_remove = [
name for name, srv in self._tool_to_server.items()
if srv == server_name
]
for name in to_remove:
del self._tools[name]
del self._tool_to_server[name]
del self._connections[server_name]
def list_tools(self) -> list[dict]:
"""列出所有已发现的MCP工具"""
return list(self._tools.values())
9.6 MCP与本地注册表的融合
class UnifiedToolRegistry:
"""统一工具注册表 - 融合本地工具和MCP工具"""
def __init__(self):
self.local_registry = ToolRegistry()
self.mcp_client = MCPClient()
self._all_tools: dict[str, ToolSpec] = {}
def register_local(self, name: str, description: str = "",
parameters: dict = None, **kwargs) -> Callable:
"""注册本地工具"""
return self.local_registry.register(name, description, parameters, **kwargs)
async def connect_mcp(self, server_name: str, transport: dict) -> bool:
"""连接MCP服务器并导入工具"""
success = await self.mcp_client.connect(server_name, transport)
if success:
# 将MCP工具转换为ToolSpec并注册
for tool_info in self.mcp_client.list_tools():
full_name = f"{server_name}.{tool_info['name']}"
spec = ToolSpec(
name=full_name,
description=tool_info["description"],
func=self._create_mcp_wrapper(full_name),
parameters=tool_info.get("inputSchema", {}),
tags=["mcp", server_name],
)
self._all_tools[full_name] = spec
return success
def _create_mcp_wrapper(self, tool_name: str) -> Callable:
"""创建MCP工具的本地包装函数"""
async def wrapper(**kwargs):
result = await self.mcp_client.call_tool(tool_name, kwargs)
# 提取文本内容
if isinstance(result, dict) and "content" in result:
texts = [c["text"] for c in result["content"] if c.get("type") == "text"]
return "\n".join(texts) if texts else str(result)
return str(result)
return wrapper
def get(self, name: str) -> Optional[ToolSpec]:
"""获取工具（本地或MCP）"""
if name in self._all_tools:
return self._all_tools[name]
return self.local_registry.get(name)
def list_all(self) -> list[ToolSpec]:
"""列出所有工具"""
local_tools = self.local_registry.list_tools()
mcp_tools = list(self._all_tools.values())
return local_tools + mcp_tools
def schemas(self) -> list[dict]:
"""返回所有工具的Schema"""
return [t.to_schema() for t in self.list_all()]
9.7 MCP协议消息格式
| MCP方法 | 方向 | 说明 |
|---|---|---|
| initialize | 客户端→服务器 | 初始化连接，交换能力信息 |
| tools/list | 客户端→服务器 | 请求工具列表 |
| tools/call | 客户端→服务器 | 调用指定工具 |
| notifications | 服务器→客户端 | 工具变更通知 |
通俗类比：MCP协议就像USB标准。以前每个设备都有自己的接口（串口、并口、PS/2…），现在统一用USB。MCP让所有工具服务都用同一种"语言"与Agent通信，Agent不需要为每个工具服务学一套新的对接方式。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/lza_csdn2019/article/details/167081697
