引言：工具是Agent的”双手”
如果说Brain层赋予了Agent思考的能力，Memory层赋予了它记忆，那么Tool层赋予的就是行动力。
一个没有工具的LLM，就像一个被封在玻璃箱里的天才——它能理解一切、规划一切，却无法触碰现实世界。它可以告诉你”你的服务器磁盘满了”，却无法帮你清理；它可以分析出”这段代码有个bug”，却无法帮你修复；它可以规划出完美的部署方案，却无法帮你执行哪怕一条命令。
Tool层是Agent与物理世界和数字世界交互的唯一通道。
在传统的软件架构中，我们有丰富的集成模式——REST API、RPC、消息队列、Webhook。但Agent的Tool层面临一个根本不同的挑战：调用决策不是程序员写死的，而是由LLM在运行时动态生成的。这意味着工具的定义、调用、安全、编排都需要全新的设计思路。
本文将深入Tool层的每一个关键设计决策：从Function Calling的底层协议，到工具设计模式，再到安全沙箱和编排调度。我们还会通过一个实战项目——构建一个文件管理工具——来串联所有知识点。
在开始之前，先回顾一下我们在六层架构中的位置：
`
┌─────────────────────────────────────┐
│ 6. Observability（可观测性） │
├─────────────────────────────────────┤
│ 5. Eval（评估） │
├─────────────────────────────────────┤
│ 4. Memory（记忆） │
├─────────────────────────────────────┤
│ 3. Tool（工具） │ ← 本篇 ★
├─────────────────────────────────────┤
│ 2. Context（上下文） │
├─────────────────────────────────────┤
│ 1. Prompt（提示词） │
└─────────────────────────────────────┘
`
Tool层向上承接Brain层的决策意图，向下对接外部系统和资源，是整个Agent架构中出错概率最高、安全风险最大、但也最能体现Agent价值的一层。
Tool层的核心职责
Tool层承担三项核心职责：定义（Define）、执行（Execute）、管控（Govern）。
1 定义：让LLM理解工具
工具的第一个挑战是可发现性（Discoverability）。LLM需要在有限的Context Window中理解每个工具能做什么、需要什么参数、会返回什么结果。
一个工具定义通常包含：
`json
{
“name”: “read_file”,
“description”: “Read the contents of a text file. Returns the file content with line numbers.”,
“parameters”: {
“type”: “object”,
“properties”: {
“path”: {
“type”: “string”,
“description”: “Absolute path to the file”
},
“offset”: {
“type”: “integer”,
“description”: “Line number to start reading (1-indexed)”,
“default”: 1
},
“limit”: {
“type”: “integer”,
“description”: “Maximum lines to read”,
“default”: 500
}
},
“required”: [“path”]
}
}
`
这里有三个关键设计点：
- 名称要自解释：
read_file比rf或tool_03好一万倍。LLM通过名称就能推断功能。 - 描述要精准：不是越长越好，而是要包含使用场景和边界条件。
- 参数定义要完整：类型、描述、默认值、约束条件——LLM需要这些信息来正确构造调用。
2 执行：连接现实世界
工具的执行层负责将LLM生成的结构化调用转化为实际操作。这听起来简单，但在生产环境中充满挑战：
`
LLM输出: {“name”: “write_file”, “arguments”: {“path”: “/etc/nginx/nginx.conf”, “content”: “…”}}
│
▼
┌───────────────┐
│ Tool层执行引擎 │
│ │
│ 1. 解析参数 │
│ 2. 权限校验 │
│ 3. 输入验证 │
│ 4. 执行操作 │
│ 5. 结果封装 │
│ 6. 错误处理 │
└───────────────┘
│
▼
返回: {“success”: true, “content”: “文件已写入”}
`
每一步都可能失败，每一步都需要精心设计。
3 管控：安全与合规
这是最容易被忽视、却最重要的一环。当一个AI Agent拥有了文件操作、网络请求、代码执行等能力时，它实际上拥有了系统级权限。
一个被Prompt Injection攻击的Agent，可能比一个被入侵的服务器更危险——因为它会”自愿”执行恶意操作。
管控层需要回答的问题包括：
- 这个工具调用是否在用户授权范围内？
- 参数是否经过了安全验证？
- 执行结果是否可能泄露敏感信息？
- 调用频率是否在合理范围内？
我们将在第五节详细讨论这些安全设计。
Function Calling机制详解
Function Calling是Tool层的通信协议，它定义了LLM如何表达”我想调用某个工具”的意图。
1 OpenAI Function Calling
OpenAI的Function Calling是最早也是最广泛采用的方案。其核心思路是：在Chat Completion API中增加一个tools参数，让LLM在需要时输出结构化的工具调用请求。
请求格式：
`json
{
“model”: “gpt-4o”,
“messages”: [
{“role”: “user”, “content”: “帮我读取 /home/user/config.yaml 的内容”}
],
“tools”: [
{
“type”: “function”,
“function”: {
“name”: “read_file”,
“description”: “Read the contents of a text file”,
“parameters”: {
“type”: “object”,
“properties”: {
“path”: {“type”: “string”, “description”: “File path to read”}
},
“required”: [“path”]
}
}
}
]
}
`
LLM的响应：
`json
{
“choices”: [{
“message”: {
“role”: “assistant”,
“content”: null,
“tool_calls”: [
{
“id”: “call_abc123”,
“type”: “function”,
“function”: {
“name”: “read_file”,
“arguments”: “{“path”: “/home/user/config.yaml”}”
}
}
]
},
“finish_reason”: “tool_calls”
}]
}
`
注意几个细节：
content为null：当LLM决定调用工具时，它不会生成文本内容，而是直接输出工具调用。arguments是JSON字符串：不是JSON对象，需要二次解析。finish_reason为tool_calls：不同于正常的stop，表示需要执行工具后继续对话。tool_calls是数组：LLM可能同时请求多个工具调用（parallel function calling）。
执行后的回传：
`json
{
“role”: “tool”,
“tool_call_id”: “call_abc123”,
“content”: “1|version: ‘3’n2|services:n3| app:n4| image: nginx:latest…”
}
`
这个设计形成了一个请求-执行-回传的完整循环。Agent框架需要实现这个循环，直到LLM不再请求工具调用为止。
2 Anthropic Tool Use
Anthropic的Tool Use在概念上与OpenAI类似，但在细节上有所不同：
`json
{
“model”: “claude-sonnet-4-20250514”,
“messages”: [{“role”: “user”, “content”: “读取配置文件”}],
“tools”: [
{
“name”: “read_file”,
“description”: “Read the contents of a text file”,
“input_schema”: {
“type”: “object”,
“properties”: {
“path”: {“type”: “string”, “description”: “File path”}
},
“required”: [“path”]
}
}
]
}
`
关键差异：
| 特性 | OpenAI | Anthropic |
|---|
|——|——–|———–|
| 参数模式定义位置 | function.parameters |
input_schema（顶级） |
|---|---|---|
| 参数传递格式 | JSON字符串 | JSON对象 |
| 并行调用 | 支持 | 支持（2024年后） |
| 工具选择控制 | tool_choice |
tool_choice（type: auto/any/tool） |
| 结果回传 | role: tool |
role: user + tool_result content block |
Anthropic的一个独特设计是tool_result作为content block：
`json
{
“role”: “user”,
“content”: [
{
“type”: “tool_result”,
“tool_use_id”: “toolu_abc123”,
“content”: “文件内容…”
}
]
}
`
这意味着工具结果可以和用户文本消息混合传递，提供了更灵活的上下文构建方式。
3 结构化输出（Structured Output）
Function Calling的底层技术是Constrained Decoding（约束解码）——在生成JSON时，强制LLM的输出符合预定义的Schema。
OpenAI在2024年将这个能力独立出来，提供了response_format参数：
`json
{
“response_format”: {
“type”: “json_schema”,
“json_schema”: {
“name”: “file_operation”,
“schema”: {
“type”: “object”,
“properties”: {
“operation”: {“type”: “string”, “enum”: [“read”, “write”, “delete”]},
“path”: {“type”: “string”},
“content”: {“type”: “string”}
},
“required”: [“operation”, “path”]
}
}
}
}
`
结构化输出的价值不仅在于Tool层，它还是Agent输出格式化的基础——让LLM的输出可以直接被下游系统解析，无需正则或NLP后处理。
4 Function Calling的演进趋势
2026年的Function Calling正在向几个方向演进：
- 更细粒度的控制：不只是”能不能调用”，还有”什么时候调用”、”调用几次”、”调用失败后怎么办”。
- 流式工具调用：对于长时间运行的工具（如代码执行），支持流式返回中间结果。
- 工具调用的可解释性：不仅输出调用参数，还要输出”为什么选择这个工具”的推理过程。
- 多模态工具：工具不仅处理文本，还处理图像、音频、视频。
工具设计模式
好的工具设计直接决定了Agent的可靠性和效率。以下是三种核心设计模式。
1 单一职责模式（Single Responsibility）
每个工具只做一件事，但把这件事做好。
这是最重要的设计原则。对比以下两种设计：
反面示例——万能工具：
`json
{
“name”: “file_manager”,
“description”: “Manage files: read, write, delete, copy, move, list, search, compress, extract…”,
“parameters”: {
“properties”: {
“action”: {“type”: “string”, “enum”: [“read”, “write”, “delete”, “copy”, “move”, “list”, “search”, “compress”, “extract”]},
“…”: “还有十几个参数”
}
}
}
`
正面示例——拆分工具：
`json
{“name”: “read_file”, “description”: “Read file contents”, “parameters”: {“path”: “…”}},
{“name”: “write_file”, “description”: “Write content to file”, “parameters”: {“path”: “…”, “content”: “…”}},
{“name”: “list_directory”, “description”: “List files in directory”, “parameters”: {“path”: “…”}},
{“name”: “search_files”, “description”: “Search files by pattern”, “parameters”: {“pattern”: “…”, “path”: “…”}}
`
单一职责的好处：
- 更准确的调用：LLM面对5个明确的工具比面对1个有15种action的工具更容易做出正确选择。
- 更少的幻觉：参数越少、越明确，LLM构造错误参数的概率越低。
- 更易维护：修改一个工具不影响其他工具。
- 更精细的权限控制：可以单独控制每个工具的访问权限。
2 组合模式（Composition）
当一个复杂操作需要多个步骤时，提供原子工具 + 组合工具的双层设计：
`
组合层（高层） 原子层（底层）
┌──────────────┐ ┌──────────────┐
│ deploy_app │──分解为──→ │ git_pull │
│ │ │ docker_build │
│ │ │ docker_push │
│ │ │ kubectl_apply│
└──────────────┘ └──────────────┘
`
原子工具给LLM自由组合的灵活性，组合工具给常见场景提供效率和可靠性。
实现方式：
`python
class ToolRegistry:
def __init__(self):
self.atomics: dict[str, Tool] = {}
self.composites: dict[str, CompositeTool] = {}
def register_atomic(self, tool: Tool):
self.atomics[tool.name] = tool
def register_composite(self, name: str, steps: list[str]):
“””组合工具 = 原子工具的有序序列”””
tools = [self.atomics[s] for s in steps]
self.composites[name] = CompositeTool(name, tools)
def get_tools_for_llm(self, include_composites: bool = True) -> list[dict]:
“””根据场景选择暴露给LLM的工具集”””
tools = list(self.atomics.values())
if include_composites:
tools.extend(self.composites.values())
return [t.to_schema() for t in tools]
`
何时暴露组合工具？
- 用户请求明确指向一个复杂操作（”部署应用”）→ 暴露组合工具
- 用户请求需要灵活探索（”帮我看看这个项目”）→ 只暴露原子工具
- 工具数量已经很多 → 优先暴露组合工具以减少LLM的选择负担
3 降级模式（Fallback）
工具调用会失败。网络超时、权限不足、资源不可用——这些都是常态而非异常。降级模式要求每个工具都有Plan B。
`python
class ToolWithFallback:
def __init__(self, primary: Tool, fallback: Tool, conditions: list[type[Exception]]):
self.primary = primary
self.fallback = fallback
self.conditions = conditions
async def execute(self, **kwargs) -> ToolResult:
try:
return await self.primary.execute(**kwargs)
except tuple(self.conditions) as e:
logger.warning(f”Primary tool failed: {e}, falling back to {self.fallback.name}”)
return await self.fallback.execute(**kwargs)
`
常见的降级策略：
| 场景 | 主工具 | 降级方案 |
|---|
|——|——–|———-|
| 网络请求 | http_request |
cached_response（返回缓存） |
|---|---|---|
| 代码执行 | sandboxed_exec |
dry_run（只输出将执行的命令） |
| 数据库查询 | db_query |
db_read_replica（读副本） |
| 文件操作 | read_file |
file_metadata（只返回元数据） |
工具安全设计
安全是Tool层的第一优先级，没有之一。一个不安全的Tool层等于给Agent一把没有保险的枪。
1 权限分级
将工具按危险程度分级，不同级别应用不同的审批策略：
`
Level 0 – 只读（Read-Only）
├── read_file → 自动执行
├── list_directory → 自动执行
└── search_files → 自动执行
Level 1 – 低风险写入（Low-Risk Write）
├── write_file（用户目录） → 自动执行，记录日志
└── create_directory → 自动执行，记录日志
Level 2 – 中风险操作（Medium-Risk）
├── write_file（系统目录） → 需要用户确认
├── execute_command → 需要用户确认
└── network_request → 需要域名白名单
Level 3 – 高风险操作（High-Risk）
├── delete_file → 需要二次确认
├── modify_system_config → 需要管理员权限
└── access_credentials → 需要特殊授权
`
实现方式：
`python
from enum import IntEnum
from dataclasses import dataclass
class PermissionLevel(IntEnum):
READ_ONLY = 0
LOW_RISK_WRITE = 1
MEDIUM_RISK = 2
HIGH_RISK = 3
@dataclass
class ToolPermission:
level: PermissionLevel
requires_confirmation: bool = False
allowed_paths: list[str] | None = None
blocked_paths: list[str] | None = None
allowed_domains: list[str] | None = None
class PermissionChecker:
def __init__(self, user_role: str):
self.user_role = user_role
self.max_auto_level = self._get_max_auto_level()
def check(self, tool_name: str, args: dict, permission: ToolPermission) -> tuple[bool, str]:
# 检查权限级别
if permission.level > self.max_auto_level:
return False, f”Tool ‘{tool_name}’ requires level {permission.level} permission, max auto-approved is {self.max_auto_level}”
# 检查路径白名单/黑名单
if permission.blocked_paths and ‘path’ in args:
for blocked in permission.blocked_paths:
if args[‘path’].startswith(blocked):
return False, f”Path ‘{args[‘path’]}’ is in blocked list”
return True, “OK”
`
2 沙箱执行
对于代码执行类工具，必须在沙箱中运行。沙箱的核心目标是隔离——即使代码是恶意的，也无法影响宿主系统。
`
┌─────────────────────────────────────┐
│ 宿主系统 │
│ │
│ ┌─────────────────────────────┐ │
│ │ 沙箱环境 │ │
│ │ │ │
│ │ ┌───────────────────────┐ │ │
│ │ │ Agent执行代码 │ │ │
│ │ │ │ │ │
│ │ │ – 受限文件系统 │ │ │
│ │ │ – 受限网络访问 │ │ │
│ │ │ – 受限CPU/内存 │ │ │
│ │ │ – 受限执行时间 │ │ │
│ │ └───────────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────┘ │
│ │
└─────────────────────────────────────┘
`
主流沙箱方案对比：
| 方案 | 隔离级别 | 启动速度 | 资源开销 | 适用场景 |
|---|
|——|———-|———-|———-|———-|
| Docker Container | 进程+文件系统+网络 | 秒级 | 中 | 生产环境 |
|---|---|---|---|---|
| gVisor | 内核级 | 秒级 | 低-中 | 高安全需求 |
| Firecracker MicroVM | VM级 | 毫秒级 | 低 | 多租户 |
| nsjail/seccomp | 进程级 | 毫秒级 | 极低 | 轻量沙箱 |
| WebAssembly | 沙箱运行时 | 微秒级 | 极低 | 跨平台、插件 |
对于Agent场景，Docker + 资源限制是最实用的方案：
`python
import docker
class DockerSandbox:
def __init__(self):
self.client = docker.from_env()
async def execute(self, code: str, language: str = “python”, timeout: int = 30) -> ExecutionResult:
container = self.client.containers.run(
image=f”sandbox-{language}:latest”,
command=self._build_command(code, language),
detach=True,
mem_limit=”256m”,
cpu_period=100000,
cpu_quota=50000, # 50% CPU
network_disabled=True, # 禁用网络
read_only=True, # 只读文件系统
volumes={
‘/tmp/workspace’: {‘bind’: ‘/workspace’, ‘mode’: ‘rw’} # 只挂载工作目录
},
# 安全选项
security_opt=[“no-new-privileges”],
cap_drop=[“ALL”],
)
try:
result = container.wait(timeout=timeout)
stdout = container.logs(stdout=True, stderr=False).decode()
stderr = container.logs(stdout=False, stderr=True).decode()
return ExecutionResult(
success=result[‘StatusCode’] == 0,
stdout=stdout,
stderr=stderr,
exit_code=result[‘StatusCode’]
)
except Exception as e:
container.kill()
return ExecutionResult(success=False, error=f”Execution timeout or error: {e}”)
finally:
container.remove(force=True)
`
3 输入验证
LLM生成的参数不可信。它可能因为幻觉生成恶意路径，可能因为Prompt Injection被诱导执行危险操作，也可能只是简单地犯了格式错误。
验证层次：
`python
class InputValidator:
“””多层输入验证”””
def validate(self, tool_name: str, args: dict, schema: dict) -> ValidationResult:
# 第一层：Schema验证（类型、必填字段）
result = self._validate_schema(args, schema)
if not result.ok:
return result
# 第二层：语义验证（路径是否合理、URL是否合法）
result = self._validate_semantics(tool_name, args)
if not result.ok:
return result
# 第三层：安全验证（是否有注入、路径遍历等）
result = self._validate_security(tool_name, args)
if not result.ok:
return result
return ValidationResult(ok=True)
def _validate_security(self, tool_name: str, args: dict) -> ValidationResult:
# 路径遍历检测
if ‘path’ in args:
path = args[‘path’]
if ‘..’ in path:
return ValidationResult(ok=False, error=”Path traversal detected”)
if path.startswith(‘/etc’) or path.startswith(‘/proc’):
return ValidationResult(ok=False, error=”System path access denied”)
# 命令注入检测
if ‘command’ in args:
cmd = args[‘command’]
dangerous = [‘rm -rf /’, ‘dd if=’, ‘mkfs’, ‘:(){:|:&};:’]
for pattern in dangerous:
if pattern in cmd:
return ValidationResult(ok=False, error=f”Dangerous command pattern detected”)
return ValidationResult(ok=True)
`
4 速率限制
防止Agent在循环中反复调用同一个工具（这是最常见的Agent故障模式之一）：
`python
import time
from collections import defaultdict
class RateLimiter:
def __init__(self):
self.calls: dict[str, list[float]] = defaultdict(list)
self.limits = {
‘default’: {‘rpm’: 60, ‘per_call_timeout’: 30},
‘execute_command’: {‘rpm’: 10, ‘per_call_timeout’: 60},
‘network_request’: {‘rpm’: 30, ‘per_call_timeout’: 15},
‘write_file’: {‘rpm’: 20, ‘per_call_timeout’: 10},
}
def check(self, tool_name: str) -> tuple[bool, str]:
config = self.limits.get(tool_name, self.limits[‘default’])
now = time.time()
window = 60.0 # 1 minute window
# 清理过期记录
self.calls[tool_name] = [t for t in self.calls[tool_name] if now – t < window]
if len(self.calls[tool_name]) >= config[‘rpm’]:
return False, f”Rate limit exceeded for ‘{tool_name}’: {config[‘rpm’]} calls/minute”
self.calls[tool_name].append(now)
return True, “OK”
`
工具编排与调度
当Agent需要完成一个复杂任务时，它往往需要调用多个工具，而且这些工具之间存在依赖关系。工具编排层负责管理这些依赖和调度。
1 串行编排
最简单的模式——前一个工具的输出作为后一个工具的输入：
`
read_file(config_path) → 解析配置 → write_file(new_config) → execute_command(restart)
`
这种模式由Brain层的ReAct循环自然驱动，Tool层只需保证每个工具的正确执行。
2 并行编排
当多个工具调用之间没有依赖时，应该并行执行以提升效率：
`python
import asyncio
class ToolOrchestrator:
async def execute_parallel(self, calls: list[ToolCall]) -> list[ToolResult]:
“””并行执行多个无依赖的工具调用”””
tasks = [self._execute_single(call) for call in calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
result if isinstance(result, ToolResult)
else ToolResult(success=False, error=str(result))
for result in results
]
async def execute_sequential(self, calls: list[ToolCall]) -> list[ToolResult]:
“””串行执行，前一个的结果可用于后一个的参数”””
results = []
for call in calls:
# 支持参数引用前序结果
call = self._resolve_references(call, results)
result = await self._execute_single(call)
results.append(result)
if not result.success:
break # 失败则中断
return results
`
3 条件编排
根据前序工具的结果决定后续步骤：
`python
class ConditionalOrchestrator:
async def execute_with_condition(self, plan: ConditionalPlan) -> list[ToolResult]:
results = []
for step in plan.steps:
result = await self._execute_single(step.call)
results.append(result)
# 评估条件
if step.condition and not step.condition.evaluate(result):
if step.on_failure == “skip”:
continue
elif step.on_failure == “abort”:
break
elif step.on_failure == “fallback”:
result = await self._execute_single(step.fallback_call)
results.append(result)
return results
`
4 工具选择策略
当可用工具很多时（比如50+），把所有工具定义都塞进Context Window是不现实的。需要动态工具选择：
`python
class ToolSelector:
def __init__(self, all_tools: list[Tool]):
self.all_tools = {t.name: t for t in all_tools}
self.tool_embeddings = self._embed_descriptions(all_tools)
def select(self, user_message: str, max_tools: int = 10) -> list[Tool]:
“””根据用户消息语义选择最相关的工具”””
query_embedding = embed(user_message)
scores = {
name: cosine_similarity(query_embedding, emb)
for name, emb in self.tool_embeddings.items()
}
top_names = sorted(scores, key=scores.get, reverse=True)[:max_tools]
return [self.all_tools[name] for name in top_names]
`
MCP协议详解（Model Context Protocol）
2024年底，Anthropic提出了Model Context Protocol（MCP），这是一个标准化的Tool集成协议，正在成为行业事实标准。
1 MCP的核心思想
MCP解决的核心问题是工具生态的碎片化。在MCP之前，每个Agent框架都有自己的工具定义格式，每个工具提供者都需要为不同框架做适配。MCP提供了一个统一的协议：
`
工具提供者 MCP协议 Agent框架
┌──────────┐ JSON-RPC ┌──────────┐ Function ┌──────────┐
│ GitHub │ ◄────────────► │ MCP │ ◄───────────► │ Hermes │
│ Server │ over stdio │ Server │ Calling │ Agent │
└──────────┘ or SSE └──────────┘ └──────────┘
`
2 MCP协议架构
MCP采用Client-Server架构，运行在Agent进程中的是MCP Client，工具提供者是MCP Server：
`python
# MCP Server 示例（Python SDK）
from mcp.server import Server
from mcp.types import Tool, TextContent
server = Server(“file-manager”)
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name=”read_file”,
description=”Read the contents of a file”,
inputSchema={
“type”: “object”,
“properties”: {
“path”: {“type”: “string”, “description”: “File path”}
},
“required”: [“path”]
}
),
Tool(
name=”write_file”,
description=”Write content to a file”,
inputSchema={
“type”: “object”,
“properties”: {
“path”: {“type”: “string”},
“content”: {“type”: “string”}
},
“required”: [“path”, “content”]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == “read_file”:
content = await read_file(arguments[“path”])
return [TextContent(type=”text”, text=content)]
elif name == “write_file”:
await write_file(arguments[“path”], arguments[“content”])
return [TextContent(type=”text”, text=”File written successfully”)]
`
3 MCP的三大原语
MCP定义了三种核心原语：
1. Tools（工具）：Agent可以调用的操作，如上所述。
2. Resources（资源）：Agent可以读取的数据源，类似REST的GET：
`python
@server.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri=”file:///home/user/project”,
name=”Project Directory”,
description=”The current project’s file structure”,
mimeType=”application/json”
)
]
@server.read_resource()
async def read_resource(uri: str) -> str:
if uri.startswith(“file://”):
path = uri[7:]
return json.dumps(list_directory(path))
`
3. Prompts（提示模板）：预定义的交互模板：
`python
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
return [
Prompt(
name=”code_review”,
description=”Review code for issues”,
arguments=[
PromptArgument(name=”language”, description=”Programming language”, required=True),
PromptArgument(name=”code”, description=”Code to review”, required=True)
]
)
]
`
4 MCP传输层
MCP支持两种传输方式：
- stdio：通过标准输入输出通信，适合本地工具（如文件操作、本地数据库）。
- SSE（Server-Sent Events）：通过HTTP通信，适合远程工具（如云API、共享服务）。
5 MCP的实际价值
MCP的最大价值在于工具生态的可复用性。一个MCP Server可以被任何支持MCP的Agent框架使用：
`
┌──────────┐
│ GitHub │
│ MCP │◄─── Hermes Agent
│ Server │◄─── Cursor
└──────────┘◄─── Claude Desktop
▲
│
┌──────────┐
│ Slack │
│ MCP │◄─── 同上所有客户端
│ Server │
└──────────┘
`
这意味着工具开发者只需实现一次MCP Server，就能被所有Agent框架调用。这是一个巨大的杠杆效应。
实战：构建一个文件管理工具
现在让我们把所有理论知识串联起来，构建一个完整的文件管理工具集。
1 设计工具集
基于单一职责原则，我们设计以下工具：
`python
TOOLS = [
{
“name”: “read_file”,
“description”: “Read the contents of a text file with line numbers. Supports pagination via offset and limit. Use this instead of cat/head/tail in terminal.”,
“parameters”: {
“type”: “object”,
“properties”: {
“path”: {
“type”: “string”,
“description”: “Path to the file (absolute, relative, or ~/path)”
},
“offset”: {
“type”: “integer”,
“description”: “Line number to start reading (1-indexed, default: 1)”,
“default”: 1
},
“limit”: {
“type”: “integer”,
“description”: “Maximum number of lines to read (default: 500, max: 2000)”,
“default”: 500
}
},
“required”: [“path”]
}
},
{
“name”: “write_file”,
“description”: “Write content to a file, completely replacing existing content. Creates parent directories automatically. Use ‘patch’ for targeted edits.”,
“parameters”: {
“type”: “object”,
“properties”: {
“path”: {“type”: “string”, “description”: “File path”},
“content”: {“type”: “string”, “description”: “Complete content to write”}
},
“required”: [“path”, “content”]
}
},
{
“name”: “search_files”,
“description”: “Search file contents or find files by name. Supports regex patterns.”,
“parameters”: {
“type”: “object”,
“properties”: {
“pattern”: {“type”: “string”, “description”: “Search pattern (regex for content, glob for filenames)”},
“path”: {“type”: “string”, “description”: “Directory to search in”},
“target”: {
“type”: “string”,
“enum”: [“content”, “files”],
“description”: “‘content’ searches inside files, ‘files’ searches by name”
}
},
“required”: [“pattern”]
}
}
]
`
2 实现执行引擎
`python
import asyncio
import os
import re
from pathlib import Path
from dataclasses import dataclass
@dataclass
class ToolResult:
success: bool
content: str
error: str | None = None
metadata: dict | None = None
class FileToolExecutor:
def __init__(self, allowed_root: str = “/home/user”, max_file_size: int = 10 * 1024 * 1024):
self.allowed_root = Path(allowed_root).resolve()
self.max_file_size = max_file_size
def _validate_path(self, path_str: str) -> Path:
“””验证并规范化路径”””
path = Path(path_str).expanduser().resolve()
# 防止路径遍历
if not str(path).startswith(str(self.allowed_root)):
raise PermissionError(f”Access denied: path must be under {self.allowed_root}”)
return path
async def read_file(self, path: str, offset: int = 1, limit: int = 500) -> ToolResult:
try:
filepath = self._validate_path(path)
if not filepath.exists():
return ToolResult(success=False, content=””, error=f”File not found: {path}”)
if filepath.stat().st_size > self.max_file_size:
return ToolResult(success=False, content=””, error=f”File too large (> {self.max_file_size} bytes)”)
lines = filepath.read_text().splitlines()
total = len(lines)
# 应用分页
start = max(0, offset – 1)
end = min(total, start + limit)
selected = lines[start:end]
# 添加行号
numbered = [f”{i+1}|{line}” for i, line in enumerate(selected, start=start)]
return ToolResult(
success=True,
content=”n”.join(numbered),
metadata={“total_lines”: total, “showing”: f”{start+1}-{end}”}
)
except PermissionError as e:
return ToolResult(success=False, content=””, error=str(e))
except Exception as e:
return ToolResult(success=False, content=””, error=f”Read error: {e}”)
async def write_file(self, path: str, content: str) -> ToolResult:
try:
filepath = self._validate_path(path)
# 创建父目录
filepath.parent.mkdir(parents=True, exist_ok=True)
# 写入文件
filepath.write_text(content)
return ToolResult(
success=True,
content=f”Successfully wrote {len(content)} bytes to {path}”,
metadata={“bytes_written”: len(content)}
)
except PermissionError as e:
return ToolResult(success=False, content=””, error=str(e))
except Exception as e:
return ToolResult(success=False, content=””, error=f”Write error: {e}”)
async def search_files(self, pattern: str, path: str = “.”, target: str = “content”) -> ToolResult:
try:
search_path = self._validate_path(path)
results = []
if target == “files”:
# 文件名搜索
for p in search_path.rglob(f”*{pattern}*”):
if len(results) >= 50:
break
results.append(str(p.relative_to(search_path)))
else:
# 内容搜索（使用正则）
regex = re.compile(pattern)
for p in search_path.rglob(“*”):
if not p.is_file() or p.stat().st_size > self.max_file_size:
continue
try:
content = p.read_text(errors=’ignore’)
for i, line in enumerate(content.splitlines(), 1):
if regex.search(line):
rel = str(p.relative_to(search_path))
results.append(f”{rel}:{i}: {line.strip()}”)
if len(results) >= 50:
break
except:
continue
if len(results) >= 50:
break
if not results:
return ToolResult(success=True, content=”No matches found.”)
return ToolResult(
success=True,
content=”n”.join(results),
metadata={“match_count”: len(results)}
)
except Exception as e:
return ToolResult(success=False, content=””, error=f”Search error: {e}”)
async def execute(self, tool_name: str, arguments: dict) -> ToolResult:
“””统一执行入口”””
dispatch = {
“read_file”: self.read_file,
“write_file”: self.write_file,
“search_files”: self.search_files,
}
if tool_name not in dispatch:
return ToolResult(success=False, content=””, error=f”Unknown tool: {tool_name}”)
return await dispatchtool_name
`
3 完整的Agent工具循环
`python
class AgentToolLoop:
def __init__(self, llm_client, tool_executor: FileToolExecutor, tools: list[dict]):
self.llm = llm_client
self.executor = tool_executor
self.tools = tools
self.max_iterations = 10
async def run(self, user_message: str) -> str:
messages = [{“role”: “user”, “content”: user_message}]
for iteration in range(self.max_iterations):
# 调用LLM
response = await self.llm.chat(
messages=messages,
tools=self.tools
)
# 如果LLM不需要调用工具，返回文本
if not response.tool_calls:
return response.content
# 执行工具调用
messages.append(response.to_message())
for tool_call in response.tool_calls:
result = await self.executor.execute(
tool_name=tool_call.function.name,
arguments=json.loads(tool_call.function.arguments)
)
# 回传结果
messages.append({
“role”: “tool”,
“tool_call_id”: tool_call.id,
“content”: result.content if result.success else f”Error: {result.error}”
})
return “Reached maximum tool call iterations.”
`
这个完整的循环展示了Tool层的核心工作流：接收LLM决策 → 验证 → 执行 → 回传 → 等待下一步决策。
常见陷阱
在Tool层的设计和实现中，有一些反复出现的陷阱值得注意。
工具数量爆炸
症状：注册了50+工具，LLM频繁选错工具或在相似工具间犹豫。
原因：工具越多，LLM的选择空间越大，出错概率越高。研究表明，当工具数量超过15-20个时，LLM的选择准确率显著下降。
解决方案：
- 严格遵循单一职责，但合并功能高度重叠的工具
- 使用动态工具选择（见第六节），根据上下文只暴露相关工具
- 提供工具分组或分类，帮助LLM缩小选择范围
权限过宽
症状：Agent被Prompt Injection攻击后，执行了删除文件、发送邮件等危险操作。
原因：开发阶段为了方便，给Agent开放了所有权限，上线后忘记收紧。
解决方案：
- 最小权限原则：默认只给只读权限，按需开放写入
- 路径白名单：只能操作用户工作目录，不能碰系统文件
- 操作审计：所有工具调用都记录日志，便于事后追溯
错误处理过于简单
症状：工具执行失败后，LLM陷入重试循环，反复调用同一个失败的工具。
原因：错误信息不够具体，LLM无法判断是应该修改参数重试、换一个工具、还是放弃。
解决方案：
`python
class ToolResult:
success: bool
content: str
error: str | None
error_type: str | None # “validation_error”, “permission_error”, “runtime_error”
retryable: bool # 是否值得重试
suggestion: str | None # 给LLM的操作建议
`
一个好的错误返回应该告诉LLM三件事：发生了什么、为什么会发生、应该怎么办。
工具描述歧义
症状：LLM在两个工具之间反复切换，或者用错误的工具做正确的事。
原因：工具描述不够清晰，存在功能重叠。
解决方案：
- 为每个工具写清”Use this when…”和”Do NOT use this when…”
- 在描述中包含示例输入和输出
- 定期用真实用户查询测试工具选择的准确性
忽视工具的幂等性
症状：网络抖动导致工具调用超时，Agent重试后执行了两次，创建了重复数据。
原因：工具不具备幂等性——多次调用和一次调用的效果不同。
解决方案：
- 为每个工具调用生成唯一ID
- 在执行前检查是否已经执行过
- 对于非幂等操作（如发送消息），实现去重机制
总结
Tool层是Agent从”思考者”变为”行动者”的关键。它的设计质量直接决定了Agent的能力边界和安全边界。
让我们回顾本文的核心要点：
架构层面：
- Tool层承担定义、执行、管控三项核心职责
- Function Calling是LLM与工具的通信协议，OpenAI和Anthropic各有特点
- MCP正在成为工具生态的标准化协议
设计层面：
- 单一职责：一个工具做一件事
- 组合模式：原子工具 + 组合工具的双层设计
- 降级模式：每个工具都有Plan B
安全层面：
- 权限分级：按危险程度分级审批
- 沙箱执行：隔离不可信代码
- 输入验证：LLM的输出不可信
- 速率限制：防止循环调用
实践层面：
- 动态工具选择应对工具数量膨胀
- 详尽的错误信息帮助LLM做出正确决策
- 幂等性设计应对网络不可靠
`
┌─────────────────────────────────────────────────┐
│ Tool层设计检查清单 │
├─────────────────────────────────────────────────┤
│ □ 每个工具职责单一、描述清晰 │
│ □ 参数Schema完整（类型、描述、默认值、约束） │
│ □ 权限分级明确，遵循最小权限原则 │
│ □ 危险操作有沙箱隔离 │
│ □ 输入验证覆盖Schema、语义、安全三层 │
│ □ 速率限制防止循环调用 │
│ □ 错误信息包含what/why/how三要素 │
│ □ 工具数量控制在合理范围内（动态选择） │
│ □ 所有工具有审计日志 │
│ □ 考虑幂等性和重试策略 │
└─────────────────────────────────────────────────┘
`
下篇预告
下一篇我们将深入Harness六层架构的第四层——Memory层（记忆层）。我们将探讨Agent如何从无状态的工具进化为有状态的智能体，包括短期记忆管理、长期记忆存储、记忆的衰减与强化机制。敬请期待。
*本文是Harness Engineering系列的第五篇。系列目录：*
发表回复