5.1 自定义工具与宿主注入

本节你会学到

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

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

Tool 基类回顾

from abc import ABC, abstractmethod
from typing import Dict, Any
from agentao.tools.base 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}"

6 个必须掌握的要点：

属性/方法	必填	说明
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；权限引擎/Plan 模式据此优化

为什么 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
• 近似但不完全匹配的工具名可修复为已注册工具名
• 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 时，调用我"。

路径解析助手

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.tools.base 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，让 PermissionEngine 和 Plan 模式能优化。
• 通过 extra_tools= 或 add_tool() 注入工具，确保能力绑定和名字校验；disable_tools / enabled_tools 只用于缩减模型可见 schema。
• execute() 里捕获异常并返回错误字符串——未捕获的异常会让整个 chat() 调用挂掉。
• 一个工具一个聚焦的职责。description 含糊会被到处误调。

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