도구 러너 (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-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-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-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-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-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-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())