Agent 框架的工具系统:定义、注册、执行工具调用。
from reson_agent import tool, ToolResult, ToolContext
@tool()
async def my_tool(arg: str, context: Optional[ToolContext] = None) -> ToolResult:
return ToolResult(
title="Success",
output="Result content"
)
| 类型 | 作用 | 定义位置 |
|---|---|---|
@tool |
装饰器,自动注册工具并生成 Schema | tools/registry.py |
ToolResult |
工具执行结果(支持记忆管理) | tools/models.py |
ToolContext |
工具执行上下文(依赖注入) | tools/models.py |
1. 定义工具
↓ @tool() 装饰器
2. 自动注册到 ToolRegistry
↓ 生成 OpenAI Tool Schema(跳过 hidden_params)
3. LLM 选择工具并生成参数
↓ registry.execute(name, args)
4. 注入框架参数(hidden_params + inject_params)
↓ 调用工具函数
5. 返回 ToolResult
↓ 转换为 LLM 消息
6. 添加到对话历史
工具参数分为三类:
query, limit)context, uid)owner, tags)@tool(
hidden_params=["context", "owner"], # 不生成 schema,LLM 看不到
inject_params={ # 声明注入规则
"owner": {"mode": "default", "key": "knowledge_config.owner"},
"tags": {"mode": "merge", "key": "knowledge_config.default_tags"},
"scopes": {"mode": "merge", "key": "knowledge_config.default_scopes"},
}
)
async def knowledge_save(
task: str, # 业务参数:LLM 填写
content: str, # 业务参数:LLM 填写
types: List[str], # 业务参数:LLM 填写
tags: Optional[Dict] = None, # 注入参数:LLM 可填,框架合并默认值
scopes: Optional[List] = None, # 注入参数:LLM 可填,框架合并默认值
owner: Optional[str] = None, # 隐藏参数:LLM 看不到,框架注入
context: Optional[ToolContext] = None, # 隐藏参数:LLM 看不到
) -> ToolResult:
"""保存知识到知识库"""
...
inject_params 声明格式:
inject_params={
"param_name": {
"mode": "default" | "merge", # 注入模式
"key": "config_obj.field", # 从 context 中取值的路径
}
}
mode: "default":LLM 未提供时注入框架值mode: "merge":框架值与 LLM 值合并。dict 按 key 合并(框架 key 不可被覆盖,LLM 可追加新 key);list 合并去重值的来源:通过 key 指定从 context 中取值的路径(如 "knowledge_config.default_tags" 表示 context["knowledge_config"].default_tags)。runner 在调用 execute() 时将配置对象放入 context,框架根据 key 路径自动取值。
注入时机:
hidden_params,不暴露给 LLMhidden_params 和 inject_params实现位置:
agent/tools/schema.py:SchemaGenerator.generate()agent/tools/registry.py:ToolRegistry.execute()from reson_agent import tool
@tool()
async def hello(name: str) -> str:
"""向用户问好"""
return f"Hello, {name}!"
要点:
@tool(hidden_params=["context", "uid"])
async def search_notes(
query: str,
limit: int = 10,
context: Optional[ToolContext] = None,
uid: str = ""
) -> str:
"""
搜索笔记
Args:
query: 搜索关键词
limit: 返回结果数量
"""
# context 和 uid 由框架注入,LLM 看不到这两个参数
...
@tool(
hidden_params=["context", "owner"],
inject_params={
"owner": {"mode": "default", "key": "knowledge_config.owner"},
"tags": {"mode": "merge", "key": "knowledge_config.default_tags"},
"scopes": {"mode": "merge", "key": "knowledge_config.default_scopes"},
}
)
async def knowledge_save(
task: str,
content: str,
types: List[str],
tags: Optional[Dict] = None, # LLM 可填,框架合并默认值
scopes: Optional[List] = None, # LLM 可填,框架合并默认值
owner: Optional[str] = None, # LLM 看不到,框架注入
context: Optional[ToolContext] = None
) -> ToolResult:
"""
保存知识
Args:
task: 任务描述
content: 知识内容
types: 知识类型
tags: 业务标签(可选,框架合并默认值)
scopes: 可见范围(可选,框架合并默认值)
"""
...
注入规则:
inject_params 的 value 是一个 dict,包含:
mode: "default"(LLM 未提供则注入)或 "merge"(与 LLM 值合并)key: 从 context 中取值的路径(如 "knowledge_config.default_tags")hidden_params 中时,LLM 不可见,框架直接注入@tool(
display={
"zh": {
"name": "搜索笔记",
"params": {
"query": "搜索关键词",
"limit": "结果数量"
}
},
"en": {
"name": "Search Notes",
"params": {
"query": "Search query",
"limit": "Result limit"
}
}
}
)
async def search_notes(query: str, limit: int = 10, uid: str = "") -> str:
"""搜索用户的笔记"""
...
from reson_agent import ToolResult
@tool()
async def read_file(path: str) -> ToolResult:
content = Path(path).read_text()
return ToolResult(
title=f"Read {path}",
output=content
)
问题:某些工具返回大量内容(如 Browser-Use 的 extract),如果每次都放入对话历史,会快速耗尽 context。
解决:ToolResult 支持双层记忆:
@tool()
async def extract_page_data(url: str) -> ToolResult:
# 假设提取了 10K tokens 的内容
full_content = extract_all_data(url)
return ToolResult(
title="Extracted page data",
output=full_content, # 完整内容(可能很长)
long_term_memory=f"Extracted {len(full_content)} chars from {url}", # 简短摘要
include_output_only_once=True # output 只给 LLM 看一次
)
效果:
output(完整内容)+ long_term_memory(摘要)long_term_memory(摘要)对话历史示例:
[User] 提取 amazon.com 的商品价格
[Assistant] 调用 extract_page_data(url="amazon.com")
[Tool]
# Extracted page data
<完整的 10K tokens 数据...>
Summary: Extracted 10000 chars from amazon.com
[User] 现在保存到文件
[Assistant] 调用 write_file(content="...")
[Tool] (此时不再包含 10K tokens,只有摘要)
Summary: Extracted 10000 chars from amazon.com
@tool()
async def risky_operation() -> ToolResult:
try:
result = perform_operation()
return ToolResult(
title="Success",
output=result
)
except Exception as e:
return ToolResult(
title="Failed",
output="",
error=str(e)
)
@tool()
async def generate_report() -> ToolResult:
report_path = create_pdf_report()
screenshot_data = take_screenshot()
return ToolResult(
title="Report generated",
output="Report created successfully",
attachments=[report_path], # 文件路径列表
images=[{
"name": "screenshot.png",
"data": screenshot_data # Base64 或路径
}]
)
工具函数可以声明需要 ToolContext 参数,框架自动注入。需要在 @tool() 装饰器中声明 hidden_params=["context"],使其对 LLM 不可见。
from reson_agent import ToolContext
@tool(hidden_params=["context"])
async def get_current_state(context: Optional[ToolContext] = None) -> ToolResult:
return ToolResult(
title="Current state",
output=f"Trace ID: {context.trace_id}\nStep ID: {context.step_id}"
)
class ToolContext(Protocol):
# 基础字段(所有工具)
trace_id: str # 当前 Trace ID
step_id: str # 当前 Step ID
uid: Optional[str] # 用户 ID
# 扩展字段(由 runner 注入)
store: Optional[TraceStore] # Trace 存储
runner: Optional[AgentRunner] # Runner 实例
goal_tree: Optional[GoalTree] # 目标树
goal_id: Optional[str] # 当前 Goal ID
config: Optional[RunConfig] # 运行配置
# 浏览器相关(Browser-Use 集成)
browser_session: Optional[Any] # 浏览器会话
page_url: Optional[str] # 当前页面 URL
file_system: Optional[Any] # 文件系统访问
sensitive_data: Optional[Dict] # 敏感数据
# 额外上下文
context: Optional[Dict[str, Any]] # 额外上下文数据
@tool(hidden_params=["context"])
async def analyze_current_page(context: Optional[ToolContext] = None) -> ToolResult:
"""分析当前浏览器页面"""
if not context or not context.browser_session:
return ToolResult(
title="Error",
error="Browser session not available"
)
# 使用浏览器会话
page_content = await context.browser_session.get_content()
return ToolResult(
title=f"Analyzed {context.page_url}",
output=page_content,
long_term_memory=f"Analyzed page at {context.page_url}"
)
Runner 在执行工具时自动创建并注入 context:
# 在 AgentRunner._agent_loop 中
context = {
"store": self.trace_store,
"trace_id": trace_id,
"goal_id": current_goal_id,
"runner": self,
"goal_tree": goal_tree,
"config": config,
}
result = await self.tools.execute(
tool_name,
tool_args,
uid=config.uid or "",
context=context
)
@tool(requires_confirmation=True)
async def delete_all_notes(uid: str = "") -> ToolResult:
"""删除所有笔记(危险操作)"""
# 执行前会等待用户确认
...
适用场景:
@tool(editable_params=["query", "filters"])
async def advanced_search(
query: str,
filters: Optional[Dict] = None,
uid: str = ""
) -> ToolResult:
"""高级搜索"""
# LLM 生成参数后,用户可以编辑 query 和 filters
...
适用场景:
场景:某些工具只在特定网站可用,减少无关工具的 context 占用。
@tool(url_patterns=["*.google.com", "www.google.*"])
async def google_advanced_search(
query: str,
date_range: Optional[str] = None,
uid: str = ""
) -> ToolResult:
"""Google 高级搜索技巧(仅在 Google 页面可用)"""
...
@tool(url_patterns=["*.github.com"])
async def github_pr_create(
title: str,
body: str,
uid: str = ""
) -> ToolResult:
"""创建 GitHub PR(仅在 GitHub 页面可用)"""
...
@tool() # 无 url_patterns,所有页面都可用
async def take_screenshot() -> ToolResult:
"""截图(所有页面都可用)"""
...
支持的模式:
# 通配符域名
"*.google.com" # 匹配 www.google.com, mail.google.com
"www.google.*" # 匹配 www.google.com, www.google.co.uk
# 路径匹配
"https://github.com/**/issues" # 匹配所有 issues 页面
# 多个模式
url_patterns=["*.github.com", "*.gitlab.com"]
使用过滤后的工具:
from reson_agent import get_tool_registry
registry = get_tool_registry()
# 根据 URL 获取可用工具
current_url = "https://www.google.com/search?q=test"
tool_names = registry.get_tool_names(current_url)
# 返回:["google_advanced_search", "take_screenshot"](不包含 github_pr_create)
# 获取过滤后的 Schema
schemas = registry.get_schemas_for_url(current_url)
# 传递给 LLM,只包含相关工具
效果:
| 场景 | 无过滤 | 有过滤 | 节省 |
|---|---|---|---|
| 在 Google 页面 | 35 工具 (~5K tokens) | 20 工具 (~3K tokens) | 40% |
| 在 GitHub 页面 | 35 工具 (~5K tokens) | 18 工具 (~2.5K tokens) | 50% |
场景:浏览器自动化需要输入密码、Token,但不想在对话历史中显示明文。
设置敏感数据:
sensitive_data = {
# 格式 1:全局密钥(适用于所有域名)
"api_key": "sk-xxxxx",
# 格式 2:域名特定密钥(推荐)
"*.github.com": {
"github_token": "ghp_xxxxx",
"github_password": "my_secret_password"
},
"*.google.com": {
"google_email": "user@example.com",
"google_password": "another_secret",
"google_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP" # TOTP secret
}
}
LLM 输出占位符:
# LLM 决定需要输入密码
{
"tool": "browser_input",
"arguments": {
"index": 5,
"text": "<secret>github_password</secret>" # 占位符
}
}
自动替换:
# 执行工具前,框架自动替换
registry.execute(
"browser_input",
arguments={"index": 5, "text": "<secret>github_password</secret>"},
context={"page_url": "https://github.com/login"},
sensitive_data=sensitive_data
)
# 实际执行:
# arguments = {"index": 5, "text": "my_secret_password"}
TOTP 2FA 支持:
# 密钥以 _bu_2fa_code 结尾,自动生成 TOTP 代码
sensitive_data = {
"*.google.com": {
"google_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP"
}
}
# LLM 输出
{
"text": "<secret>google_2fa_bu_2fa_code</secret>"
}
# 自动替换为当前的 6 位数字验证码
{
"text": "123456" # 当前时间的 TOTP 代码
}
完整示例:
from reson_agent import get_tool_registry, ToolContext
# 设置敏感数据
sensitive_data = {
"*.github.com": {
"github_token": "ghp_xxxxxxxxxxxxx",
"github_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP"
}
}
# 执行工具(LLM 输出的参数包含占位符)
result = await registry.execute(
"github_api_call",
arguments={
"endpoint": "/user",
"token": "<secret>github_token</secret>",
"totp": "<secret>github_2fa_bu_2fa_code</secret>"
},
context={"page_url": "https://github.com"},
sensitive_data=sensitive_data
)
# 实际调用时参数已被替换:
# {
# "endpoint": "/user",
# "token": "ghp_xxxxxxxxxxxxx",
# "totp": "123456"
# }
安全性:
<secret>key</secret> 占位符自动记录:
每个工具调用自动记录:
查询统计:
from reson_agent import get_tool_registry
registry = get_tool_registry()
# 获取所有工具统计
all_stats = registry.get_stats()
print(all_stats)
# {
# "search_notes": {
# "call_count": 145,
# "success_count": 142,
# "failure_count": 3,
# "average_duration": 0.32,
# "success_rate": 0.979,
# "last_called": 1704123456.78
# },
# ...
# }
# 获取单个工具统计
search_stats = registry.get_stats("search_notes")
# 获取 Top 工具
top_tools = registry.get_top_tools(limit=5, by="call_count")
# ['search_notes', 'read_file', 'browser_click', ...]
top_by_success = registry.get_top_tools(limit=5, by="success_rate")
fastest_tools = registry.get_top_tools(limit=5, by="average_duration")
优化工具排序:
# 根据使用频率优化工具顺序
def get_optimized_schemas(registry, current_url):
# 获取可用工具
tool_names = registry.get_tool_names(current_url)
# 按调用次数排序(高频工具排前面)
all_stats = registry.get_stats()
sorted_tools = sorted(
tool_names,
key=lambda name: all_stats.get(name, {}).get("call_count", 0),
reverse=True
)
# 返回排序后的 Schema
return registry.get_schemas(sorted_tools)
监控和告警:
# 监控工具失败率
stats = registry.get_stats()
for tool_name, tool_stats in stats.items():
if tool_stats["call_count"] > 10 and tool_stats["success_rate"] < 0.8:
logger.warning(
f"Tool {tool_name} has low success rate: "
f"{tool_stats['success_rate']:.1%} "
f"({tool_stats['failure_count']}/{tool_stats['call_count']} failures)"
)
# 监控执行时间
for tool_name, tool_stats in stats.items():
if tool_stats["average_duration"] > 5.0:
logger.warning(
f"Tool {tool_name} is slow: "
f"average {tool_stats['average_duration']:.2f}s"
)
完整示例:浏览器自动化工具
@tool(
requires_confirmation=False,
editable_params=["query"],
url_patterns=["*.google.com"],
display={
"zh": {"name": "Google 搜索", "params": {"query": "搜索关键词"}},
"en": {"name": "Google Search", "params": {"query": "Query"}}
}
)
async def google_search(
query: str,
ctx: ToolContext,
uid: str = ""
) -> ToolResult:
"""
在 Google 执行搜索
仅在 Google 页面可用,支持敏感数据注入
"""
# 使用浏览器会话
if not ctx.browser_session:
return ToolResult(title="Error", error="Browser session not available")
# 敏感数据已在 registry.execute() 中自动处理
# 例如 query 中的 <secret>api_key</secret> 已被替换
# 执行搜索
await ctx.browser_session.navigate(f"https://google.com/search?q={query}")
# 提取结果
results = await ctx.browser_session.extract_results()
return ToolResult(
title=f"Search results for {query}",
output=json.dumps(results),
long_term_memory=f"Searched Google for '{query}', found {len(results)} results",
include_output_only_once=True
)
使用:
# 设置环境
registry = get_tool_registry()
sensitive_data = {"*.google.com": {"search_api_key": "sk-xxxxx"}}
# Agent 在 Google 页面时
current_url = "https://www.google.com"
# 获取工具(自动过滤)
tool_names = registry.get_tool_names(current_url)
# 包含 google_search(匹配 *.google.com)
# LLM 决定使用工具(可能包含敏感占位符)
tool_call = {
"name": "google_search",
"arguments": {
"query": "site:github.com <secret>search_api_key</secret>"
}
}
# 执行(自动替换敏感数据)
result = await registry.execute(
tool_call["name"],
tool_call["arguments"],
context={"page_url": current_url, "browser_session": browser},
sensitive_data=sensitive_data
)
# 查看统计
stats = registry.get_stats("google_search")
print(f"Success rate: {stats['success_rate']:.1%}")
参考 opencode 实现的文件操作和命令执行工具
框架提供一组内置的基础工具,用于文件读取、编辑、搜索和命令执行等常见任务。这些工具参考了 opencode 的成熟设计,在 Python 中重新实现。
实现位置:
agent/tools/builtin/agent/tools/adapters/vendor/opencode/ (git submodule)详细文档:参考 docs/tools-adapters.md
| 工具 | 功能 | 参考 |
|---|---|---|
read_file |
读取单个文件(文本 / 图片 / PDF) | opencode read.ts |
read_images |
批量读取图片,支持自动降采样和网格拼图 | 自研 |
edit_file |
智能文件编辑(多种匹配策略) | opencode edit.ts |
write_file |
写入文件(创建或覆盖) | opencode write.ts |
bash_command |
执行 shell 命令 | opencode bash.ts |
glob_files |
文件模式匹配 | opencode glob.ts |
grep_content |
内容搜索(正则表达式) | opencode grep.ts |
agent |
创建 Agent 执行任务(单任务 delegate / 多任务并行 explore) | 自研 |
evaluate |
评估目标执行结果是否满足要求 | 自研 |
toolhub_health |
检查 ToolHub 远程工具库服务状态 | 自研 |
toolhub_search |
搜索/发现 ToolHub 远程工具 | 自研 |
toolhub_call |
调用 ToolHub 远程工具(图片参数支持本地文件路径) | 自研 |
ask_knowledge |
向知识库查询信息(通过 KnowHub Librarian) | 自研 |
upload_knowledge |
上传调研结果到知识库 | 自研 |
read_file vs read_images| 场景 | 工具 |
|---|---|
| 读取 1 张图片 / 文本 / PDF | read_file |
| 批量读取 2 张以上图片 | read_images |
| 需要 AI 对多张图做对比选择(选图、挑错、横向比较) | read_images 且 layout="grid" |
| 需要对多张图逐张独立分析 | read_images 且 layout="separate" |
read_images 默认 layout="grid" — 多张图拼成一张带索引编号的网格图(1,2,3…),省 token 的同时让 LLM 能在单次注视中做横向对比。拼图和降采样在内部组合使用:先降采样每张缩略图,再拼成整图,最终图片大小约等于一张普通图的开销,而非所有原图的累积。
Grid 模式的 16 张硬上限: grid 模式下单次调用最多 16 张图片。超过会报错,需要分批调用。上限来自于 LLM 内部图片缩放的物理限制——Claude/Qwen-VL 会把图片缩到长边约 1568 像素,当拼图里格子太多时,每格会糊到无法识别。16 张对应 4×4 布局,每格约 300px,缩放后仍能保持约 280px,人物和场景细节仍然可辨。如需处理更多图片,或切换到 layout="separate"(无数量限制但每张图都有独立的结构开销 token)。
自适应布局: grid 模式下根据图片数量动态选择列数和缩略图尺寸,小批量时每张图更清晰:
| 图片数 | 布局 | 每格大小 |
|---|---|---|
| 2 张 | 2 列 | 500px |
| 3-4 张 | 2 列 | 450px |
| 5-6 张 | 3 列 | 400px |
| 7-9 张 | 3 列 | 380px |
| 10-12 张 | 4 列 | 320px |
| 13-16 张 | 4 列 | 300px |
关于标签/标题: read_images 的拼图不显示文件名,只显示索引序号——因为本地文件名(如 IMG_1234.jpg)对 LLM 理解内容没有帮助,而索引到原始路径的对照表通过返回文本提供,LLM 可以用"第 3 张"这种引用方式精确指代。对比之下 search_posts / youtube_search 的拼图会显示 label(帖子/视频标题),因为这些是内容型元数据,有实际信息量。这一差异反映在 build_image_grid(labels=...) 参数上:传 None 只画序号,传列表则在每格下方画标题。
网格和降采样的实现在 agent/tools/utils/image.py,search_posts 和 youtube_search 等工具也复用同一套拼图逻辑。
创建子 Agent 执行任务。通过 task 参数的类型自动区分模式:
| task 类型 | 模式 | 并行执行 | 工具权限 |
|---|---|---|---|
str(单任务) |
delegate | ❌ | 完整(除 agent/evaluate 外) |
List[str](多任务) |
explore | ✅ | 只读(read_file, grep_content, glob_files, goal) |
@tool(description="创建 Agent 执行任务")
async def agent(
task: Union[str, List[str]],
messages: Optional[Union[Messages, List[Messages]]] = None,
continue_from: Optional[str] = None,
agent_type: Optional[str] = None,
skills: Optional[List[str]] = None,
context: Optional[dict] = None,
) -> Dict[str, Any]:
参数说明:
task: 任务描述(字符串=单任务,列表=多任务并行)messages: 预置消息(None/1D 列表/2D 列表)continue_from: 继续已有 trace(仅单任务)agent_type: 子 Agent 类型,决定 preset 和默认 skills(如 "tool_research")skills: 附加到 system prompt 的 skill 名称列表,覆盖 preset 默认值context: 框架自动注入的上下文messages 参数:
None:无预置消息Messages(1D 列表):所有 agent 共享List[Messages](2D 列表):per-agent 独立消息运行时判断:messages[0] 是 dict → 1D 共享;是 list → 2D per-agent。
agent_type 与 Presets:
agent_type 参数指定预定义的 Agent 配置(工具权限、system prompt、skills 等)presets.json 中定义自定义 preset,支持从 .prompt 文件加载 system promptagent/docs/architecture.md 的 "Agent 预设" 章节和 examples/production/ 示例单任务(delegate):
continue_from 参数续跑已有 Sub-Trace多任务(explore):
asyncio.gather() 并行执行,显著提升效率continue_from评估指定 Goal 的执行结果,提供质量评估和改进建议。
@tool(description="评估目标执行结果是否满足要求")
async def evaluate(
messages: Optional[Messages] = None,
target_goal_id: Optional[str] = None,
continue_from: Optional[str] = None,
context: Optional[dict] = None,
) -> Dict[str, Any]:
criteria 参数——代码自动从 GoalTree 注入目标描述messages 中target_goal_id 默认为当前 goal_idSub-Trace 结构:
agent/evaluate 调用创建独立的 Sub-Trace{parent_id}@{mode}-{序号}-{timestamp}-001parent_trace_id 和 parent_goal_id 建立父子关系Goal 集成:
agent/evaluate 调用会将 Goal 标记为 type: "agent_call"agent_call_mode 记录使用的模式sub_trace_ids 记录所有创建的 Sub-Tracesummary 包含格式化的汇总结果实现位置:agent/tools/builtin/subagent.py
from agent.tools.builtin import read_file, edit_file, bash_command
# 读取文件
result = await read_file(file_path="config.py", limit=100)
print(result.output)
# 编辑文件(智能匹配)
result = await edit_file(
file_path="config.py",
old_string="DEBUG = True",
new_string="DEBUG = False"
)
# 执行命令
result = await bash_command(
command="git status",
timeout=30,
description="Check git status"
)
Read Tool 特性:
Edit Tool 特性:
Bash Tool 特性:
内置工具参考 vendor/opencode/ 中的实现,通过 git submodule 管理:
# 更新 opencode 参考
cd vendor/opencode
git pull origin main
cd ../..
git add vendor/opencode
git commit -m "chore: update opencode reference"
# 查看最近变更
cd vendor/opencode
git log --oneline --since="1 month ago" -- packages/opencode/src/tool/
更新后,检查是否需要同步改进到 Python 实现。
将 Browser-Use 的 25 个工具适配为你的工具系统:
from browser_use import BrowserSession, Tools as BrowserUseTools
from reson_agent import tool, ToolResult, ToolContext
class BrowserToolsAdapter:
"""Browser-Use 工具适配器"""
def __init__(self):
self.session = BrowserSession(headless=False)
self.browser_tools = BrowserUseTools()
async def __aenter__(self):
await self.session.__aenter__()
return self
async def __aexit__(self, *args):
await self.session.__aexit__(*args)
def register_all(self, registry):
"""批量注册所有 Browser-Use 工具"""
for action_name, registered_action in self.browser_tools.registry.actions.items():
self._adapt_action(registry, action_name, registered_action)
def _adapt_action(self, registry, action_name, registered_action):
"""适配单个 Browser-Use action"""
@tool()
async def adapted_tool(args: dict, ctx: ToolContext) -> ToolResult:
# 构建 Browser-Use 需要的 special context
special_context = {
'browser_session': self.session,
'page_url': ctx.page_url,
'file_system': ctx.file_system,
}
# 执行 Browser-Use action
result = await registered_action.function(
params=registered_action.param_model(**args),
**special_context
)
# 转换 ActionResult -> ToolResult
return ToolResult(
title=action_name,
output=result.extracted_content or '',
long_term_memory=result.long_term_memory,
include_output_only_once=result.include_extracted_content_only_once,
error=result.error,
attachments=result.attachments or [],
images=result.images or [],
metadata=result.metadata or {}
)
# 注册到你的 registry
registry.register(adapted_tool, schema=generate_schema(registered_action))
from reson_agent import AgentRunner
async def main():
async with BrowserToolsAdapter() as browser:
# 创建 Agent
agent = AgentRunner(
task="在 Amazon 找最便宜的 iPhone 15",
tools=[], # 空列表
)
# 批量注册浏览器工具
browser.register_all(agent.tool_registry)
# 现在 Agent 有 25 个浏览器工具 + 其他工具
result = await agent.run()
| 工具类型 | 数量 | Token 占用 | 占比(200K) |
|---|---|---|---|
| Browser-Use 工具 | 25 | ~4,000 | 2% |
| 你的自定义工具 | 10 | ~1,000 | 0.5% |
| 总计 | 35 | ~5,000 | 2.5% |
结论:完全可接受,且 Prompt Caching 会优化后续调用。
# 好:清晰的动词 + 名词
@tool()
async def search_notes(...): ...
@tool()
async def create_document(...): ...
# 不好:模糊或过长
@tool()
async def do_something(...): ...
@tool()
async def search_and_filter_notes_with_advanced_options(...): ...
# 好:返回 ToolResult 或结构化字典
@tool()
async def get_weather(city: str) -> ToolResult:
data = fetch_weather(city)
return ToolResult(
title=f"Weather in {city}",
output=json.dumps(data, indent=2)
)
# 不好:返回纯文本
@tool()
async def get_weather(city: str) -> str:
return "The weather is sunny, 25°C, humidity 60%" # 难以解析
# 好:捕获异常并返回 ToolResult
@tool()
async def risky_operation() -> ToolResult:
try:
result = dangerous_call()
return ToolResult(title="Success", output=result)
except Exception as e:
logger.error(f"Operation failed: {e}")
return ToolResult(title="Failed", error=str(e))
# 不好:让异常传播(会中断 Agent 循环)
@tool()
async def risky_operation() -> str:
return dangerous_call() # 可能抛出异常
# 好:大量数据用 include_output_only_once
@tool()
async def fetch_all_logs() -> ToolResult:
logs = get_last_10000_logs() # 很大
return ToolResult(
title="Fetched logs",
output=logs,
long_term_memory=f"Fetched {len(logs)} log entries",
include_output_only_once=True # 只给 LLM 看一次
)
# 不好:大量数据每次都传给 LLM
@tool()
async def fetch_all_logs() -> str:
return get_last_10000_logs() # 每次都占用 context
# 好:单一职责,细粒度
@tool()
async def search_notes(query: str) -> ToolResult: ...
@tool()
async def get_note_detail(note_id: str) -> ToolResult: ...
@tool()
async def update_note(note_id: str, content: str) -> ToolResult: ...
# 不好:功能过多,难以使用
@tool()
async def manage_notes(
action: Literal["search", "get", "update", "delete"],
query: Optional[str] = None,
note_id: Optional[str] = None,
content: Optional[str] = None
) -> ToolResult:
# 太复杂,LLM 容易用错
...
@tool()
async def search_notes(
query: str,
filters: Optional[Dict[str, Any]] = None,
sort_by: str = "relevance",
limit: int = 10,
uid: str = ""
) -> ToolResult:
"""
搜索用户的笔记
使用语义搜索查找相关笔记,支持过滤和排序。
Args:
query: 搜索关键词(必需)
filters: 过滤条件,例如 {"type": "markdown", "tags": ["work"]}
sort_by: 排序方式,可选 "relevance" | "date" | "title"
limit: 返回结果数量,默认 10,最大 100
Returns:
ToolResult 包含搜索结果列表
Example:
搜索包含 "项目计划" 的工作笔记:
{
"query": "项目计划",
"filters": {"tags": ["work"]},
"limit": 5
}
"""
...
工具设计为可跨 Agent 框架使用(本框架 Agent、Claude Code、其他 LLM IDE 等),遵循以下原则:
| 问题 | 答案 | 选择 |
|---|---|---|
| 工具调用之间是否有进程内状态需要保持?(浏览器 session、数据库连接、缓存) | 否 | CLI |
| 同上 | 是 | MCP |
| 是否需要 Claude Desktop、Cursor 等客户端原生识别? | 需要 | MCP |
一个工具想同时作为 Agent tool(@tool 注册)和 CLI 工具使用,需要满足以下要求:
1. 文件末尾添加自包含的 if __name__ == "__main__" 块
参数解析、asyncio.run、结果输出这些 CLI 样板代码直接内联在工具文件里,不要抽取到共享 cli.py 模块——这样每个工具文件可以独立迁移到其他项目。
# 示例:agent/tools/builtin/toolhub.py 末尾
if __name__ == "__main__":
import sys, asyncio, os, uuid
COMMANDS = {"health": toolhub_health, "search": toolhub_search, "call": toolhub_call}
def _parse_args(argv):
kwargs = {}
for arg in argv:
if arg.startswith("--") and "=" in arg:
k, v = arg.split("=", 1)
k = k.lstrip("-").replace("-", "_")
try: v = json.loads(v)
except: pass
kwargs[k] = v
return kwargs
# trace_id 三级回退:CLI 参数 > 环境变量 > 自动生成
cmd = sys.argv[1]
kwargs = _parse_args(sys.argv[2:])
trace_id = kwargs.pop("trace_id", None) or os.getenv("TRACE_ID") or f"cli-{uuid.uuid4().hex[:8]}"
set_trace_context(trace_id)
result = asyncio.run(COMMANDS[cmd](**kwargs))
# 输出 JSON(注意 double-encoding 问题)
...
2. 输出统一为 JSON 格式
{
"trace_id": "...",
"output": "...", // 原生 dict/list/str,不要预先 json.dumps
"error": "...", // 可选
"metadata": {...} // 可选
}
3. trace_id 三级回退策略
对于需要会话语义(同一 trace_id 内多次调用共享状态)的工具(librarian、toolhub 的图片输出目录等):
--trace_id=xxx(显式)TRACE_ID(同一 shell session 共享)cli-{random}(兜底)外部 Agent 只需 export TRACE_ID=session-xxx 一次,后续所有 CLI 调用自动归到同一会话。
4. 二进制产出写文件,JSON 返回路径
像 read_images 这种产出图片/大文件的工具,CLI 模式下不要把 base64 塞进 stdout(刷屏 + 调用方还要解码)。应该:
--out=<path> 指定输出路径<path>out_path 供调用方用 Read 工具查看5. 避免双重 JSON 编码
如果工具内部已经 json.dumps() 把 result dict 塞进了 ToolResult.output,CLI 层再 json.dumps(result.output) 会产生双重转义("output": "{\"model\": ..." 这种反人类形式)。CLI 层要在输出前检测并解码:
output_value = result.output
if isinstance(output_value, str):
stripped = output_value.lstrip()
if stripped.startswith(("{", "[")):
try:
output_value = json.loads(output_value)
except (json.JSONDecodeError, ValueError):
pass # 非 JSON 文本,保持原样
CLI 工具对外暴露给 Claude Code(或其他支持 skill 的客户端)时,需要配套写一个 SKILL.md:
位置: ~/.claude/skills/<name>/SKILL.md(用户全局)或项目级 .claude/skills/<name>/SKILL.md
格式:
---
name: <skill-name>
description: <一句话,描述用途和触发时机。这是 Claude Code 决定何时加载该 skill 的唯一依据>
---
# <Skill Name>
<简短一段话介绍工具>
## 用法
```bash
python <绝对路径>/tool.py <子命令> --key=value
--key=... 参数说明<调用后怎么解读输出,典型 workflow>
**尺寸原则:** SKILL.md **越短越好**。它每次触发时都会进入 context 占据 token。和 `agent/docs/tools.md` 的职责区分:
| 文件 | 读者 | 触发 | 长度 |
|------|------|------|------|
| `SKILL.md` | **运行时的 Claude Code**(动态加载) | 每次匹配自动加载到 context | **短**(20 行以内为佳) |
| `agent/docs/tools.md` | **开发者**(静态阅读) | 从不自动加载 | 长,可以详细展开原理、设计取舍 |
SKILL.md 只写"调用这个工具所需的最小信息集",原理和细节放到 docs。
**当前已安装的 skill**(`~/.claude/skills/`):
- `toolhub/` — 搜索和调用 ToolHub 远程 AI 工具
- `knowhub/` — 查询和上传 KnowHub 知识库
- `stitch-images/` — 批量图片拼成网格供 Read 一次查看
### ToolHub 图片管线
`toolhub_call` 内置完整的图片处理管线,无需单独的上传/下载工具:
- **输入**:`params` 中的图片参数(`image`、`image_url`、`mask_image`、`pose_image`、`images`)可直接传本地文件路径,系统自动上传
- **输出**:生成的图片自动保存到 `outputs/` 目录,返回结果中 `saved_files` 包含本地路径
### MCP 集成(有状态工具组)
对于需要维持 session 的工具组,使用 MCP server。两种注册方式:
**1. 使用现成的 MCP server**(推荐)
例如浏览器工具直接用 browser-use 原生 MCP:
```json
// .mcp.json(项目根目录;不要写在 settings.json,Claude Code 不会从那里读 mcpServers)
{
"mcpServers": {
"browser-use": {
"command": "/Users/sunlit/.pyenv/versions/3.13.1/bin/python",
"args": ["-m", "browser_use.skill_cli.main", "--mcp"],
"env": {
"OPENAI_API_KEY": "sk-...",
"OPENAI_BASE_URL": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"BROWSER_USE_LLM_MODEL": "qwen-plus"
}
}
}
}
注意事项:
command 必须用绝对路径,不要用 pyenv shim(shim 在 Claude Code 子进程里无法正确解析).mcp.json,不是 ~/.claude/settings.json(后者只管 permissions/outputStyle 等)LLMEntry 不支持 base_url),可以通过环境变量绕过(如 OPENAI_BASE_URL 是 OpenAI SDK 原生环境变量)2. 为自研有状态工具组写 MCP server
当你有一组需要共享 session 的自研工具时,用 mcp Python SDK 写一个 server,每个工具作为 @app.tool() 暴露。server 进程内维护 session 状态。避免造私有 stdio 协议。
| 特性 | 状态 | 说明 |
|---|---|---|
| 基础注册 | ✅ 已实现 | @tool() 装饰器 |
| Schema 生成 | ✅ 已实现 | 自动从函数签名生成 |
| 双层记忆 | ✅ 已实现 | ToolResult 支持 long_term_memory |
| 依赖注入 | ✅ 已实现 | ToolContext 提供上下文 |
| UI 元数据 | ✅ 已实现 | display, requires_confirmation, editable_params |
| 域名过滤 | ✅ 已实现 | url_patterns 参数 + URL 匹配器 |
| 敏感数据 | ✅ 已实现 | <secret> 占位符 + TOTP 支持 |
| 工具统计 | ✅ 已实现 | 自动记录调用次数、成功率、执行时间 |
核心设计原则: