工具執行器（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() 會將狀態標記為已修改，因此 runner 會跳過
        # 本次迭代的自動附加。請自行附加 assistant 訊息與
        # 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,
)

# 使用串流時，runner 會回傳 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())