工具

Agent 可以理解用户的请求，但要查询真实数据或执行操作，还需要有调用外部能力的途径。而工具（Tool）就是 Agent 的手脚：所有的处理逻辑都可以封装成工具，然后 Agent 再根据用户请求决定是否调用。

本节将以一个查询订单的示例开始，逐步说明如何定义工具、模型怎样调用它，以及怎样处理参数、结果和运行时数据。

自定义工具

假设我们已经有一个查询订单的 Python 函数。要想 Agent 能调用它，就必须用 Bub SDK 中的 @tool 装饰器将其标记，并在初始化 Agent 时传递：

from bub.core import Agent
from bub.core.tools import tool


ORDERS = {
    "A123": {
        "order_id": "A123",
        "status": "shipped",
        "item": "机械键盘",
        "amount": 399.0,
    }
}


@tool
def get_order(order_id: str) -> dict[str, str | float]:
    """按订单号查询订单状态、商品和金额。"""
    if order := ORDERS.get(order_id):
        return order
    return {"order_id": order_id, "error": "order_not_found"}


agent = Agent(
    tools=[get_order],
    system_prompt="用户询问订单时，使用工具查询，不要猜测订单状态。",
)

@tool 装饰器会把带类型标注的 Python 函数转换为 Agent 可用的工具。工具函数负责实际业务逻辑，而模型则根据工具名、参数类型和说明来帮助它判断何时以及怎样调用。

工具调用流程

模型并不会直接执行 Python 代码。而是它先提出工具调用请求，然后 Bub 再校验参数并执行对应函数，最后把结果交回模型：

所以当我们调用 agent.run() 方法时，可以从输出的结果中分别看到模型请求的工具和工具返回的结果：

async with agent.run(prompt, model=model, tape="tools-demo") as stream:
    async for event in stream:
        if event.kind == "tool_call":
            for call in event.data["tool_calls"]:
                print(f"模型请求：{call['name']}，参数：{call['arguments']}")
        elif event.kind == "tool_result":
            print(f"工具结果：{event.data['tool_results']}")
        elif event.kind == "text":
            print(event.data["delta"], end="", flush=True)

一次工具调用会经过两个步骤：

1. 模型先请求 get_order 工具，Bub 执行函数并把结果交回模型；
2. 模型再根据这个结果生成回答。

但是否调用工具仍由模型决定。

工具说明与约束

使用 @tool 装饰器标记的 Python 函数，其参数类型会被转换为 JSON Schema，而当中的 docstring 则转换为该工具的说明书。因此，参数名和描述都是工具接口的一部分，我们在封装时应确保函数签名和描述的准确性，以确保模型能正确调用。

@tool
def get_order(order_id: str) -> dict[str, str | float]:
    """按订单号查询订单。order_id 是用户提供的完整编号，例如 A123。"""
    ...

如果输入参数较多或需要如范围、长度等约束时，可以借 Pydantic 这样的第三方校验库帮助我们，之后将其作为 @tool 装饰器的 model 参数传入：

from pydantic import BaseModel, Field

from bub.core.tools import tool


class RefundInput(BaseModel):
    order_id: str = Field(description="完整订单编号，例如 A123")
    amount: float = Field(gt=0, description="退款金额，必须大于 0")
    reason: str = Field(min_length=1, description="用户提供的退款原因")


@tool(name="refund_order", model=RefundInput)
def refund_order(param: RefundInput) -> dict[str, str | float]:
    """为已支付订单申请退款；执行前必须完成权限检查。"""
    return {
        "order_id": param.order_id,
        "amount": param.amount,
        "status": "refund_requested",
    }

在执行中，如因参数类型不匹配或 Pydantic 校验失败时，Bub 会把错误作为工具错误结果交回模型。模型可以据此修正参数后重试（但不保证重试一定成功）。

注意，在编写实际的业务逻辑函数时，应区分业务结果和执行错误：

情况	默认行为
参数不符合类型或 Pydantic 约束	转为参数校验工具错误，交回模型
工具抛出普通异常	转为工具错误，交回模型
工具返回业务错误对象，例如 order_not_found	作为正常工具结果交回模型
存储、模型或事件处理器中断调用	异常可能从事件迭代抛出，在集成部分处理

业务错误对象不是 Python 异常，以稳定的结构化结果返回，可以让模型和调用方识别“订单不存在”这类业务状态。

在工具中传入上下文

出于某种需要，有时我们还需要在工具中获取到当前会话、请求级权限或事件总线等信息。可以通过 Bub 注入的 ToolContext 对象来获取：

from bub.core.tools import ToolContext, tool


@tool(context=True)
async def add_order_note(note: str, *, context: ToolContext) -> str:
    """向当前会话追加一条订单备注。"""
    await context.tape.append_event(
        "order.note",
        {"text": note, "request_id": context.state.get("request_id")},
    )
    return f"备注已写入会话 {context.tape.name}"

于是在调用过程中传入 state 参数，工具就能读取到 context.state 的数据：

async with agent.run(
    prompt,
    model=model,
    tape="tools-demo",
    state={"session_id": "tools-demo", "request_id": "req-01"},
) as stream:
    ...

提示

context 参数由 Bub 注入，不会出现在模型可见的参数 Schema 中。而 context.state 只在当前调用内共享，不会自动持久化或恢复。

异步和批量调用

@tool 装饰器不仅适用于普通函数，还适用于异步函数，通常来说对于可能会阻塞程序运行的逻辑我们都应尽可能将其作为异步函数调用，必要时放到单独的线程中执行：

import time
import asyncio

def build_report_blocking(order_id: str) -> None:
    time.sleep(3)

@tool
async def export_order_report(order_id: str) -> str:
    """生成订单报表并返回下载地址。"""
    return await asyncio.to_thread(build_report_blocking, order_id)

需要说明的是，模型可能在一次响应中可能请求多个工具。所以避免达到系统自身的资源瓶颈，Bub 允许我们传入 max_tool_concurrency 参数来限制：

from bub.core import Agent
from bub.core.tools import tool


ORDERS = {
    "A123": {
        "order_id": "A123",
        "status": "shipped",
        "item": "机械键盘",
        "amount": 399.0,
    }
}


@tool
def get_order(order_id: str) -> dict[str, str | float]:
    """按订单号查询订单状态、商品和金额。"""
    if order := ORDERS.get(order_id):
        return order
    return {"order_id": order_id, "error": "order_not_found"}


agent = Agent(
    tools=[get_order],
    system_prompt="用户询问订单时，使用工具查询，不要猜测订单状态。",
    max_tool_concurrency=2,
)

另外，传给 Agent(tools=[...]) 的还可以是其他库的工具，比如 republic.tools.WebSearch()。但它们将由模型提供商执行，不经过本地 ToolExecutor，也不受 max_tool_concurrency 的本地并发限制。

直接调用工具

如果你有自己的想法，明确大概要调什么命令，可以给 Agent 设置 command_prefix 命令前缀。命令仍使用同一份工具实现和参数校验，但不调用模型：

agent = Agent(tools=[get_order], command_prefix=",")

async with agent.run(",get_order A123", model=model, tape="tools-demo") as stream:
    async for event in stream:
        if event.kind == "text":
            print(event.data["delta"], end="")

这和普通函数调用的区别在于，直接命令会产生 Bub 定义事件，以便你现有 Agent 流程逻辑可以统一处理。但调用发生错误时，不会像模型工具错误那样交回模型修正。

代码模式

当工具较多或任务需要条件逻辑时，可以让模型通过 run_code 工具来编写 Python 代码，以满足更灵活的处理需要或在一次执行中组合多个工具：

from bub.core.codemode import CodeMode

code_mode = CodeMode([get_order])

agent = Agent(
    tools=[code_mode.run_code],
    system_prompt=[
        "使用可用工具完成任务；可以编写 Python 调用 tools.*。",
        code_mode.prompt,
    ],
)

模型将生成如下类似可以调用的 Python 代码并执行：

order = await tools.get_order(order_id="A123")
print(order)

安全环境与执行环境

默认情况下，由 Agent 使用的所有工具或代码将在运行 Python 进程的宿主机上被执行，但这并不是独立且安全的沙箱。必要时应进行隔离，并配置具备文件系统、网络和权限等限制。

Bub 允许你通过自定义 Environment 来使用远程或隔离环境作为执行容器。完整协议见 Environment。