工具开发实战

工具是 agent 的手和脚。一个 agent 的能力上限，几乎完全由工具质量决定——模型智力是租来的，工具是你自己的。

一、工具描述就是提示词

模型选工具、传参数的唯一依据是 description 和参数 schema。三条铁律：

1. 说清「什么时候用」+「什么时候别用」
   ✗ "查询数据库"
   ✓ "查询订单数据库。需要知道用户订单状态、金额、时间时使用；
      用户只问商品信息时不要用本工具（用 search_products）"

2. 每个参数写清楚格式与示例
   "date": "日期，格式 YYYY-MM-DD，如 2026-09-09。
            用户说「昨天」时先换算成具体日期再传"

3. 输出格式在 description 里约定
   "返回 JSON：{status, amount, created_at}。
    订单不存在时返回 {error: 'not_found'}"

更深层的原理（工具描述即提示词、MCP 生态）见 Harness 工具系统篇。

二、三类真实工具的实现模板

1. HTTP API 工具（最常见）

import httpx

def make_api_tool():
    return {
        "name": "search_orders",
        "description": "按用户手机号查询最近订单列表",
        "input_schema": {
            "type": "object",
            "properties": {
                "phone": {"type": "string", "description": "用户手机号，11 位"},
                "limit": {"type": "integer", "description": "返回条数，默认 5，最大 20"},
            },
            "required": ["phone"],
        },
    }

async def search_orders(phone: str, limit: int = 5) -> str:
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(
                "https://api.internal.example.com/orders",
                params={"phone": phone, "limit": min(limit, 20)},
            )
            r.raise_for_status()
            return r.text[:20_000]          # 截断！工具输出直接吃上下文
    except httpx.TimeoutException:
        return "ERROR: 订单服务超时（10s），建议告知用户稍后再试"
    except httpx.HTTPStatusError as e:
        return f"ERROR: 订单服务返回 {e.response.status_code}"

2. 数据库工具（参数化，防注入）

# 绝不拼接 SQL —— 模型生成的参数也是「不可信输入」
async def query_db(sql: str, params: dict) -> str:
    # 进一步收紧：只允许白名单查询
    if not sql.strip().upper().startswith("SELECT"):
        return "ERROR: 本工具只允许 SELECT 查询"
    async with db.acquire() as conn:
        rows = await conn.fetch(sql, params)      # 参数化执行
        if not rows:
            return "（无结果）"
        return json.dumps([dict(r) for r in rows[:20]], default=str)[:20_000]

工具描述里直接写明约束：

"description": "查询业务库（只读）。只支持 SELECT。
 表：users(id, name, phone), orders(id, user_id, status, amount, created_at)。
 status 取值：pending/paid/shipped/refunded"

把表结构写进 description，模型就能自己拼对 SQL——工具描述是模型的 API 文档。

3. 有副作用的工具（写操作要防护）

async def refund_order(order_id: str) -> str:
    # 防护 1：幂等 —— 重复调用不重复退款
    if await already_refunded(order_id):
        return "该订单已退款，无需重复操作"
    # 防护 2：金额校验 —— 超阈值人工介入
    order = await get_order(order_id)
    if order.amount > 5000:
        return "ERROR: 金额超过 5000，需人工审批，已创建审批工单"
    # 防护 3：审计
    await audit_log("refund", order_id)
    return await do_refund(order_id)

三、错误处理：错误是给模型看的

工具报错 ≠ 抛异常给用户。错误信息回注给模型，让它自己决定下一步：

# ✗ 错误：异常冒泡，整个会话崩掉
# ✗ 错误：返回 "failed"，模型不知道为什么、怎么办
# ✓ 正确：可行动的错误信息

def tool_wrapper(fn):
    async def wrapped(**kwargs):
        try:
            return await fn(**kwargs)
        except ValidationError as e:
            return f"ERROR: 参数不合法 —— {e}。请检查参数格式后重试"
        except PermissionError:
            return "ERROR: 无权限执行此操作，请换一种方式或告知用户"
        except Exception as e:
            return f"ERROR: {type(e).__name__}: {e}"
    return wrapped

错误信息设计的公式：发生了什么 + 为什么 + 建议怎么办。模型读到「手机号必须是 11 位数字」会自己修正重试；读到 "error" 只能瞎猜。

四、并行工具执行

模型一轮可以返回多个 tool_use（比如同时查三个订单）。串行执行浪费 wall-clock：

import asyncio

async def run_tools(content_blocks, handlers):
    async def one(block):
        fn = handlers[block.name]
        result = await fn(**block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": result,
        }

    # 无依赖的工具调用全部并发；有依赖时模型会分轮请求
    return await asyncio.gather(*(one(b) for b in content_blocks if b.type == "tool_use"))

注意：tool_result 的顺序要与 tool_use 一一对应（用 tool_use_id 关联），乱序会导致 API 400。

五、工具测试

工具是普通函数——单测覆盖 handler，不测模型：

# 项目根/test/ai/agent-dev/test_tools.py（按目录约定放置）
import pytest

@pytest.mark.asyncio
async def test_search_orders_truncation():
    """工具输出必须截断，防止吃爆上下文"""
    result = await search_orders("13800000000", limit=20)
    assert len(result) <= 20_000

@pytest.mark.asyncio
async def test_refund_idempotent():
    """退款幂等：第二次调用不重复扣款"""
    first = await refund_order("o_123")
    second = await refund_order("o_123")
    assert "已退款" in second

def test_rejects_non_select():
    with pytest.raises(AssertionError):
        ...

模型行为层面的测试（「模型会不会选对工具」）属于评估集范畴——构造任务 → 跑 agent → 断言调用了正确工具，见 07 生产化工程。

六、工具设计的反模式清单

✗ 一个大而全的工具（do_everything(action: str)）
  → 拆小：每个工具一个明确意图，模型选择才准

✗ 返回整个 JSON dump 不截断
  → 截断 + 分页参数：「大结果用 search + get_detail 两层工具」

✗ 错误抛异常 / 返回空字符串
  → 返回带建议的 ERROR 文本

✗ 工具名用中文拼音或缩写（cxdd, tksx）
  → 用清晰的英文命名 + 详细 description

✗ 写操作不加幂等与审计
  → 模型重试是常态，重复副作用是事故

✗ 参数让模型传自由文本日期
  → schema 约束格式，description 给示例

本篇小结

• 工具质量决定 agent 上限；description 是写给模型看的 API 文档
• 三大模板：HTTP（截断+超时）、SQL（白名单+参数化）、写操作（幂等+审计）
• 错误当数据回注：发生了什么 + 为什么 + 怎么办
• 无依赖工具并发执行；handler 用单测覆盖，选工具行为用 eval 覆盖

下一篇：结构化输出——让模型的输出变成程序能直接消费的数据。