【第3回】Claude API Python 実践入門 — 非同期処理とエラーハンドリングで本番品質に仕上げる

AI

Claude API の Tool Use(関数呼び出し)は、Claudeと外部システムを連携させるときに必須の機能です。「そろそろ覚えないとまずいな」と思いつつ後回しにしてたんですが、自分でエージェントっぽいものを作ろうとしたときにどうしても必要になって、ちゃんと調べました。

この記事では、ツール定義の書き方から呼び出しループの実装、並列ツール呼び出しへの対応、エラーハンドリングまでをまとめています。

  • Tool Use(関数呼び出し)の仕組みと基本フロー
  • ツール定義とPythonでの呼び出しループの実装
  • 並列ツール呼び出しへの対応方法
  • tool_choice による制御と無限ループ対策

前回のおさらい

前回(第2回)はストリーミングレスポンスの扱い方を紹介しました。stream=True にして逐次出力する方法や、AsyncAnthropic を使った非同期ストリーミングの実装まで触れましたね。今回はその発展として「Tool Use(ツール使用)」、いわゆる関数呼び出し機能を取り上げます。

Tool Useとは何か

Claude API の Tool Use は、いわゆる function calling 相当の仕組みです。ツール定義をリクエストに含めておくと、Claudeが必要と判断したタイミングで stop_reason: "tool_use"tool_use コンテンツブロックを含むレスポンスを返してきます。

イメージとしては——

  • 「今日の天気は?」→ Claude: 「get_weather ツールを呼んで」と返答
  • アプリ側が実際にAPIを叩いて結果を取得
  • その結果をClaudeに渡す → 最終的な自然言語の回答が生成される

Claudeがコードを実行するわけではなく、「何を呼べばいいか」を指示するだけです。実行は自分たちのコードが担います。この切り分けを最初に理解しておくと混乱が減ります。

余談ですが、Anthropic側で動く「サーバーツール(server tools)」もあって、たとえば Web 検索は API が検索自体を実行して結果をモデルに渡してくれます(レスポンスには server_tool_use のブロックが出ます)。自前で「検索APIを叩く関数」を書かなくていいケースがあるので便利です。
とはいえ、自前の関数を呼ぶ「クライアントツール」も基本中の基本なので、今回はここをしっかり押さえます。

Claude API Tool Use の基本実装フロー

クライアントツールのフローはざっくりこんな感じです:ツール定義(名前・説明・パラメータのスキーマ)をリクエストに含め、Claudeが必要と判断したら stop_reason: "tool_use" で返ってくる。レスポンス内の tool_use ブロックを取り出して関数を実行し、tool_result として結果を渡して再度リクエスト、stop_reason: "end_turn" になったら完了です。

この「ループ」がツール使用の核心です。1回で終わらないこともあるので、ちゃんとループで回す実装にしておく必要があります。

ツール定義の書き方

ツールは次のような辞書形式で定義します。input_schema はJSON Schema形式です。

tools = [
    {
        "name": "get_weather",
        "description": "指定した都市の現在の天気を取得する。気温はセルシウス、天気は日本語で返す。",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "都市名(例: 東京、大阪、札幌)"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度の単位"
                }
            },
            "required": ["city"]
        }
    }
]

ツール定義では description フィールドが重要で、何を返すかや返却フォーマットまで明示しておくと、Claudeが正しく呼び出しを組み立てやすくなります。自分はここをサボって最初だいぶ苦労しました。

Pythonでの呼び出しループ実装

ツール定義ができたら、呼び出しループを実装します。以下がシンプルな実装例です。

import anthropic
import json

client = anthropic.Anthropic()

def get_weather(city: str, unit: str = "celsius") -> dict:
    # 実際はAPIを叩く。ここではダミー
    return {"city": city, "temp": 22, "condition": "晴れ", "unit": unit}

def run_tool(name: str, inputs: dict):
    if name == "get_weather":
        return get_weather(**inputs)
    raise ValueError(f"Unknown tool: {name}")

def chat_with_tools(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text

        if response.stop_reason == "tool_use":
            # assistantの応答(content blocks)をそのままメッセージ履歴に追加
            messages.append({"role": "assistant", "content": response.content})

            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue

                try:
                    result = run_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False),
                    })
                except Exception as e:
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(e),
                        "is_error": True,
                    })

            # tool_result は「次の user メッセージの content」に配列で渡す
            messages.append({"role": "user", "content": tool_results})

ポイントをいくつか:

  • response.content(アシスタントの返答)をそのまま messages に追加する必要がある。テキストだけ追加してはダメ
  • is_error: truetool_result に含めると、Claudeが「ツールが失敗した」ことを前提にリカバリを試みやすくなります
  • tool_use_idblock.id を正しく対応させること。複数ツールが同時に呼ばれることがあるので、ループで処理する

並列ツール呼び出しへの対応

Claudeはデフォルトで複数ツールを並列に呼ぶことがあります。たとえば「東京と大阪の天気を比べて」みたいなプロンプトだと、get_weather(東京)get_weather(大阪) を同じレスポンス内で複数の tool_use ブロックとして返してくることがあります。

上のコードはすでに for block in response.content で全 tool_use ブロックを処理しているので、並列呼び出しにも対応できています。複数の tool_result を配列にまとめて一度に返すのがポイントです。

# 並列呼び出し時もまとめて返す(上のコードがそのまま使える)
messages.append({
    "role": "user",
    "content": [
        {"type": "tool_result", "tool_use_id": "id_1", "content": "..."},
        {"type": "tool_result", "tool_use_id": "id_2", "content": "..."},
    ]
})

正直、最初はこの仕様に気づかず「1件ずつ返せばいいんでしょ」と思っていたら、並列で呼ばれたときに壊れました。ちゃんとまとめて返すこと。

なお、並列を抑制したい場合は tool_choicedisable_parallel_tool_use を付けて止められます(後述)。

実用的なラッパーを作る

毎回ループを手書きするのはしんどいので、ツール定義と実装関数を紐付けたシンプルなラッパーを作っておくと楽です。

from typing import Callable

class ToolRegistry:
    def __init__(self):
        self._tools: list[dict] = []
        self._handlers: dict[str, Callable] = {}

    def register(self, schema: dict, handler: Callable):
        self._tools.append(schema)
        self._handlers[schema["name"]] = handler

    @property
    def schemas(self) -> list[dict]:
        return self._tools

    def run(self, name: str, inputs: dict):
        if name not in self._handlers:
            raise ValueError(f"Tool not found: {name}")
        return self._handlers[name](**inputs)


# 使い方
registry = ToolRegistry()

registry.register(
    schema={
        "name": "get_weather",
        "description": "指定都市の天気を取得する",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    handler=get_weather,
)

# client.messages.create に渡す tools は registry.schemas

ツールが増えてくると if name == "xxx" が大量発生するので、このくらいは整備しておいた方が精神的に楽です。

エラーハンドリングと注意点

max_tokensの設定

ツールを使うターンではClaudeの出力に加えてツール呼び出しの情報も含まれるため、max_tokens が足りなくなりやすいです。

「いくつ以上が目安」と断定するのはちょい難しくて、タスクとモデル次第で必要量が変わります。個人的には「まずは大きめで始めて、stop_reason == "max_tokens" が出るようなら上げる」くらいの調整が安全かなと。余談ですが、出力を盛りがちなプロンプトを書くと気づいたらすぐ消費してます。

tool_choice による関数呼び出しの制御

デフォルトだと「使うかどうかはClaudeが判断する」モード(type: "auto")ですが、明示的に制御することもできます:

# Claudeに任せる(推奨のデフォルト)
tool_choice={"type": "auto"}

# なるべくツールを使わせる(any)
tool_choice={"type": "any"}

# 特定のツールを強制
tool_choice={"type": "tool", "name": "get_weather"}

# ツールを使わせない(テキストのみ)
tool_choice={"type": "none"}

さらに、並列ツール呼び出しを抑制したい場合は disable_parallel_tool_use を付けられます。

# 並列を無効化(最大1ツールに抑える)
tool_choice={"type": "auto", "disable_parallel_tool_use": True}

テスト時や、ツールを絶対に呼んでほしい場面では "any" を使うと便利です。

無限ループ対策

まれにツールが繰り返し呼ばれ続けるケースがあります。ループ回数の上限を設けておくと安心です。

MAX_TURNS = 10

turn = 0
while turn < MAX_TURNS:
    turn += 1
    response = client.messages.create(...)

    if response.stop_reason == "end_turn":
        break
    if response.stop_reason != "tool_use":
        break
    # ツール処理...

else:
    raise RuntimeError("Tool loop exceeded max turns")

本番で動かすものには入れておいた方がいいです。実際に止まらなくなったことがあって、Claudeも悪くないんですが入力構造が微妙だったみたいで……。

※この記事にはプロモーションが含まれます

ちなみに、PLAUD NOTE(AIボイスレコーダー。録音から文字起こし・要約まで自動で行える)も気になっています。PLAUD NOTE

まとめ

Tool UseはClaude APIの中でも使い勝手の幅が広い機能で、「外部システムと連携するエージェント」を作るときには避けて通れない仕組みです。今回の内容を整理するとこんな感じです。

  • ツール定義は description をしっかり書くと精度が上がる
  • 呼び出しループは stop_reason で分岐し、response.content をそのまま履歴に追加する
  • 並列呼び出しを想定して tool_result は配列でまとめて返す
  • 本番コードには無限ループ対策を必ず入れる

基本フローを押さえて、エラーハンドリングと無限ループ対策を入れれば、そこそこ実用的なものが作れると思います。個人的にはエージェント系の実装が一番楽しいので、ぜひ試してみてください。

タイトルとURLをコピーしました