消息工具

工具运行器（SDK）

使用 SDK 的工具运行器自动处理代理循环、错误包装和类型安全。

工具运行器会处理代理循环、错误包装和类型安全，因此您无需自行处理。当您需要人工介入审批、自定义日志记录或条件执行时，请改用手动循环。

工具运行器会自动完成以下工作，而无需您手动处理工具调用、工具结果和对话管理：

在 Claude 调用工具时运行工具
处理请求/响应周期
管理对话状态
提供类型安全和验证

基本用法

使用 SDK 辅助工具定义工具，然后使用工具运行器来运行它们。

根据 SDK 的工具签名，工具会以字符串或内容块（文本、图像或文档块）的形式返回其结果，因此工具可以返回多模态结果。返回的字符串会成为单个文本内容块。要返回结构化数据（例如 JSON 对象或数字），请先将其编码为字符串。

使用 @beta_tool 装饰器通过类型提示和文档字符串来定义工具。

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)

@beta_tool 装饰器会检查函数参数和文档字符串，为您推导出 JSON schema。

迭代工具运行器

工具运行器是一个可迭代对象，会产出来自 Claude 的消息。在每次迭代中，运行器会检查 Claude 是否请求了工具使用。如果是，它会运行该工具并自动将结果发送回 Claude，然后产出来自 Claude 的下一条消息以继续您的循环。

您可以在任何迭代中使用 break 语句结束循环。运行器会一直循环，直到 Claude 返回不包含工具使用的消息，或者在您设置了 max_iterations 的情况下达到该上限。

如果您不需要中间消息，可以直接获取最终消息：

使用 runner.until_done() 获取最终消息。

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)

高级用法

在循环内部，您可以读取每条响应消息，并在下一次 API 调用之前修改运行器的状态。每次迭代遵循以下生命周期：

运行器使用其当前状态向 Messages API 发送请求。
运行器将响应消息产出给您的循环体。
您的循环体运行。您可以读取该消息，并可选择修改运行器的状态。
当您的循环体返回时，运行器会检查您是否修改了其消息历史。
- 如果您没有修改消息历史： 如果消息包含工具调用，运行器会追加助手消息和工具结果，然后继续。如果没有工具调用，循环退出。
- 如果您修改了消息历史： 运行器会跳过其自动追加，并原样使用您的状态。请参阅接管消息历史。

接管消息历史

默认情况下，运行器会为您管理对话状态：在每个工具调用轮次之后，它会将助手消息和任何工具结果追加到其自己的消息历史中。当您想要重试某个轮次（丢弃响应并重新发送）、注入后续消息或自行构建工具结果时，您可以接管消息历史。

您可以通过在循环体内部修改运行器的消息来接管。具体方法取决于 SDK。请参阅后面的各语言标签页。

当您在某次迭代中接管时，运行器不会追加该轮次的助手消息或工具结果。您需要负责保持对话的有效性：自行追加助手消息和工具结果（如果您希望该轮次生效），有条件地修改状态以便在没有工具调用时循环仍能退出，并传入 max_iterations 来限制循环。所有七个 SDK 都支持 max_iterations。

使用 generate_tool_call_response() 检查或计算工具结果。在循环内部调用 append_messages() 会告知运行器您正在自行管理历史，因此请在您追加的内容中包含助手消息和工具结果。

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 会跳过
        # 本次迭代的自动追加。您需要自行追加助手消息和
        # 工具结果，以及任何后续内容。
        runner.append_messages(
            message,
            tool_response,
            {"role": "user", "content": "Please be concise."},
        )
    # 当没有工具调用时，保持状态不变，以便循环退出。

要在不接管消息历史的情况下更改请求参数（例如 max_tokens），请使用 set_messages_params()。运行器仍会自动追加助手消息和工具结果。

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

自动上下文管理

对于长时间运行的代理任务，Python、TypeScript 和 Ruby 工具运行器支持自动压缩，当令牌使用量超过阈值时会生成摘要，使对话能够超越上下文窗口限制继续进行。这三个 SDK 都已弃用此客户端选项，转而推荐服务器端的上下文编辑，后者在每个 SDK 中都可用。Go、Java、C# 和 PHP 工具运行器不包含客户端压缩。

调试工具执行

当工具抛出异常时，工具运行器会捕获它，并将错误作为带有 is_error: true 的工具结果返回给 Claude。工具结果携带异常的消息（在 Python 中为其类型和消息），而不是完整的堆栈跟踪。

SDK 记录的内容因语言而异。每当工具引发未处理的异常时，Python SDK 会通过标准 logging 模块记录完整的异常，包括其堆栈跟踪。Python、TypeScript 和 Java SDK 会读取 ANTHROPIC_LOG 环境变量来开启 SDK 的日志记录，其中包括请求和响应的详细信息：

# 以 info 级别记录日志
export ANTHROPIC_LOG=info

# 以 debug 级别记录日志以获得更详细的输出
export ANTHROPIC_LOG=debug

Go、Ruby、C# 和 PHP SDK 不读取 ANTHROPIC_LOG。除 Python 外，没有 SDK 会记录失败的工具：要查看工具失败的原因，请在工具函数内部捕获并记录异常，然后再返回或重新抛出它。

拦截工具错误

默认情况下，工具错误会被传回给 Claude，然后 Claude 可以做出相应的响应。但是，您可能希望检测错误并以不同方式处理它们，例如提前停止执行或实现自定义错误处理。

在 Python 和 TypeScript SDK 中，使用工具响应方法（Python 中的 generate_tool_call_response()，TypeScript 中的 generateToolResponse()）来拦截工具结果，并在它们被发送给 Claude 之前检查错误。其他 SDK 不公开该钩子。它们的标签页描述了最接近的替代方案：

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 是一个字典：{"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)

修改工具结果

您可以在工具结果被发送回 Claude 之前修改它们。这对于添加元数据（例如 cache_control 以在工具结果上启用提示缓存）或转换工具输出非常有用。

在 Python 和 TypeScript SDK 中，使用工具响应方法获取工具结果，然后在运行器继续之前修改它。是显式追加修改后的结果还是就地变更它，取决于 SDK。请参阅每个标签页中的代码注释。

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 是一个字典：{"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)

流式传输

启用流式传输以增量处理每个轮次的响应。每次迭代会产出一个流对象，您可以迭代该对象以获取事件。

设置 stream=True 并使用 get_final_message() 获取累积的消息。

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())

后续步骤

严格工具使用

通过语法约束采样对 Claude 的工具输入强制执行 JSON Schema 合规性。

处理工具调用

解析 tool_use 块、格式化 tool_result 响应，并使用 is_error 处理错误。

并行工具使用

启用、格式化和禁用并行工具调用，并提供消息历史指导和故障排除。

定义工具

指定工具 schema、编写有效的描述，并控制 Claude 何时调用您的工具。

Was this page helpful?

消息工具

工具运行器（SDK）

使用 SDK 的工具运行器自动处理代理循环、错误包装和类型安全。

工具运行器会处理代理循环、错误包装和类型安全，因此您无需自行处理。当您需要人工介入审批、自定义日志记录或条件执行时，请改用手动循环。

工具运行器会自动完成以下工作，而无需您手动处理工具调用、工具结果和对话管理：

在 Claude 调用工具时运行工具
处理请求/响应周期
管理对话状态
提供类型安全和验证

基本用法

使用 SDK 辅助工具定义工具，然后使用工具运行器来运行它们。

使用 @beta_tool 装饰器通过类型提示和文档字符串来定义工具。

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)

@beta_tool 装饰器会检查函数参数和文档字符串，为您推导出 JSON schema。

迭代工具运行器

如果您不需要中间消息，可以直接获取最终消息：

使用 runner.until_done() 获取最终消息。

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)

高级用法

在循环内部，您可以读取每条响应消息，并在下一次 API 调用之前修改运行器的状态。每次迭代遵循以下生命周期：

运行器使用其当前状态向 Messages API 发送请求。
运行器将响应消息产出给您的循环体。
您的循环体运行。您可以读取该消息，并可选择修改运行器的状态。
当您的循环体返回时，运行器会检查您是否修改了其消息历史。
- 如果您没有修改消息历史： 如果消息包含工具调用，运行器会追加助手消息和工具结果，然后继续。如果没有工具调用，循环退出。
- 如果您修改了消息历史： 运行器会跳过其自动追加，并原样使用您的状态。请参阅接管消息历史。

接管消息历史

您可以通过在循环体内部修改运行器的消息来接管。具体方法取决于 SDK。请参阅后面的各语言标签页。

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 会跳过
        # 本次迭代的自动追加。您需要自行追加助手消息和
        # 工具结果，以及任何后续内容。
        runner.append_messages(
            message,
            tool_response,
            {"role": "user", "content": "Please be concise."},
        )
    # 当没有工具调用时，保持状态不变，以便循环退出。

要在不接管消息历史的情况下更改请求参数（例如 max_tokens），请使用 set_messages_params()。运行器仍会自动追加助手消息和工具结果。

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

自动上下文管理

调试工具执行

# 以 info 级别记录日志
export ANTHROPIC_LOG=info

# 以 debug 级别记录日志以获得更详细的输出
export ANTHROPIC_LOG=debug

拦截工具错误

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 是一个字典：{"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)

修改工具结果

您可以在工具结果被发送回 Claude 之前修改它们。这对于添加元数据（例如 cache_control 以在工具结果上启用提示缓存）或转换工具输出非常有用。

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 是一个字典：{"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)

流式传输

启用流式传输以增量处理每个轮次的响应。每次迭代会产出一个流对象，您可以迭代该对象以获取事件。

设置 stream=True 并使用 get_final_message() 获取累积的消息。

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())

后续步骤

严格工具使用

通过语法约束采样对 Claude 的工具输入强制执行 JSON Schema 合规性。

处理工具调用

解析 tool_use 块、格式化 tool_result 响应，并使用 is_error 处理错误。

并行工具使用

启用、格式化和禁用并行工具调用，并提供消息历史指导和故障排除。

定义工具

指定工具 schema、编写有效的描述，并控制 Claude 何时调用您的工具。

Was this page helpful?

基本用法

迭代工具运行器

高级用法

接管消息历史

自动上下文管理

调试工具执行

拦截工具错误

修改工具结果

流式传输

后续步骤

基本用法

迭代工具运行器

高级用法

接管消息历史

自动上下文管理

调试工具执行

拦截工具错误

修改工具结果

流式传输

后续步骤

基本用法

迭代工具运行器

高级用法

接管消息历史

自动上下文管理

调试工具执行

拦截工具错误

修改工具结果

流式传输

后续步骤

基本用法

迭代工具运行器

高级用法

接管消息历史

自动上下文管理

调试工具执行

拦截工具错误

修改工具结果

流式传输

后续步骤