通过网关跑 Claude Tool Use:流式、错误处理与一轮 Agent 循环的真实成本

Tool Use 是每个 Claude agent 的核心。也是流量一旦经过网关,最容易"安静地坏掉"的地方。这篇文章把一轮 agent 循环里的协议来回、流式增量、重试边界和真实 token 成本拆开讲清楚,所有示例都跑在透明转发的 BUZZ 上。

为什么 Tool Use 这部分最容易出问题

纯文本 completion 走网关其实很宽容。请求上去、token 流回来,即使中间是个会缓冲响应、会"清洗" JSON 的粗糙代理,最终也基本能跑出能用的结果。Tool Use 不一样。它是嵌在同一个 HTTP 流里的双向协议,对模型产出的 tool_use 块和客户端必须回写的 tool_result 块有严格顺序约束。中间任何一层只要碰了请求体、把 JSON 空白做归一化、丢掉"未知字段",或者把 SSE 流缓冲到结束才一次性吐出来,迟早会让 agent 循环出现一种特别难复现的崩坏。

这就是为什么 BUZZ 把 tool-use 流量作为透明转发的核心场景。你发上去的字节就是上游收到的字节,上游返回的字节就是你拿到的字节。本文把这条契约在工程上意味着什么、失败模式藏在哪里、以及一旦把 Anthropic 在开启 tools 时引入的额外开销算进来,一轮 agent 循环到底烧多少钱,全部拆开讲。

Tool Use 的一个完整来回长什么样

Anthropic Messages API 把工具调用表达成 content block,而不是另开一条侧通道。一次正常的 assistant 回合返回的是 content block 列表,每个块带一个 type:text、tool_use 或 thinking。模型决定调用工具时,响应体里会出现这样一段:

{
  "id": "msg_01Abc",
  "type": "message",
  "role": "assistant",
  "model": "claude-sonnet-4-6",
  "stop_reason": "tool_use",
  "content": [
    {
      "type": "text",
      "text": "I'll look that up for you."
    },
    {
      "type": "tool_use",
      "id": "toolu_01XyZ",
      "name": "get_weather",
      "input": {"city": "Tokyo", "units": "metric"}
    }
  ],
  "usage": {"input_tokens": 412, "output_tokens": 87}
}

客户端需要做的:执行工具、抓住返回值,然后发起下一次请求 —— 把上面那个 assistant 回合原样回放,再追加一个新的 user 回合,user 回合里带一个匹配的 tool_result 块:

{
  "model": "claude-sonnet-4-6",
  "max_tokens": 1024,
  "tools": [...],
  "messages": [
    {"role": "user", "content": "What's the weather in Tokyo?"},
    {"role": "assistant", "content": [
      {"type": "text", "text": "I'll look that up for you."},
      {"type": "tool_use", "id": "toolu_01XyZ", "name": "get_weather",
       "input": {"city": "Tokyo", "units": "metric"}}
    ]},
    {"role": "user", "content": [
      {"type": "tool_result", "tool_use_id": "toolu_01XyZ",
       "content": "18C, partly cloudy"}
    ]}
  ]
}

这套协议有三件事对网关至关重要。第一,tool_result 里的 tool_use_id 必须和模型吐出来的那个 id 一字节不差。第二,assistant 回合必须原样回放,所有块按原顺序,否则模型会"丢失上下文",可能重复调用同一个工具,也可能直接幻觉地认为工具已经跑过。第三,会话每轮都把整段历史回放上来,所以 Prompt Cache 在长循环里几乎是必需品。

透明转发:网关一字节不动地转什么

BUZZ 转发 Claude tool-use 流量时,任何参与协议的字段都不动。具体包括:

• tools 数组,以及每个工具的 name、description、input_schema,还有挂在工具定义上的 cache_control 标记。
• tool_choice 字段,四种形态:auto、any、{"type":"tool","name":"..."}、none。
• messages[].content 里的每一个块,包括 tool_use 和 tool_result,id 和顺序全部保留。
• stream: true 时整条 SSE 流,逐字节、逐事件边界转发。
• 各类 beta 头,例如 Prompt Caching、computer use 实验所需的 anthropic-beta。

工程意义就是:任何已经能跑在 api.anthropic.com 上的 agent 框架,只改一个变量就能跑在 BUZZ 上。Python 官方 SDK 里就是 Anthropic(base_url="https://buzzai.cc", api_key=...)。如果你更喜欢用 OpenAI 兼容形式调用 Claude,base URL 是 https://buzzai.cc/v1,Tool Use 会按 OpenAI 的 tool-call 形态转发。Claude Code 本身可以用 https://buzzai.cc/sh/claudecode.sh 这条一行安装命令把 CLI 指向网关,不用动任何项目文件。

流式 + Tool Use:典型踩坑点

流式响应通过 Server-Sent Events 推送。一旦开启 Tool Use,同一条流里会交错出现文本增量和工具入参增量,客户端必须从这些事件里把每个 block 拼回来。最少要处理这些事件:

• message_start —— 响应开始,带 usage.input_tokens。
• content_block_start —— 在索引 N 上开了一个新块。content_block 字段告诉你它是 text、tool_use 还是 thinking。
• content_block_delta —— 真正的增量数据。文本是 text_delta;Tool Use 是 input_json_delta,里面 partial_json 字符串需要你拼接。
• content_block_stop —— 索引 N 的块结束。
• message_delta —— 带 stop_reason 和最终的 usage.output_tokens。
• message_stop —— 整个响应结束。

有个坑要警惕:input_json_delta 的每个 chunk 单独看 不是合法 JSON。你必须把同一个块的所有 delta 拼完,在 content_block_stop 时一次性 parse。那种"顺手在路上 validate JSON"的代理会直接破坏这条流;那种"对每个 chunk 调 json.loads"的客户端在第二个字节就会崩。下面这段 handler 用 Anthropic Python SDK,把 base URL 指向 BUZZ,正确处理整条流:

import json
import os
from anthropic import Anthropic

client = Anthropic(
    base_url="https://buzzai.cc",
    api_key=os.environ["BUZZ_API_KEY"],
)

tools = [{
    "name": "get_weather",
    "description": "Get current weather for a city.",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "units": {"type": "string", "enum": ["metric", "imperial"]},
        },
        "required": ["city"],
    },
}]

# Accumulator: index -> {"type": ..., "name": ..., "id": ..., "input_buf": ""}
blocks = {}

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Weather in Tokyo, metric please."}],
) as stream:
    for event in stream:
        t = event.type

        if t == "content_block_start":
            cb = event.content_block
            blocks[event.index] = {
                "type": cb.type,
                "name": getattr(cb, "name", None),
                "id": getattr(cb, "id", None),
                "input_buf": "",
                "text_buf": "",
            }

        elif t == "content_block_delta":
            d = event.delta
            block = blocks[event.index]
            if d.type == "text_delta":
                block["text_buf"] += d.text
                print(d.text, end="", flush=True)
            elif d.type == "input_json_delta":
                block["input_buf"] += d.partial_json

        elif t == "content_block_stop":
            block = blocks[event.index]
            if block["type"] == "tool_use":
                # Now and only now is the JSON complete.
                block["input"] = json.loads(block["input_buf"] or "{}")
                print(f"\n[tool_use] {block['name']}({block['input']})")

    final = stream.get_final_message()
    print(f"\nstop_reason={final.stop_reason}, usage={final.usage}")

三个细节值得记住。input_buf 只在 content_block_stop 时 parse,绝不提前。content_block_start 里抓到的 id,就是你下一轮提交结果时要回写的 tool_use_id。SDK 的 get_final_message() 会给你一个完全拼好的 message 对象,可以直接 append 到 messages 作为下一轮的 assistant 回合,不用自己手动重建。

成本拆解:一轮 Agent 循环到底烧多少 token

Tool Use 在你的 prompt 和模型响应之上,会稳定加上两笔 overhead,而且都按输入 token 计费:

• Tool-Use 系统提示。一旦请求里 tools 非空,Claude 会内部 prepend 一段教模型"怎么生成 tool_use 块"的系统提示。Claude 4.x 上每次请求是 346 input tokens。
• 工具定义。tools 数组里每个工具被序列化成 JSON Schema 计入输入。像上面那个 get_weather 这样的小定义大约 60-100 tokens;一个带嵌套对象、enum 和详细描述的复杂 schema,可能跑到 200-400 tokens。

把它具体化。一次单轮调用触发一个工具:600 tokens 用户提示,3 个工具定义平均每个 120 tokens,346 tokens 的 tool-use 开销,模型回了 40 tokens 的开场文本加上一个 tool_use 块,块里 JSON 入参 30 tokens。按 /api/pricing 上 Sonnet 的价目结算:

第 2 轮接着算。客户端跑了工具,假设拿到 250 tokens 的结果,把整段历史加上 tool_result 重新发上去。同样的 346 tokens 系统提示和 360 tokens 工具定义又收一次,因为只要请求带 tools 它们就会被计入。本轮新输入是原来 600 tokens 用户提示 + 70 tokens 原样回放的 assistant 回合 + 250 tokens 的 tool_result,输入小计 1626 tokens。如果模型这一轮回了 180 tokens 输出,完整两轮循环的总账是 1306 + 70 + 1626 + 180 = 3182 tokens(按 Sonnet 价)。

有两个能压成本的杠杆。第一,在 tools 数组末尾加一个 cache_control 断点;第二轮起 346 + 360 = 706 tokens 的 overhead 走 cache read 价,只要基础输入价的 0.1x。第二,如果你有 system prompt,也把它标成可缓存。对一个跑 5-10 轮的长 agent 循环来说,这两个断点通常能把输入账单砍一半。具体倍数和实时单价见 https://buzzai.cc/api/pricing —— 网关原样转发 cache_control,缓存行为和直连 Anthropic 一致。

错误处理与重试边界

一个 agent 循环里有 两条独立的 重试边界:一条是对网关的 API 调用,一条是本地工具执行本身。把这两条混在一起处理,是 agent 循环里产生重复副作用最常见的原因 —— 比如发了两次通知、扣了两次款。

规则按边界分别套:

• 上游 5xx(502 / 503 / 504 / 529)。这些来自网关或更上游。请求要么没执行,要么执行了但客户端这边没看到任何副作用。指数退避重试。三次,1s / 3s / 9s 加 jitter,是个还不错的默认值。
• 429 限流。有 retry-after 就遵守,没有就走和 5xx 一样的退避。BUZZ 把上游的限流头原样透传。
• 除 429 外的其它 4xx。不要重试。400 通常意味着请求本身就构造错了,常见原因是 tool_use_id 缺失,或者 tool_result 没匹配上之前发出过的 tool_use。修构造逻辑,不是修网络。
• 工具执行失败。不要重试 Claude 调用。返回一个 is_error: true 的 tool_result,带一句简短描述。模型下一轮会自适应,通常会换参数重试,或者反过来问用户。
• 工具执行超时。形态一样:杀掉本地任务,返回 {"type":"tool_result","tool_use_id": "...", "is_error": true, "content": "Tool timed out after 30s"}。会话继续,模型决定下一步。
• tool 入参 JSON 不合法。Claude 4.x 上很少见,但模型被要求嵌入大字符串时偶尔会发生。在 parse 拼好的 input_buf 时 catch json.JSONDecodeError,返回一个描述解析失败的 is_error tool_result。模型下一轮自己改正。

把这些粘在一起的核心纪律是:工具执行失败是模型需要的数据,不是要往上抛的异常。把它包成一个结构化的 tool_result,会话状态保持合法。把 assistant 回合扔掉、悄悄重试,才是产生重复扣款和"agent 神志不清"的源头。

一个能直接抄走的 Agent 循环

下面是一段 50 行左右的完整 agent 循环,可以直接粘进脚本。它连到 BUZZ、注册两个工具,跑到模型停止调用工具为止,同时管住 turn 数和 token 总预算。代码刻意写得紧凑,方便你一眼读完。

import json
import os
from anthropic import Anthropic

client = Anthropic(base_url="https://buzzai.cc", api_key=os.environ["BUZZ_API_KEY"])

TOOLS = [
    {"name": "get_weather", "description": "Current weather for a city.",
     "input_schema": {"type": "object", "required": ["city"],
        "properties": {"city": {"type": "string"}}}},
    {"name": "search_docs", "description": "Search internal documentation.",
     "input_schema": {"type": "object", "required": ["q"],
        "properties": {"q": {"type": "string"}, "limit": {"type": "integer"}}}},
]

def execute_tool(name, args):
    try:
        if name == "get_weather":
            return f"{args['city']}: 18C, partly cloudy"
        if name == "search_docs":
            return f"Top result for {args['q']!r}: see internal-runbook#42"
        return {"is_error": True, "content": f"unknown tool {name}"}
    except Exception as e:
        return {"is_error": True, "content": f"{type(e).__name__}: {e}"}

def run_agent(user_prompt, max_turns=15, token_budget=200_000):
    messages = [{"role": "user", "content": user_prompt}]
    total_tokens = 0
    for turn in range(max_turns):
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            tools=TOOLS,
            messages=messages,
        )
        total_tokens += resp.usage.input_tokens + resp.usage.output_tokens
        if total_tokens > token_budget:
            raise RuntimeError(f"token budget exceeded: {total_tokens}")

        # Replay the assistant turn verbatim.
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason != "tool_use":
            return resp, messages

        # Build a single user turn with one tool_result per tool_use block.
        results = []
        for block in resp.content:
            if block.type != "tool_use":
                continue
            output = execute_tool(block.name, block.input)
            if isinstance(output, dict) and output.get("is_error"):
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "is_error": True, "content": output["content"]})
            else:
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": str(output)})
        messages.append({"role": "user", "content": results})
    raise RuntimeError(f"agent did not converge within {max_turns} turns")

if __name__ == "__main__":
    final, history = run_agent("What's the weather in Tokyo, and find docs on rate limiting.")
    text = next((b.text for b in final.content if b.type == "text"), "")
    print(text)

几个细节值得点出来。assistant 回合直接 append 的是 resp.content,SDK 对象能干净地往下一次请求里 round-trip,不需要手动重建。同一个 assistant 回合里每个 tool_use 块,在下一轮 user 回合里都对应 一个 tool_result 块,顺序保持一致,全部塞进同一条 user 消息。execute_tool 永远不抛异常出循环 —— 失败被包装成结构化的 is_error 结果,会话保持合法。最后两道闸 max_turns 和 token_budget,主要防两种失控:模型一直调工具不收尾,或者某个工具返回了 50k token 的大块、模型反复读它。

把这段代码直接对着网关跑,base_url 已经设好。BUZZ_API_KEY 在 https://buzzai.cc/ 账户里取,可用模型列表见 https://buzzai.cc/models。

FAQ

结语

Tool Use 是 agent 系统赢得或失去可靠性的地方。协议本身不大,但很不宽容:tool_use_id 必须精确 round-trip,assistant 回合必须原样回放,SSE 流不能被缓冲,input_json_delta 里的 JSON 必须等拼完再 parse。透明网关不在这些事情上添乱 —— 字节原样转发,流式契约保持,内容什么都不留,Tool Use 的开销按 Anthropic 一样的方式计费,价格再叠加网关的折扣。

如果你从零开始,上面那段 agent 循环就是一份完整参考。如果你已经有 agent 框架,切到网关只是改一个变量 —— Python 里 base_url="https://buzzai.cc",TypeScript 里同理;Claude Code 走 https://buzzai.cc/sh/claudecode.sh 即可。实时单价在 /api/pricing,可用模型列表在 /models。