在前六篇文章中，我们的 Agent 已经拥有了多渠道接入、自主推理、动态技能和长短期记忆。但要让它真正“干活”，还需要一双能操控现实系统的双手——工具。OpenClaw 内置了 Shell 执行、浏览器自动化、HTTP 请求等工具，并通过沙箱保障安全。今天，我们将构建一个企业级工具系统，涵盖工具注册与发现、JSON Schema 标准化描述、安全沙箱、超时与熔断以及动态加载，让 Agent 安全地操控企业内部 API、数据库和文件系统。
系列文章：
手把手构建企业级 Agent 框架：从 OpenClaw 架构到自主实现
手把手构建企业级 Agent 框架（二）：Gateway 网关与多渠道接入
手把手构建企业级 Agent 框架（三）：Pi Agent 运行时与 ReAct 循环
手把手构建企业级 Agent 框架（四）：Skill 系统——知识注入与能力扩展
手把手构建企业级 Agent 框架（五）：多智能体并行与任务委派，突破单 Agent 的性能与上下文瓶颈
手把手构建企业级 Agent 框架（六）：双源记忆系统 — 让 Agent 真正记住你，跨越会话的智能回忆
一、工具系统的核心设计原则
OpenClaw 的工具哲学是：“强大的能力需要强大的约束”。在企业环境中，这对工具系统提出了更严格的要求：
- 声明式描述：每个工具必须提供标准的 JSON Schema，让 LLM 准确理解用途和参数。
- 最小权限：工具执行时只能访问其声明所需的资源，禁止越权。
- 资源隔离：高风险工具（如代码执行、Shell 命令）必须在沙箱中运行。
- 可观测：每次调用工具都必须记录审计日志，包括耗时、结果和调用者身份。
- 动态加载：新增工具无需重启服务，通过插件目录或远程注册中心热加载。
工具系统架构：
- ToolRegistry：核心注册表，管理工具定义和实例。
- ToolExecutor：执行层，封装超时、重试、审计逻辑。
- Sandbox：沙箱层，隔离高风险执行环境。
- ToolLoader：动态加载器，从插件目录发现工具。
二、架构图与数据流
执行流程：
- Agent 输出工具调用指令，包含工具名和参数。
ToolExecutor从注册表获取工具定义，检查调用者是否有权使用。- 决定执行模式：安全敏感工具进入沙箱，普通工具直接异步调用。
- 执行过程受超时控制，结果记录到审计日志后返回 Agent。
三、接口与数据结构设计
3.1 工具定义标准
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Callable
@dataclass
classToolDefinition:
name:str
description:str
parameters_schema: Dict # JSON Schema
fn: Callable # 异步或同步函数
required_roles: List[str]= field(default_factory=list)# RBAC
timeout_seconds:int=30
sandbox:bool=False# 是否需沙箱执行
tags: List[str]= field(default_factory=list)
3.2 ToolRegistry 接口
classToolRegistry:
def__init__(self):...
defregister(self, tool: ToolDefinition)->None:...
defget(self, name:str)-> Optional[ToolDefinition]:...
defget_all_schemas(self)-> List[Dict]:...
defunregister(self, name:str)->None:...
3.3 ToolExecutor 接口
classToolExecutor:
def__init__(self, registry: ToolRegistry, sandbox: Sandbox):...
asyncdefexecute(self, name:str, params: Dict, context:'SecurityContext')-> Any:...
四、代码实战：构建完整的工具系统
4.1 项目结构
eclaw-tools/
├── tool_registry.py # 注册表
├── tool_executor.py # 执行器（含超时、审计）
├── sandbox.py # 沙箱抽象层
├── builtin_tools/ # 内置工具插件目录
│ ├── __init__.py
│ ├── http_tool.py
│ ├── order_tool.py
│ └── file_tool.py
├── agent.py # 集成后的 Agent
└── test_tools.py
4.2 ToolRegistry 实现
# tool_registry.py
from typing import Dict, List, Optional
from dataclasses import dataclass, field
@dataclass
classToolDefinition:
name:str
description:str
parameters_schema: Dict
fn:callable
required_roles: List[str]= field(default_factory=list)
timeout_seconds:int=30
sandbox:bool=False
tags: List[str]= field(default_factory=list)
classToolRegistry:
def__init__(self):
self._tools: Dict[str, ToolDefinition]={}
defregister(self, tool: ToolDefinition):
self._tools[tool.name]= tool
defget(self, name:str)-> Optional[ToolDefinition]:
return self._tools.get(name)
defget_all_schemas(self)-> List[Dict]:
return[
{
"name": t.name,
"description": t.description,
"parameters": t.parameters_schema
}
for t in self._tools.values()
]
defunregister(self, name:str):
self._tools.pop(name,None)
deflist_tools(self)-> List[str]:
returnlist(self._tools.keys())
4.3 审计日志与安全上下文
# audit.py
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Dict
@dataclass
classAuditLog:
tool_name:str
params: Dict
result: Any
user_id:str
tenant_id:str
duration_ms:float
success:bool
error:str=""
timestamp:float= field(default_factory=time.time)
classAuditLogger:
def__init__(self):
self.logger = logging.getLogger("tool_audit")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler("tool_audit.log")
handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))
self.logger.addHandler(handler)
deflog(self, audit: AuditLog):
self.logger.info(f"tool={audit.tool_name} user={audit.user_id} "
f"duration={audit.duration_ms}ms success={audit.success} "
f"params={audit.params} result_summary={str(audit.result)[:100]}")
# security_context.py
from dataclasses import dataclass, field
from typing import List
@dataclass
classSecurityContext:
user_id:str
tenant_id:str
roles: List[str]= field(default_factory=list)
4.4 沙箱抽象层
# sandbox.py
import asyncio
import subprocess
import json
from typing import Any, Dict
classSandbox:
"""沙箱抽象基类"""
asyncdefexecute(self, tool_name:str, params: Dict, code:str=None)-> Any:
raise NotImplementedError
classDockerSandbox(Sandbox):
"""使用 Docker 容器隔离执行（生产推荐）"""
asyncdefexecute(self, tool_name:str, params: Dict, code:str=None)-> Any:
# 构造 docker run 命令，挂载只读文件系统，限制网络和 CPU
cmd =[
"docker","run","--rm",
"--network","none",
"--cpus","0.5",
"--memory","128m",
"eclaw-sandbox:latest",
"python","-c", code
]
proc =await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr =await asyncio.wait_for(proc.communicate(), timeout=30)
if proc.returncode !=0:
return{"error": stderr.decode()}
return{"result": stdout.decode()}
except asyncio.TimeoutError:
proc.kill()
return{"error":"沙箱执行超时"}
classProcessPoolSandbox(Sandbox):
"""进程池沙箱（轻量级，适用于非完全不可信代码）"""
asyncdefexecute(self, tool_name:str, params: Dict, code:str=None)-> Any:
try:
# 使用 subprocess 在独立进程中执行
proc =await asyncio.create_subprocess_exec(
"python","-c", code,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr =await asyncio.wait_for(proc.communicate(), timeout=15)
return{"stdout": stdout.decode(),"stderr": stderr.decode()}
except asyncio.TimeoutError:
return{"error":"进程执行超时"}
4.5 ToolExecutor 实现（核心）
# tool_executor.py
import asyncio
import time
from typing import Any, Dict
from tool_registry import ToolRegistry
from security_context import SecurityContext
from sandbox import Sandbox
from audit import AuditLogger, AuditLog
classToolExecutor:
def__init__(self, registry: ToolRegistry, sandbox: Sandbox =None):
self.registry = registry
self.sandbox = sandbox or ProcessPoolSandbox()
self.audit = AuditLogger()
asyncdefexecute(self, name:str, params: Dict, context: SecurityContext)-> Any:
tool = self.registry.get(name)
ifnot tool:
raise ValueError(f"工具 '{name}' 未注册")
# 权限检查
if tool.required_roles:
ifnotset(tool.required_roles)&set(context.roles):
raise PermissionError(f"用户 {context.user_id} 无权使用工具 {name}")
start = time.time()
success =True
error_msg =""
result =None
try:
if tool.sandbox:
# 沙箱执行模式（用于代码/Shell 类工具）
code = params.get("code","")
result =await asyncio.wait_for(
self.sandbox.execute(name, params, code),
timeout=tool.timeout_seconds
)
else:
# 普通直接执行
fn = tool.fn
if asyncio.iscoroutinefunction(fn):
result =await asyncio.wait_for(
fn(params, context),
timeout=tool.timeout_seconds
)
else:
result =await asyncio.wait_for(
asyncio.to_thread(fn, params, context),
timeout=tool.timeout_seconds
)
except asyncio.TimeoutError:
success =False
error_msg =f"工具 {name} 执行超时 ({tool.timeout_seconds}s)"
result ={"error": error_msg}
except PermissionError as e:
success =False
error_msg =str(e)
result ={"error": error_msg}
except Exception as e:
success =False
error_msg =str(e)
result ={"error": error_msg}
duration =(time.time()- start)*1000
self.audit.log(AuditLog(
tool_name=name,
params=params,
result=result,
user_id=context.user_id,
tenant_id=context.tenant_id,
duration_ms=duration,
success=success,
error=error_msg
))
ifnot success:
raise RuntimeError(error_msg)
return result
4.6 内置工具示例
# builtin_tools/http_tool.py
import aiohttp
from tool_registry import ToolDefinition
asyncdefhttp_get(params, ctx):
url = params["url"]
headers = params.get("headers",{})
asyncwith aiohttp.ClientSession()as session:
asyncwith session.get(url, headers=headers, timeout=10)as resp:
return{"status": resp.status,"body":await resp.text()}
http_tool_def = ToolDefinition(
name="http_get",
description="发送 HTTP GET 请求获取资源",
parameters_schema={
"type":"object",
"properties":{
"url":{"type":"string","description":"请求 URL"},
"headers":{"type":"object","description":"请求头"}
},
"required":["url"]
},
fn=http_get,
required_roles=["user"],
timeout_seconds=15
)
# builtin_tools/order_tool.py
from tool_registry import ToolDefinition
# 模拟订单数据库
ORDERS ={"O123":{"status":"已发货","eta":"2026-05-20"},"O456":{"status":"待发货"}}
asyncdefquery_order(params, ctx):
order_id = params.get("order_id")
order = ORDERS.get(order_id)
ifnot order:
return{"error":f"订单 {order_id} 不存在"}
return order
order_tool_def = ToolDefinition(
name="query_order",
description="查询订单状态，需要 order_id",
parameters_schema={
"type":"object",
"properties":{"order_id":{"type":"string"}},
"required":["order_id"]
},
fn=query_order,
required_roles=["user","agent"],
tags=["order","internal"]
)
# builtin_tools/file_tool.py
from tool_registry import ToolDefinition
from pathlib import Path
asyncdefread_file(params, ctx):
path = params["path"]
# 路径安全检查
if".."in path or path.startswith("/etc"):
return{"error":"不允许访问系统敏感路径"}
content = Path(path).read_text(encoding="utf-8")
return{"content": content[:2000]}# 限制返回长度
file_tool_def = ToolDefinition(
name="read_file",
description="读取文件内容（限制访问范围）",
parameters_schema={
"type":"object",
"properties":{"path":{"type":"string"}},
"required":["path"]
},
fn=read_file,
required_roles=["admin"],# 仅管理员可读文件
timeout_seconds=10,
tags=["file","internal"]
)
4.7 动态加载器
# tool_loader.py
import importlib
import pkgutil
from pathlib import Path
from tool_registry import ToolRegistry
classToolLoader:
@staticmethod
defload_from_directory(registry: ToolRegistry, directory:str="builtin_tools"):
"""从指定目录动态加载所有工具模块"""
package = Path(directory)
for module_info in pkgutil.iter_modules([str(package)]):
module = importlib.import_module(f"{directory}.{module_info.name}")
# 每个模块应导出 *_tool_def 对象
for attr_name indir(module):
if attr_name.endswith("_tool_def"):
tool_def =getattr(module, attr_name)
registry.register(tool_def)
print(f"已加载工具：{tool_def.name}")
4.8 集成到 Agent 运行时
# agent.py 中的修改
from tool_executor import ToolExecutor
from sandbox import ProcessPoolSandbox
classAgentRuntime:
def__init__(self,...):
self.tools = ToolRegistry()
# 动态加载内置工具
ToolLoader.load_from_directory(self.tools,"builtin_tools")
# 注册委派等特殊工具（如前文）
if subagent_pool:
register_delegation_tools(self.tools, subagent_pool)
self.executor = ToolExecutor(self.tools, sandbox=ProcessPoolSandbox())
# ... 其他初始化
asyncdefrun(self, session_id, user_input, ctx=None):
# ... 在工具调用阶段使用 executor
if action =="tool_call":
tool_result =await self.executor.execute(
tool_name, tool_params,
SecurityContext(user_id=ctx.user_id, tenant_id=ctx.tenant_id, roles=ctx.security_roles)
)
五、运行与测试
# test_tools.py
import asyncio
from tool_registry import ToolRegistry
from tool_executor import ToolExecutor
from security_context import SecurityContext
from tool_loader import ToolLoader
asyncdefmain():
registry = ToolRegistry()
ToolLoader.load_from_directory(registry,"builtin_tools")
executor = ToolExecutor(registry)
ctx = SecurityContext(user_id="user_001", tenant_id="tenant_abc", roles=["user"])
print("=== 测试1：查询订单 ===")
result =await executor.execute("query_order",{"order_id":"O123"}, ctx)
print(result)
print("\n=== 测试2：HTTP 请求 ===")
result =await executor.execute("http_get",{"url":"https://httpbin.org/get"}, ctx)
print(result["status"],len(result["body"]))
print("\n=== 测试3：权限不足 ===")
try:
await executor.execute("read_file",{"path":"/etc/passwd"}, ctx)# user 无 admin 角色
except RuntimeError as e:
print(f"权限错误：{e}")
if __name__ =="__main__":
asyncio.run(main())
📌 生产环境增强建议：
- 使用 Firecracker MicroVM 替代 Docker，达到更高级别的安全隔离。
- 工具参数增加 输入验证（如正则、长度限制），防止注入攻击。
- 集成 OpenTelemetry，为每次工具调用生成 trace span。
- 工具执行统计面板，展示调用频率、失败率和平均延迟。
六、与 OpenClaw 的对标思考
🔍 “我们做了什么” vs “OpenClaw 为什么这样做”？
工具注册：OpenClaw 通过内置的 Tool 模块直接注册，JavaScript 的动态性让插件加载很自然。我们通过 Python 的pkgutil实现同样效果，且保留静态类型检查的优势。
沙箱隔离：OpenClaw 依赖 Node.js 的vm2或子进程，我们的沙箱抽象支持从简单进程池到 Docker/Firecracker 的无缝升级，更适合企业对安全的苛刻要求。
审计与权限：这是企业版的重点增强。我们的 ToolExecutor 在每个执行点都集成了 RBAC 检查和审计日志，而 OpenClaw 个人版较少考虑这些。
超时与熔断：企业级服务必须防止工具长时间占用资源，我们内置了超时控制，并预留了熔断器接口（失败率过高时自动禁用工具）。
本文的工具系统继承了 OpenClaw 功能丰富的思想，并通过分层设计、安全沙箱和全方位审计，使其达到企业生产标准。
七、总结与下一步
本文我们：
- 设计了以 ToolRegistry + ToolExecutor + Sandbox 为核心的企业级工具架构。
- 实现了完整的工具注册、沙箱执行、超时控制、权限校验和审计日志。
- 编写了三个内置工具示例（HTTP、订单查询、文件读取）并通过动态加载器注册。
- 将安全执行器集成到 Agent 运行时，替换了之前的直接工具调用。
下一篇文章预告：《第8篇：安全、权限与企业集成》——我们将系统化地加固整个框架，从 Gateway 到工具层的纵深防御，集成 SSO 和全链路审计，让 Agent 满足企业安全合规要求。
