消息工具基础设施

细粒度工具流式传输

为对延迟敏感的应用程序流式传输工具输入，无需服务器端 JSON 缓冲。

此功能符合零数据保留（ZDR）的条件。当您的组织签订了 ZDR 协议时，通过此功能发送的数据在 API 响应返回后不会被存储。

"Fine-grained tool streaming"（细粒度工具流式传输）在所有模型和所有平台上均可用。它支持对工具使用参数值进行流式传输，无需缓冲或 JSON 验证，从而减少开始接收大型参数时的延迟。

使用细粒度工具流式传输时，您可能会收到无效或不完整的 JSON 输入。请确保在您的代码中处理这些边缘情况。

如何使用细粒度工具流式传输

细粒度工具流式传输在 Claude API、AWS 上的 Claude Platform、Amazon Bedrock、Vertex AI 和 Microsoft Foundry 上均受支持。要使用此功能，请在您希望启用细粒度流式传输的任何用户定义工具上将 eager_input_streaming 设置为 true，并在您的请求中启用流式传输。

以下是如何通过 API 使用细粒度工具流式传输的示例：

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-8",
    tools=[
        {
            "name": "make_file",
            "description": "Write text to a file",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "The filename to write text to",
                    },
                    "lines_of_text": {
                        "type": "array",
                        "description": "An array of lines of text to write to the file",
                    },
                },
                "required": ["filename", "lines_of_text"],
            },
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Can you write a long poem and make a file called poem.txt?",
        }
    ],
) as stream:
    final_message = stream.get_final_message()

print(f"Input tokens: {final_message.usage.input_tokens}")
print(f"Output tokens: {final_message.usage.output_tokens}")

在此示例中，细粒度工具流式传输使 Claude 能够将一首长诗的各行流式传输到工具调用 make_file 中，而无需缓冲以验证 lines_of_text 参数是否为有效的 JSON。这意味着您可以在参数到达时实时看到其流式传输，而无需等待整个参数完成缓冲和验证。

使用细粒度工具流式传输时，工具输入数据块会更快开始到达，因为服务器跳过了 JSON 验证缓冲。作为副作用，数据块通常更长，且包含更少的令牌中间断点。

由于细粒度流式传输在发送参数时不进行缓冲或 JSON 验证，因此无法保证生成的流会以有效的 JSON 字符串结束。特别是，如果达到了停止原因 max_tokens，流可能会在参数中途结束，导致参数不完整。您通常需要编写特定的支持代码来处理达到 max_tokens 的情况。

累积工具输入增量

当 tool_use 内容块进行流式传输时，初始的 content_block_start 事件包含 input: {}（一个空对象）。这是一个占位符。实际输入以一系列 input_json_delta 事件的形式到达，每个事件携带一个 partial_json 字符串片段。要组装完整的输入，请拼接这些片段，并在块关闭时解析结果。

如果您的 SDK 提供了累加器辅助工具（如本页第一个示例中所使用的），它会为您处理这些操作。手动模式适用于没有辅助工具的 SDK，或者当您需要在块关闭之前对部分输入做出响应时。

累积约定如下：

在收到 type: "tool_use" 的 content_block_start 时，初始化一个空字符串：input_json = ""
对于每个 type: "input_json_delta" 的 content_block_delta，进行追加：input_json += event.delta.partial_json
在收到 content_block_stop 时，解析累积的字符串：json.loads(input_json)

初始的 input: {}（对象）与 partial_json（字符串）之间的类型不匹配是有意设计的。空对象标记了内容数组中的位置；增量字符串则构建实际的值。

client = anthropic.Anthropic()

tool_inputs: dict[int, str] = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=1024,
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
    for event in stream:
        match event.type:
            case "content_block_start" if event.content_block.type == "tool_use":
                tool_inputs[event.index] = ""
            case "content_block_delta" if event.delta.type == "input_json_delta":
                tool_inputs[event.index] += event.delta.partial_json
            case "content_block_stop" if event.index in tool_inputs:
                parsed = json.loads(tool_inputs[event.index])
                print(f"Tool input: {parsed}")

当您需要在块关闭之前对部分输入做出响应时（例如，渲染进度指示器），请使用手动模式。否则，请优先使用您的 SDK 的累加器辅助工具（本页第一个示例中使用了该工具）。

处理工具响应中的无效 JSON

使用细粒度工具流式传输时，您可能会从模型收到无效或不完整的 JSON。如果您需要在错误响应块中将此无效 JSON 传回给模型，可以将其包装在一个 JSON 对象中（使用合理的键名）以确保正确处理。例如：

{
  "INVALID_JSON": "<your invalid json string>"
}

这种方法有助于模型理解该内容是无效的 JSON，同时保留原始的格式错误数据以供调试之用。

包装无效 JSON 时，请确保正确转义无效 JSON 字符串中的任何引号或特殊字符，以保持包装对象中的 JSON 结构有效。

后续步骤

流式传输消息

服务器发送事件和流事件类型的完整参考。

处理工具调用

执行工具并以所需的消息格式返回结果。

工具参考

Anthropic 架构工具及其版本字符串的完整目录。

Was this page helpful?

消息工具基础设施

细粒度工具流式传输

为对延迟敏感的应用程序流式传输工具输入，无需服务器端 JSON 缓冲。

此功能符合零数据保留（ZDR）的条件。当您的组织签订了 ZDR 协议时，通过此功能发送的数据在 API 响应返回后不会被存储。

使用细粒度工具流式传输时，您可能会收到无效或不完整的 JSON 输入。请确保在您的代码中处理这些边缘情况。

如何使用细粒度工具流式传输

以下是如何通过 API 使用细粒度工具流式传输的示例：

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-8",
    tools=[
        {
            "name": "make_file",
            "description": "Write text to a file",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "The filename to write text to",
                    },
                    "lines_of_text": {
                        "type": "array",
                        "description": "An array of lines of text to write to the file",
                    },
                },
                "required": ["filename", "lines_of_text"],
            },
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Can you write a long poem and make a file called poem.txt?",
        }
    ],
) as stream:
    final_message = stream.get_final_message()

print(f"Input tokens: {final_message.usage.input_tokens}")
print(f"Output tokens: {final_message.usage.output_tokens}")

累积工具输入增量

累积约定如下：

在收到 type: "tool_use" 的 content_block_start 时，初始化一个空字符串：input_json = ""
对于每个 type: "input_json_delta" 的 content_block_delta，进行追加：input_json += event.delta.partial_json
在收到 content_block_stop 时，解析累积的字符串：json.loads(input_json)

初始的 input: {}（对象）与 partial_json（字符串）之间的类型不匹配是有意设计的。空对象标记了内容数组中的位置；增量字符串则构建实际的值。

client = anthropic.Anthropic()

tool_inputs: dict[int, str] = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=1024,
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
    for event in stream:
        match event.type:
            case "content_block_start" if event.content_block.type == "tool_use":
                tool_inputs[event.index] = ""
            case "content_block_delta" if event.delta.type == "input_json_delta":
                tool_inputs[event.index] += event.delta.partial_json
            case "content_block_stop" if event.index in tool_inputs:
                parsed = json.loads(tool_inputs[event.index])
                print(f"Tool input: {parsed}")

处理工具响应中的无效 JSON

{
  "INVALID_JSON": "<your invalid json string>"
}

这种方法有助于模型理解该内容是无效的 JSON，同时保留原始的格式错误数据以供调试之用。

包装无效 JSON 时，请确保正确转义无效 JSON 字符串中的任何引号或特殊字符，以保持包装对象中的 JSON 结构有效。

后续步骤

流式传输消息

服务器发送事件和流事件类型的完整参考。

处理工具调用

执行工具并以所需的消息格式返回结果。

工具参考

Anthropic 架构工具及其版本字符串的完整目录。

Was this page helpful?

如何使用细粒度工具流式传输

累积工具输入增量

处理工具响应中的无效 JSON

后续步骤

如何使用细粒度工具流式传输

累积工具输入增量

处理工具响应中的无效 JSON

后续步骤

如何使用细粒度工具流式传输

累积工具输入增量

处理工具响应中的无效 JSON

后续步骤

如何使用细粒度工具流式传输

累积工具输入增量

处理工具响应中的无效 JSON

后续步骤