はじめに / 対象と前提

Claude API のツール実行(Tool Use)を実装していると、あるタイミングから tool_use ブロックが 1 回の応答に複数個 返ってくるようになる。並列ツール実行(parallel tool use)だ。

ここで素直に「1 個ずつ処理して、1 個ずつ返す」と書くと、API が 400 を返して会話が完全に止まる。自分は最初これで 2 時間溶かした。

想定読者: Claude API で自前のツールループを書いている人(フレームワーク非依存)
前提: Python 3.13.x / anthropic Python SDK(pip install -U anthropic)/ モデルは claude-sonnet-5
対象外: Claude Code や Agent SDK 側の話。ここは 生の Messages API を自分で回す場合の話

TL;DR

Claude は 1 応答で複数の tool_use を返す。全部の結果を「次の 1 通の user メッセージ」にまとめて返すのが唯一の正解
分割送信・返し漏れはどちらも 400 tool_use ids were found without tool_result blocks になる
ツールが失敗しても投げっぱなしにせず、is_error: true の tool_result として返す
並列が邪魔なときは tool_choice に disable_parallel_tool_use: true

手順 / 動かし方

1. 複数 tool_use を受け取る

import anthropic

client = anthropic.Anthropic()
messages = [{"role": "user", "content": "東京と大阪の天気を比べて"}]

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=2048,
    tools=TOOLS,
    messages=messages,
)

tool_uses = [b for b in resp.content if b.type == "tool_use"]
print(resp.stop_reason, len(tool_uses))  # -> tool_use 2

stop_reason == "tool_use" のとき、content の中に tool_use ブロックが 1 個以上 入っている。2 個以上来る前提でコードを書く。

2. 実際に並列で走らせる

モデルが「並列で呼んでいい」と判断しただけで、並列実行するのは自分のコード。ここを直列で回すと速度メリットはゼロになる。

import asyncio

async def run_tool(block):
    try:
        result = await TOOL_IMPLS[block.name](**block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": str(result),
        }
    except Exception as e:  # 例外を外へ逃がさない
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": f"{type(e).__name__}: {e}",
            "is_error": True,
        }

results = await asyncio.gather(*(run_tool(b) for b in tool_uses))

3. 1 通にまとめて返す

messages.append({"role": "assistant", "content": resp.content})
messages.append({"role": "user", "content": list(results)})  # ← まとめて 1 通

resp2 = client.messages.create(
    model="claude-sonnet-5", max_tokens=2048, tools=TOOLS, messages=messages,
)

assistant の応答は resp.content を そのまま 積む。テキストブロックだけ抜いて詰め直すと tool_use が消えて破綻する。

ハマりどころ

1. tool_result を分けて送ると 400

最初に書いたのがこれ。ツールごとに user メッセージを作って append していた。

anthropic.BadRequestError: 400
messages.1: `tool_use` ids were found without `tool_result` blocks immediately after:
toolu_01A... Each `tool_use` block must have a corresponding `tool_result` block
in the next message.

原因: 仕様上、直前の assistant にある tool_use は すべて次の 1 メッセージ内 で回収されなければならない。2 通に分けると 1 通目の時点で「未回収の id がある」と判定される。

回避策: tool_result ブロックを配列に貯めて、最後に 1 回だけ append する。順番は tool_use の並びと一致していなくても tool_use_id で突合されるので問題ない。ただし tool_result は content の 先頭側 に置くこと(補足テキストを足すなら後ろ)。

2. ツールが 1 個コケると全部道連れ

asyncio.gather をデフォルトのまま使い、ツール内の例外をそのまま上げると gather ごと落ちる。結果として 成功した分の tool_result も送られず、リトライしても同じ 400 が出続ける。

回避策: 上のコードのように run_tool 内で例外を捕まえ、is_error: True で返す。エラーを隠さず本文に入れるとモデルが自分で別のツールを試したり、引数を直して再呼び出ししてくれる。「例外を握り潰す」のではなく「例外をモデルへの入力に変換する」のがポイント。

3. ストリーミングで入力 JSON が混ざる

stream=True で受けると、複数ツールの引数が input_json_delta として流れてくる。ここで partial_json を 1 本の文字列に連結すると、2 つのツールの JSON が混ざって壊れる。

buffers = {}
for event in stream:
    if event.type == "content_block_start":
        buffers[event.index] = ""
    elif event.type == "content_block_delta" and event.delta.type == "input_json_delta":
        buffers[event.index] += event.delta.partial_json  # index ごとに分ける

回避策: バッファは必ず event.index をキーにした辞書で持つ。イベントは index が交互に来る前提で書く。

背景・補足

並列ツール実行を そもそも切りたい ケース(順序に副作用がある、DB 書き込みを直列化したい等)は、モデル側に出させない方が早い。

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=2048,
    tools=TOOLS,
    tool_choice={"type": "auto", "disable_parallel_tool_use": True},
    messages=messages,
)

これで 1 応答あたりの tool_use は最大 1 個になる。往復回数は増えるので、レイテンシとのトレードオフになる。

なお extended thinking を併用している場合、assistant を積み直すときに thinking ブロックも落とさずそのまま含める必要がある。ここを削ると署名検証で弾かれる。

まとめ

複数 tool_use は「1 応答 : 1 まとめ返し」が鉄則。分割送信は 400
例外は上げずに is_error: true の tool_result に変換して返す
実際の並列化は自分のコードの責任(asyncio.gather)
ストリーミングのバッファは index キーで分ける
直列化したいなら disable_parallel_tool_use: True の一行で済む