5.1 自定义工具与宿主注入

本节你会学到

• Tool 子类的 7 大要素：name / description / parameters / execute / requires_confirmation / is_read_only / copies_to_subagents
• 怎样写一个 LLM 真用得上的 description（这件事比代码本身更重要）
• 宿主如何注入、替换、移除或白名单化工具
• Tool / Skill / MCP 三选一的判断标准

自定义工具是让 Agent 调用你的业务 API 的首选方式——比起让 LLM 读你的 OpenAPI 规范再生成 HTTP 请求，直接写一个 Tool 子类既更可靠又更安全。

Tool 基类回顾

from abc import ABC, abstractmethod
from typing import Dict, Any
from agentao.host import Tool

class MyTool(Tool):
    @property
    def name(self) -> str:
        return "unique_tool_name"          # 全局唯一

    @property
    def description(self) -> str:
        return "给 LLM 看的一句话描述——决定 LLM 是否调用这个工具"

    @property
    def parameters(self) -> Dict[str, Any]:
        """JSON Schema，描述参数（会直接传给 LLM 做 function calling）。"""
        return {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "..."},
                "limit": {"type": "integer", "default": 10},
            },
            "required": ["query"],
        }

    @property
    def requires_confirmation(self) -> bool:
        return True     # 写操作、网络、跑命令都应 True

    @property
    def is_read_only(self) -> bool:
        return False    # 纯读返回 True，可优化权限判断

    def execute(self, **kwargs) -> str:
        """真实逻辑。返回字符串，LLM 会读它做下一步决策。"""
        query = kwargs["query"]
        limit = kwargs.get("limit", 10)
        ...
        return f"Found {len(results)} items: {results}"

7 个必须掌握的要点：

属性/方法	必填	说明
name	✅	全局唯一标识；冲突会被 override 并打 warning
description	✅	LLM 唯一的决策依据——写清楚"什么时候用、参数含义、返回什么"
parameters	✅	JSON Schema；任何 OpenAI function calling 支持的 schema 都能用
execute(**kwargs) -> str	✅	返回纯字符串；不能返回 dict/bytes
requires_confirmation	❌	写/网络/危险操作应 True，走 confirm_tool 流程
is_read_only	❌	只读模式可以运行它时为 True：不写文件、不跑命令、不改会话之外的状态。只读模式会拒绝所有返回 False 的工具。它还决定一批调用怎么跑：模型一次回复里的调用，只有每个工具都返回 True 时才并行，否则按顺序逐个执行。只读模式之外，子代理调用（agent_*）会让这一批逐个执行，因为子代理能写文件。后台启动不会：启动返回不代表子代理已经完成，依赖其写入的读取要等任务完成
copies_to_subagents	❌	True 表示允许子代理使用该工具，且以副本形式——见下文。默认 False：你的工具不会进入任何子代理

让子代理能用你的工具

默认情况下，宿主工具不会出现在任何子代理里，它占用的那个名字也一起消失：你注册为 read_file 的工具，不会让内置的 read_file 从它底下露出来。要开启，需显式声明：

class DeployTool(Tool):
    @property
    def copies_to_subagents(self) -> bool:
        return True

之后每次 spawn 会注册你这个实例的一个 copy.copy，而不是实例本身。返回 True 是一个运行时无法校验的承诺，它同时声明三件事：

1. 这个工具适合交给子任务；
2. 它能承受 copy.copy ——浅拷贝，所以子代理的调用无法改写你实例上的属性；
3. 副本与原对象仍然共享的每一个依赖——闭包、HTTP 客户端、连接池、容器句柄——都能 承受父代理与子代理的并发使用，而两者跑在不同线程上。

第 3 条最容易出问题，因为浅拷贝只分离实例自身的属性。如果你的工具需要更深的分离， 实现 __copy__；同一个承诺覆盖它所做的一切。

机制还有三点：

• 每次 spawn 一个副本，而不是每次调用一个，所以工具可以在一个子任务的多次调用之间 保留状态。
• 拷贝或声明本身抛异常，则该工具缺席，并打一条点名异常的 warning。绝不会退回共享原实例， 也绝不会用它可能覆盖掉的那个内置工具顶上。__copy__ 返回 self、或返回名字不同的工具， 同样按缺席处理。
• 必须写成 @property（或直接用一个布尔类属性）。只写 def copies_to_subagents(self) 会被读作没有声明并打日志——绑定方法无论返回什么都是真值，若按真值处理，你写 return False 反而会把工具放进子代理。

这个声明并不盖过 agent 定义里的 tools: 白名单：定义没列出的工具，即使声明了也仍然 缺席。它也不对语义作任何保证——持有当前这个会话状态的工具，比如 CLI 自己的 update_goal，无论拷贝多深都不该声明它。

为什么 execute 只能返回字符串？

因为工具结果要注入到 LLM 消息历史（OpenAI function calling 的 role:tool 消息）。非字符串结果不兼容。正确做法：

def execute(self, **kwargs) -> str:
    data = call_my_api(kwargs)
    # 用 JSON 字符串表达结构化数据，LLM 会解析
    return json.dumps({
        "status": "ok",
        "data": data,
        "count": len(data),
    }, ensure_ascii=False)

大返回（> 几十 KB）应先截断或分页，否则占爆上下文窗口。

工具调用规范化

在工具调用写回对话历史或真正执行之前，Agentao 会规范化模型产出的 function-call payload：

• 参数字符串在能安全修复时会被解析并重新输出为紧凑 JSON
• 工具名只有在规范化（首尾空白、大小写、分隔符、camelCase、结尾的 Tool 后缀）后精确等于某个 已注册名时才会被修复（已注册名也会按同样规则规范化，因此 camelCase 工具名同样可达）；仅仅是 相似的名字会报告为未找到，由模型重新发起调用
• lone UTF-16 surrogate 字符会在 assistant/tool 消息发回严格 provider API 前被清洗
• 每个 assistant tool_call_id 都会得到对应的 role:tool 消息，包括参数解析错误和循环保护中止

这是韧性层，不是 schema 质量的替代。parameters 仍要精确，description 仍要明确；危险操作或业务关键字段仍应在 execute() 内二次校验后再产生副作用。

怎样写一个 "能让 LLM 用对的" description

这是比写代码更关键的环节。差的描述让 LLM 误调用；好的描述让 LLM 知道什么时候该用、用完怎么处理。

❌ 反例

description = "Get orders"

LLM 不知道：什么叫 order？给谁的？参数传什么？返回格式？

✅ 正面示范

description = """
查询客户订单。用于：用户询问"我的订单""最近一单""订单详情"时。

参数：
- `customer_id` (必填): 客户 ID，从用户会话里拿，不要凭空猜
- `status`: 订单状态过滤 ("pending" / "shipped" / "delivered" / "all")，默认 "all"
- `limit`: 返回条数，默认 10，最大 50

返回：JSON 字符串，含 orders 数组；每条包含 id/status/total/created_at 字段。

注意：
- 不要把 customer_id 暴露给用户；响应里只讲订单内容
- 如果 orders 为空，直接告知用户"无订单"
"""

经验：用第二人称写给 LLM 本身看——"当用户说 X 时，调用我"。

描述里的每一句都要属实。 模型按描述行事。描述说返回"全部行"、工具实际最多返回 2000 行，或者工具有 5 种状态、描述只列了 4 种，模型就会基于错误的事实行动。数字从代码里取，不要手写；写明每个默认值；列全工具可能返回的每个取值。Agentao 的内置工具遵守这条规则，并有测试检查（tests/test_tool_description_facts.py）。

路径解析助手

Tool 基类提供两个辅助方法处理文件路径：

class MyFileTool(Tool):
    def execute(self, path: str, **kw) -> str:
        # _resolve_path: 支持 ~ 展开；绝对路径原样；相对路径拼 working_directory
        p = self._resolve_path(path)
        return p.read_text()

self.working_directory 是 Agentao 在注册时自动绑定的——多实例场景下每个 Agent 的工具各自指向不同目录，不会串线。用这两个助手而不是直接 Path(raw) 能免费获得多租户隔离。

注册工具

契约化的注入方式是构造期 Agentao(extra_tools=[...]) 或运行期 agent.add_tool(...)。两者都会替你绑定 working_directory / filesystem / shell，并校验保留名。

from pathlib import Path
from agentao import Agentao
from agentao.transport import SdkTransport

agent = Agentao(
    working_directory=Path("/tmp/session-x"),
    transport=SdkTransport(),
    extra_tools=[MyTool()],          # 第一次 chat() 起可见
)

agent.add_tool(AnotherTool())        # 下一次 chat() / arun() 可见
agent.remove_tool("web_fetch")       # 存在则返回 True

仅在契约 API 不适用时才直接使用底层注册表。agent.tools.register(...) 会绕过能力绑定与校验，而且冲突处理更弱（replace=False 时只打 warning 然后覆盖）：

my_tool = MyTool()
my_tool.working_directory = agent.working_directory   # 显式绑定（否则用进程 cwd）
agent.tools.register(my_tool)

⚠️ 注意：

• extra_tools 只能从代码传入：传已经构造好的 Tool / AsyncToolBase 实例。它不会从 JSON 加载。
• 同名 extra_tools 会刻意替换内置或 agent 工具；名字必须唯一，且不能使用保留的 mcp_ 前缀。
• add_tool(tool) 遇到重名会抛异常，除非显式传 replace=True；remove_tool(name) 对不存在的名字返回 False。
• add_tool / remove_tool 用于两轮之间。模型可见的 schema 在每次 chat() / arun() 前快照一次，本轮中途不会改变。

选择工具面

宿主也可以收缩模型可见的工具面：

你想要…	用
增加自定义工具，或替换内置工具实现	extra_tools= / add_tool(..., replace=True)
隐藏少数不适用的内置工具	disable_tools={...}
只保留一小组 agentao 自有工具	enabled_tools={...}
收缩到仅剩自有工具 + MCP	enabled_tools=set() 加 extra_tools=[...]
会话中途改动工具面	两轮之间用 add_tool() / remove_tool()

disable_tools 与 enabled_tools 互斥。disable_tools 只跳过内置工具。enabled_tools 裁剪内置 / agent 路径工具，并保留 extra_tools、MCP 工具（mcp_*）与仅 plan 工具。

不是安全边界

这些 API 缩减的是模型所见 schema，不是授权。如果某个工具对某租户永远不该运行，应交给权限引擎执行，而不是只靠工具白名单。

完整示例：调用业务 API 的工具

"""你的 SaaS 后端把订单查询暴露给 Agent。"""
import json
from typing import Dict, Any
from agentao.host import Tool

class GetCustomerOrdersTool(Tool):
    def __init__(self, backend_client, tenant_id: str):
        super().__init__()
        self.backend = backend_client
        self.tenant_id = tenant_id   # 绑定到会话

    @property
    def name(self) -> str:
        return "get_customer_orders"

    @property
    def description(self) -> str:
        return (
            "Query this tenant's customer orders. "
            "Use when the user asks about 'my orders', 'order status', etc. "
            "Args: customer_id (required), status (optional: pending/shipped/delivered/all, default all), "
            "limit (optional int, max 50, default 10). "
            "Returns JSON: {status, orders:[{id, status, total, created_at}]}. "
            "Never expose internal tenant_id or api tokens in your reply."
        )

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string"},
                "status": {
                    "type": "string",
                    "enum": ["pending", "shipped", "delivered", "all"],
                    "default": "all",
                },
                "limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 10},
            },
            "required": ["customer_id"],
        }

    @property
    def requires_confirmation(self) -> bool:
        return False    # 读 API，不需要用户额外确认

    @property
    def is_read_only(self) -> bool:
        return True

    def execute(self, **kwargs) -> str:
        try:
            orders = self.backend.list_orders(
                tenant_id=self.tenant_id,
                customer_id=kwargs["customer_id"],
                status=kwargs.get("status", "all"),
                limit=min(kwargs.get("limit", 10), 50),
            )
        except Exception as e:
            return json.dumps({"status": "error", "message": str(e)})
        return json.dumps({
            "status": "ok",
            "orders": [o.to_dict() for o in orders],
        }, ensure_ascii=False)


# --- 在 Web 处理里的使用 ---
def make_agent_for_tenant(tenant, backend):
    return Agentao(
        working_directory=Path(f"/tmp/{tenant.id}"),
        transport=SdkTransport(...),
        extra_tools=[
            GetCustomerOrdersTool(backend, tenant.id),
            CreateRefundTool(backend, tenant.id),
            SendEmailTool(backend, tenant.id),
        ],
    )

⚠️ 写 Tool 的常见陷阱

上线前先确认这几条

真实生产 bug，每一条都付过学费：

• ❌ 在 execute() 里抛异常 —— 整个 chat() 调用直接挂掉
• ❌ description 写得太泛 —— LLM 会到处误调
• ❌ 有副作用的工具忘了 requires_confirmation=True —— 等于把上膛的枪交给 LLM
• ❌ 参数没边界 —— LLM 可能传 limit=99999
• ❌ 返回数据太大 —— 直接撑爆上下文窗口

下面每一条都附完整模式 + 修法。

❌ 在 execute 里抛异常

def execute(self, **kwargs) -> str:
    return self.backend.create_invoice(...)  # 抛 HTTPError 怎么办？

抛出去会把 Agent 整个 chat() 打断。捕获并返回错误字符串让 LLM 看见并决定怎么处理：

def execute(self, **kwargs) -> str:
    try:
        result = self.backend.create_invoice(...)
        return json.dumps({"status": "ok", "id": result.id})
    except BackendError as e:
        return json.dumps({"status": "error", "message": str(e)})

❌ 描述过于宽泛

"Do things with customer data" —— LLM 会乱用它。每个工具只做一件事，描述只讲这件事。

❌ 忘记 requires_confirmation=True

写操作、退款、发邮件、跑命令、删数据——一切有副作用的都应确认。没加这个就等于给 LLM 一把没有保险的枪。

❌ 参数没做边界检查

LLM 可能传 limit=99999。工具里务必 clamp/校验：

limit = min(max(1, kwargs.get("limit", 10)), 50)

❌ 返回数据太大

return json.dumps(all_1000_orders)   # 可能 500KB

大返回占爆 context 且 LLM 处理慢。先截断、分页、摘要，让 LLM 决定要不要翻下一页：

return json.dumps({
    "status": "ok",
    "orders": orders[:10],
    "total_count": len(orders),
    "has_more": len(orders) > 10,
    "next_cursor": cursor if len(orders) > 10 else None,
})

工具 vs 技能 vs MCP：怎么选

你的需求	用什么
调 HTTP API / 数据库 / 内存对象	工具（本节）
教 LLM "按公司规范做事"	技能（5.2）
集成外部现成的工具服务（GitHub、文件系统、数据库）	MCP（5.3）

一个产品通常三者混用：工具封自己的业务、MCP 接第三方、技能统一风格。

TL;DR

• Tool 必须返回字符串（role:tool 消息）；不能直接返回 dict 或 bytes。业务数据要 JSON-stringify 并控制大小。
• description 是 LLM 决策的唯一输入：明确 什么时候用、参数含义、返回结构、硬规则。
• 有副作用（写、删、网络、执行）的工具一律 requires_confirmation=True；只有对会话之外没有任何影响的工具才设 is_read_only=True，因为只读模式正是据此决定能否运行它。
• 通过 extra_tools= 或 add_tool() 注入工具，确保能力绑定和名字校验；disable_tools / enabled_tools 只用于缩减模型可见 schema。
• execute() 里捕获异常并返回错误字符串——未捕获的异常会让整个 chat() 调用挂掉。
• 一个工具一个聚焦的职责。description 含糊会被到处误调。

→ 下一节：5.2 技能与插件目录