はじめに / 対象と前提
Claude API のツール実行(Tool Use)を実装していると、あるタイミングから tool_use ブロックが 1 回の応答に複数個 返ってくるようになる。並列ツール実行(parallel tool use)だ。
ここで素直に「1 個ずつ処理して、1 個ずつ返す」と書くと、API が 400 を返して会話が完全に止まる。自分は最初これで 2 時間溶かした。
- 想定読者: Claude API で自前のツールループを書いている人(フレームワーク非依存)
-
前提: Python 3.13.x /
anthropicPython 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の一行で済む