ツールランナー（SDK）

import json
from anthropic import Anthropic, beta_tool

client = Anthropic()


@beta_tool
def get_weather(location: str, unit: str = "fahrenheit") -> str:
    """Get the current weather in a given location.

    Args:
        location: The city and state, e.g. San Francisco, CA
        unit: Temperature unit, either 'celsius' or 'fahrenheit'
    """
    return json.dumps({"temperature": "20°C", "condition": "Sunny"})


@beta_tool
def calculate_sum(a: int, b: int) -> str:
    """Add two numbers together.

    Args:
        a: First number
        b: Second number
    """
    return str(a + b)


runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[get_weather, calculate_sum],
    messages=[
        {
            "role": "user",
            "content": "What's the weather like in Paris? Also, what's 15 + 27?",
        }
    ],
)
for message in runner:
    print(message)

client = anthropic.Anthropic()
# ...
runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[get_weather, calculate_sum],
    messages=[
        {
            "role": "user",
            "content": "What's the weather like in Paris? Also, what's 15 + 27?",
        }
    ],
)
final_message = runner.until_done()
for block in final_message.content:
    if block.type == "text":
        print(block.text)

runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    max_iterations=10,
    tools=[get_weather],
    messages=[{"role": "user", "content": "What's the weather in San Francisco?"}],
)

for message in runner:
    tool_response = runner.generate_tool_call_response()
    if tool_response is not None:
        # append_messages() は状態を変更済みとしてフラグを立てるため、ランナーはこの
        # イテレーションでの自動追加をスキップします。アシスタントメッセージと
        # tool result、および任意のフォローアップを自分で追加してください。
        runner.append_messages(
            message,
            tool_response,
            {"role": "user", "content": "Please be concise."},
        )
    # ツール呼び出しがない場合は、ループが終了するように状態をそのままにします。

for message in runner:
    runner.set_messages_params(lambda params: {**params, "max_tokens": 2048})

client = anthropic.Anthropic()
# ...
runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[my_tool],
    messages=[{"role": "user", "content": "Run my_tool with the query 'hello'."}],
)

for message in runner:
    tool_response = runner.generate_tool_call_response()

    if tool_response is not None:
        # tool_responseはdictです: {"role": "user", "content": [...]}
        # いずれかのツール結果にエラーがあるか確認
        for block in tool_response["content"]:
            if block.get("is_error"):
                # オプション1: 例外を発生させてループを停止
                raise RuntimeError(f"Tool failed: {json.dumps(block['content'])}")

                # オプション2: ログに記録して続行（Claudeに処理を任せる）
                # logger.error(f"Tool error: {json.dumps(block['content'])}")

    # メッセージを通常どおり処理
    print(message.content)

client = anthropic.Anthropic()
# ...
runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[search_documents],
    messages=[
        {
            "role": "user",
            "content": "Search for information about the climate of San Francisco",
        }
    ],
)

for message in runner:
    tool_response = runner.generate_tool_call_response()

    if tool_response is not None:
        # tool_responseはdict形式です: {"role": "user", "content": [...]}
        # ツール結果を変更してキャッシュ制御を追加します
        for block in tool_response["content"]:
            if block["type"] == "tool_result":
                # cache_controlを追加してこのツール結果をキャッシュします
                block["cache_control"] = {"type": "ephemeral"}

        # 変更したレスポンスを追加します（これにより元のレスポンスの自動追加が防止されます）
        runner.append_messages(message, tool_response)

    print(message.content)

client = anthropic.Anthropic()
# ...
runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[calculate_sum],
    messages=[{"role": "user", "content": "What is 15 + 27?"}],
    stream=True,
)

# ストリーミング時、ランナーはBetaMessageStreamを返します
for message_stream in runner:
    for event in message_stream:
        print("event:", event)
    print("message:", message_stream.get_final_message())

print(runner.until_done())