Builtin tools cn
Jianmu 提供了一组开箱即用的内建工具,覆盖计算、代码执行、文件操作、HTTP 请求和 Web 搜索等常见 Agent 场景。所有内建工具都建立在统一的 Tool 抽象之上,通过 BuiltinToolProvider 按名解析,并可与本地或 Docker 执行沙箱配合使用。本文重点说明这套工具系统的结构、约束和典型工具的行为边界。
工具系统架构概览¶
在深入逐个工具之前,先建立对内建工具系统整体架构的认知。下面的 Mermaid 图展示了工具从定义到执行的核心通路:
flowchart TB
subgraph 定义层
A[Tool 抽象基类] --> B[内建工具类]
C["@tool 装饰器"] --> D[FunctionTool]
end
subgraph 注册层
E[BuiltinToolProvider] --> F["_ensure_builtins()"]
F --> G["register_tool(name, cls)"]
G --> H["_builtins 字典"]
I[别名系统] --> H
end
subgraph 装配层
J[ToolSet.resolve] --> E
J --> K["merge_groups()"]
K --> L[去重后的 Tool 列表]
end
subgraph 执行层
M["Tool.execute(args, runner=?)"]
M --> N{runner 存在?}
N -->|是| O["Sandbox.run_tool(tool, args)"]
N -->|否| P["_execute_local(call_args)"]
O --> Q[LocalSandbox / DockerSandbox]
end
B --> H
D --> L
L --> M
参考实现:base.py、provider.py、set.py
核心要点:工具定义是静态的,注册是懒加载的,装配支持按名引用和别名映射,执行则根据配置选择本地或容器化沙箱。先理解这个分层结构,再看各个具体工具会清晰很多。
Tool 基类与执行契约¶
所有内建工具都直接或间接继承 Tool 基类。该基类定义了工具必须实现的核心接口和运行时执行契约。
class Tool(ABC):
name: str = "unnamed_tool"
description: str = "No description provided"
input_schema: Dict[str, Any] = {"type": "string", "description": "Tool input string"}
output_schema: Dict[str, Any] = {"type": "string", "description": "Tool output string"}
parallel_safe: bool = True
effect_tags: tuple[str, ...] = ()
@abstractmethod
def run(self, input: Any) -> Any: ...
async def execute(self, args, *, injected=None, prefer_injected=True,
runner=None, runner_event_context=None) -> Any: ...
参考实现:base.py
execute() 方法是工具的统一入口。它的核心职责有二:参数合并和执行委托。当 runner 参数不为 None 时,工具执行被委托给外部沙箱(如 DockerSandbox);否则直接在进程中执行 run()。参数合并逻辑 _merge_call_args 支持 injected 运行时注入,通过 prefer_injected 控制优先级——当为 True 时注入参数覆盖显式参数。
参考实现:base.py
工具还具备 as_node() 方法,可将自身包装为行为树节点 ToolNode,使其既能被 LLM Agent 循环调用,也能嵌入显式工作流分支中。
参考实现:base.py
| 属性 | 类型 | 用途 |
|---|---|---|
name |
str |
工具的唯一标识,LLM function calling 和 ToolSet 查找均依赖此名 |
description |
str |
工具描述,注入 LLM prompt 帮助模型判断何时调用 |
input_schema |
dict |
JSON Schema 格式的输入规范,转换为 OpenAI function calling parameters |
output_schema |
dict |
JSON Schema 格式的输出规范 |
parallel_safe |
bool |
标记工具是否可并行调用,Guard 系统据此决策 |
effect_tags |
tuple |
副作用标签,如 ("filesystem:write",),用于约束检查 |
@tool 装饰器:函数即工具¶
对于简单场景,继承 Tool 类可能过于繁琐。@tool 装饰器允许将任意 Python 函数直接转换为 FunctionTool 实例:
@tool(name="my_tool", description="does something")
def my_tool(x: int, y: int) -> int:
return x + y
参考实现:decorator.py
FunctionTool 在初始化时自动检测函数的同步/异步特性——若是协程函数,则 run 方法直接 await;若是同步函数,则保留原始调用。工具名和描述优先使用装饰器参数,其次回退到函数名和 docstring。
参考实现:decorator.py
计算器 (CalculatorTool)¶
CalculatorTool 是最简洁的内建工具,用于执行基础算术运算。
class CalculatorTool(Tool):
name = "calculator"
description = "Performs basic math calculations..."
input_schema = {"type": "string", "description": "Python math expression, e.g. '2+2'"}
def run(self, input: str) -> str:
allowed_chars = set("0123456789+-*/().% ")
if not all(c in allowed_chars for c in input):
return f"Error: Only math expressions allowed, got: {input}"
result = eval(input)
return str(result)
参考实现:calculator.py
这个工具的关键在于输入约束:它通过字符白名单 0123456789+-*/().% 严格限制输入,只允许数字和基本运算符。任何包含字母或其他特殊字符的输入都会被拒绝。这里的 eval() 运行在受控输入范围内,本质上是表达式求值器,而不是通用代码执行器。如果需要更复杂的 Python 计算,应使用 PythonREPLTool。
Python REPL (PythonREPLTool)¶
PythonREPLTool 提供完整的 Python 代码执行能力,是内建工具中功能最强大同时风险最高的工具。
class PythonREPLTool(Tool):
name = "python_repl"
input_schema = {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "The Python code to execute"
}
},
"required": ["code"]
}
参考实现:python_repl.py
执行策略:eval → exec 双阶段¶
_execute_code() 采用先 eval 后 exec 的策略。对于表达式(如 2+2),eval() 直接返回结果值;若遇到 SyntaxError(即输入是语句而非表达式),则回退到 exec() 执行。两个阶段的 stdout 和 stderr 通过 contextlib.redirect_stdout/stderr 统一捕获:
def _execute_code(self, code: str) -> str:
with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
try:
eval_result = eval(code, self._globals, self._locals)
if eval_result is not None:
result = str(eval_result)
else:
result = stdout_capture.getvalue()
except SyntaxError:
exec(code, self._globals, self._locals)
result = stdout_capture.getvalue()
# stderr 合并...
参考实现:python_repl.py
安全与资源控制¶
| 控制维度 | 默认值 | 说明 |
|---|---|---|
timeout |
30.0 秒 | 单次执行的最大时间限制,超时返回错误信息 |
max_output_length |
10000 字符 | 输出截断阈值,超长结果末尾附加截断提示 |
allowed_modules |
None(全部允许) |
可限制为白名单列表 |
globals_dict |
{"__builtins__": __builtins__} |
自定义全局命名空间,可移除危险内置函数 |
参考实现:python_repl.py
run() 方法是异步的,内部通过 loop.run_in_executor() 将同步的 _execute_code() 放入线程池执行,并用 asyncio.wait_for() 实施超时控制。这确保了 REPL 执行不会阻塞事件循环。
参考实现:python_repl.py
生产环境警告:该工具执行任意 Python 代码。强烈建议配合 DockerSandbox 使用以实现真正的进程隔离。在 jianmu.yaml 中设置 execution.tool_runner: docker 即可启用。
文件读写工具族¶
文件操作工具包含三个独立类:FileReadTool、FileWriteTool 和 ListDirTool,共享统一的路径验证逻辑。
路径安全验证¶
三个工具的核心安全机制是 _validate_path() 方法,实现三层防护:
def _validate_path(self, path: str) -> tuple[Path, Optional[str]]:
file_path = Path(path).resolve()
# 第一层:base_path 目录限制
if self.base_path:
file_path.relative_to(base_path) # ValueError → 拒绝访问
# 第二层:扩展名白名单
if self.allowed_extensions:
if file_path.suffix.lower() not in self.allowed_extensions:
return file_path, f"Extension not allowed: {ext}"
# 第三层:存在性与类型检查
if not file_path.exists(): ...
if not file_path.is_file(): ...
# 第四层:文件大小限制(仅 FileReadTool)
if size > self.max_file_size: ...
参考实现:file.py
FileReadTool¶
读取文件全部文本内容。默认限制 1MB,可通过 max_file_size 调整。文件读取通过 loop.run_in_executor() 在线程池中异步执行,避免阻塞。遇到 UnicodeDecodeError 时返回明确的编码错误信息。
参考实现:file.py
FileWriteTool¶
写入或追加文本内容到文件。支持 create_dirs=True(默认)自动创建父目录。append 参数控制模式——False 覆盖写入,True 追加。写入后返回 "Wrote {path} ({size} bytes)" 格式的状态信息。
参考实现:file.py
ListDirTool¶
列出目录内容,输出格式化的表格文本。每个条目标注类型(FILE 或 DIR)、大小(自动选择 B/KB/MB 单位)和名称,所有条目按名称排序。
参考实现:file.py
| 参数 | FileReadTool | FileWriteTool | ListDirTool |
|---|---|---|---|
base_path |
✅ 目录沙箱 | ✅ 目录沙箱 | ✅ 目录沙箱 |
allowed_extensions |
✅ 扩展名白名单 | ✅ 扩展名白名单 | ❌ |
max_file_size |
✅ 1MB 默认 | ❌ | ❌ |
max_content_size |
❌ | ✅ 1MB 默认 | ❌ |
create_dirs |
❌ | ✅ 自动创建 | ❌ |
HTTP 工具 (HTTPTool)¶
HTTPTool 基于 aiohttp 实现,支持 GET、POST、PUT、DELETE 四种 HTTP 方法,具备自动重试和超时控制。
class HTTPTool(Tool):
name = "http_request"
input_schema = {
"type": "object",
"properties": {
"url": {"type": "string", "description": "The URL to request"},
"method": {"type": "string", "enum": ["GET", "POST", "PUT", "DELETE"]},
"headers": {"type": "object", "description": "Optional HTTP headers"},
"body": {"type": "object", "description": "Optional JSON body"},
"retries": {"type": "integer", "description": "Retry count", "default": 3}
},
"required": ["url"]
}
参考实现:http.py
重试与超时机制¶
run() 方法实现了完整的重试循环:每次请求失败时捕获 asyncio.TimeoutError 和 aiohttp.ClientError,记录 last_error 并在耗尽重试次数后返回最终错误信息。重试之间无延迟——可根据需要自行扩展退避策略。
参考实现:http.py
响应处理¶
响应类型根据 Content-Type 智能处理:application/json 类型自动解析并格式化(json.dumps(indent=2)),其他类型直接返回文本。响应体超过 max_response_length(默认 50000 字符)时截断并附加提示。
参考实现:http.py
依赖声明:HTTPTool 需要可选依赖 aiohttp。若未安装,BuiltinToolProvider 在注册时会静默跳过(except ImportError: pass),而直接 from jianmu.tool.builtin import HTTPTool 会得到一个占位类,实例化时抛出 RuntimeError 并提示 pip install aiohttp。
参考实现:builtin/__init__.py
搜索工具 (DuckDuckGoSearchTool)¶
DuckDuckGoSearchTool 提供无需 API Key 的 Web 搜索能力,适合获取实时信息、新闻和通用知识。
依赖延迟解析¶
该工具采用巧妙的懒加载策略,在 __init__() 时才解析依赖:
def _resolve_ddgs_cls():
errors = []
for module_name in ("ddgs", "duckduckgo_search"):
try:
module = importlib.import_module(module_name)
ddgs_cls = getattr(module, "DDGS", None)
if ddgs_cls is not None:
return ddgs_cls
errors.append(AttributeError(...))
except ModuleNotFoundError as exc:
errors.append(exc)
raise RuntimeError(
"DuckDuckGoSearchTool requires optional dependency 'ddgs' or 'duckduckgo-search'. "
f"Import details: {detail}"
)
参考实现:duckduckgo.py
它按优先顺序尝试导入 ddgs(新版库名)和 duckduckgo_search(旧版库名),任一成功即返回 DDGS 类。全部失败时抛出包含详细诊断信息的 RuntimeError。
搜索执行¶
搜索通过同步的 DDGS().text() 调用完成,包装在 loop.run_in_executor() 中以适配异步运行环境。结果格式化为 [序号] 标题\nURL: 链接\nSnippet: 摘要\n 的结构化文本。
参考实现:duckduckgo.py
编程辅助工具 (Glob / Grep / FileInfo / StrReplace)¶
coding.py 中的四个工具专为编码场景设计,均继承自 _WorkspaceTool 基类,具备统一的 base_path 工作空间边界约束。
class _WorkspaceTool(Tool):
def __init__(self, *, base_path=None):
self.base_path = Path(base_path).expanduser().resolve() if base_path else None
def _resolve_path(self, path):
# 相对路径相对于 base_path 解析
# 绝对路径验证是否在 base_path 子树内
参考实现:coding.py
| 工具 | 功能 | 关键参数 |
|---|---|---|
GlobSearchTool |
通配符文件搜索 | pattern(glob 模式)、path、include_hidden、limit(200) |
GrepSearchTool |
文件内容搜索 | query、regex、case_sensitive、glob(文件过滤)、limit(200) |
FileInfoTool |
文件/目录元数据 | path → 返回 size / modified / type / extension |
StrReplaceTool |
文本替换(副作用) | path、old、new、regex、count(0=全部) |
参考实现:coding.py
StrReplaceTool 是唯一声明了 effect_tags = ("filesystem:write",) 和 parallel_safe = False 的工具——这意味着 Guard 系统会对其施加写操作保护,且不会与其他文件写入操作并发执行。
参考实现:coding.py
Bash 工具 (BashTool)¶
BashTool 通过 asyncio.create_subprocess_shell() 执行任意 shell 命令。成功时返回 stdout,失败时返回 "Error ({returncode}):\n{stderr}"。
async def run(self, command=None, input=None, **kwargs):
command = str(command or input or kwargs.get("cmd") or "").strip()
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
参考实现:bash.py
命令参数支持三个回退来源:command → input → kwargs["cmd"],为空时返回错误。该工具不施加任何命令白名单或路径限制,生产环境应配合 DockerSandbox 使用。
BuiltinToolProvider:注册、别名与懒加载¶
内建工具的按名解析由 BuiltinToolProvider 统一管理。核心机制如下:
sequenceDiagram
participant C as 调用方
participant P as BuiltinToolProvider
participant R as _builtins 字典
participant M as 工具模块
C->>P: resolve_named_tools(["calculator", "web_search"])
P->>P: _ensure_builtins()
P->>M: import CalculatorTool, register
P->>M: import DuckDuckGoSearchTool, register
P->>P: canonical_tool_name("web_search")
Note over P: alias "web_search" → "duckduckgo_search"
P->>R: get("calculator") → CalculatorTool()
P->>R: get("duckduckgo_search") → DuckDuckGoSearchTool()
P-->>C: [CalculatorTool(), DuckDuckGoSearchTool()]
参考实现:provider.py
内置别名表¶
| 别名 | 规范名 | 说明 |
|---|---|---|
file_read |
read_file |
语义化别名 |
file_write |
write_file |
语义化别名 |
web_search |
duckduckgo_search |
抽象为通用搜索概念 |
python_eval |
python_repl |
更直观的功能名 |
参考实现:provider.py
_ensure_builtins() 在首次使用时以 try/except 包裹每个模块导入,确保单个工具的导入失败(如缺少可选依赖)不会阻止其他工具的注册。
参考实现:provider.py
ToolSet 装配与执行¶
ToolSet 是工具装配的外观类,支持三种构造方式:
# 按名解析
ts = ToolSet.resolve(["calculator", "python_repl", "read_file"])
# 从实例构建
ts = ToolSet.from_tools([CalculatorTool(), my_custom_tool])
# 从 Provider 收集
ts = ToolSet.from_providers([BuiltinToolProvider()])
参考实现:set.py
merge() 方法实现了确定性的去重语义(last-wins),execute() 方法则委托给工具实例的 execute(),并支持 runner 参数传递沙箱。
参考实现:set.py
沙箱执行集成¶
工具执行与沙箱的衔接通过 Tool.execute() 中的 runner 参数实现:
async def execute(self, args, *, runner=None, ...):
call_args = self._merge_call_args(args, ...)
if runner is not None:
return await runner.run(self, call_args, event_context=...)
return await self._execute_local(call_args)
参考实现:base.py
ToolRunner 外观类封装了 LocalSandbox(进程内执行)和 DockerSandbox(容器隔离执行)的统一接口:
class ToolRunner:
async def run(self, tool, args, *, event_context=None):
tool_name = getattr(tool, "name", type(tool).__name__)
emit_runtime_event(bus, "execution.started", ...)
try:
result = await self._sandbox.run_tool(tool, args)
except Exception:
emit_runtime_event(bus, "execution.failed", ...)
raise
emit_runtime_event(bus, "execution.completed", ...)
return result
参考实现:runner.py
每次工具执行都伴随 execution.started → execution.completed(或 execution.failed)的遥测事件流,为可观测性提供完整的生命周期追踪。
完整工具清单¶
| 规范名 | 类 | 依赖 | 副作用 | 并行安全 |
|---|---|---|---|---|
calculator |
CalculatorTool |
无 | 无 | ✅ |
python_repl |
PythonREPLTool |
无 | 代码执行 | ⚠️ |
read_file |
FileReadTool |
无 | 无 | ✅ |
write_file |
FileWriteTool |
无 | 文件写入 | ✅ |
list_dir |
ListDirTool |
无 | 无 | ✅ |
http_request |
HTTPTool |
aiohttp |
网络请求 | ✅ |
duckduckgo_search |
DuckDuckGoSearchTool |
ddgs 或 duckduckgo-search |
网络请求 | ✅ |
glob_search |
GlobSearchTool |
无 | 无 | ✅ |
grep_search |
GrepSearchTool |
无 | 无 | ✅ |
file_info |
FileInfoTool |
无 | 无 | ✅ |
str_replace |
StrReplaceTool |
无 | 文件写入 | ❌ |
bash |
BashTool |
无 | shell 执行 | ⚠️ |
阅读建议¶
本文档聚焦于内建工具的个体设计与使用方式。要理解工具如何在 Agent 运行时中被调用,请参阅 工具与技能节点:ToolExecutor、SkillNode 与约束联动;要了解 @tool 装饰器与自定义工具的开发,请参阅 工具抽象:Tool 基类、@tool 装饰器与 ToolSet 工具集装配;沙箱隔离执行的细节则在 执行环境:LocalSandbox 与 DockerSandbox 的工具隔离运行 中深入阐述。