Messages工具基础设施

细粒度工具流式传输

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

"Fine-grained tool streaming"（细粒度工具流式传输）会在 Claude 生成工具输入的同时将其传送到您的客户端，而无需服务器端缓冲或 JSON 验证。跳过缓冲步骤可以缩短大型参数（例如文档或代码块）首个片段的到达时间，并且这些片段通过与标准工具使用相同的流式传输消息事件到达。

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

所有模型均在 Claude API、Amazon Bedrock、Claude Platform on AWS、Google Cloud 和 Microsoft Foundry 上支持细粒度工具流式传输。要使用它，请在您希望启用细粒度流式传输的任何用户定义工具上将 eager_input_streaming 设置为 true，并在您的请求中启用流式传输。

eager_input_streaming 字段是可选的。将其设置为 true 会为该工具开启细粒度流式传输，而省略它则会使用标准的缓冲流式传输，即 API 在将每个参数值流式传回之前对其进行缓冲和验证。例外情况是仍然发送旧版 fine-grained-tool-streaming-2025-05-14 beta 标头的请求，该标头会为未设置此字段的工具开启细粒度流式传输。按工具设置的字段取代了该标头，并且即使请求仍然发送该标头，显式的 false 也会使该工具保持缓冲流式传输。旧版标头不能与计算机使用或浏览器使用工具集条目组合使用：API 会拒绝同时发送两者的请求，因此请移除该标头，并在需要它的用户定义工具上设置 eager_input_streaming。有关字段定义，请参阅工具参考。

以下示例为 make_file 工具开启细粒度流式传输，并要求 Claude 写一首长诗，这样工具输入就足够大，可以观察它流式传入：

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-5",
    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:
    for event in stream:
        if event.type == "input_json":
            print(event.partial_json, end="", flush=True)
    final_message = stream.get_final_message()

print()
for block in final_message.content:
    if block.type == "tool_use":
        print(f"Complete tool input: {block.input}")

每个标签页都为 make_file 工具开启了细粒度流式传输。SDK 标签页会在每个输入片段到达的那一刻将其打印出来，然后在流结束后打印完整的累积输入。cURL 标签页显示原始事件流，CLI 标签页使用 jq 仅打印片段。由于打印出的片段会拼接成完整的工具输入，这首诗会随着 Claude 的书写填满您的终端：

{"filename": "poem.txt", "lines_of_text": ["The Wanderer's Journey", "", "I.", "", "Beneath the vast and star-strewn sky,", "Where silver moonbeams softly lie,", ...
Complete tool input: {"filename": "poem.txt", "lines_of_text": ["The Wanderer's Journey", ...]}

如果没有 eager_input_streaming，API 会在将每个参数值流式传回之前对其进行缓冲和验证，因此对于大型参数，在 Claude 完成生成之前不会打印任何内容。启用它后，片段会在 Claude 开始生成该参数时立即开始到达，并且它们通常更长，单词中途断开的情况更少。

累积工具输入增量

累积约定与标准工具使用流式传输相同，因此本节在使用和不使用 eager_input_streaming 的情况下均适用。有关事件格式，请参阅流式传输消息中的输入 JSON 增量。细粒度工具流式传输改变的是您可以对结果做出的假设：服务器在不验证片段的情况下流式传输它们，因此累积的字符串可能不是有效的 JSON。

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

如果您的 SDK 提供了累积器辅助工具（如上一个示例中的 Python、TypeScript、Go、Java 和 Ruby 标签页那样），它会为您处理这一切。手动模式适用于没有辅助工具的 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 时，解析累积的字符串

请像以下 SDK 示例那样对解析进行防护。响应也可能在参数中途因 max_tokens 而停止。请检查停止原因，并决定是使用更高的 max_tokens 重试请求，还是修复不完整的输入。

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

client = anthropic.Anthropic()

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

with client.messages.stream(
    model="claude-opus-5",
    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:
                raw_input = tool_inputs[event.index]
                try:
                    parsed = json.loads(raw_input)
                except json.JSONDecodeError:
                    # 累积的字符串不保证是有效的 JSON。
                    # 请参阅本页的“处理工具响应中的无效 JSON”部分。
                    print(f"Invalid tool input: {raw_input}")
                else:
                    print(f"Tool input: {parsed}")

处理工具响应中的无效 JSON

使用细粒度工具流式传输时，工具调用的累积输入可能是无效或不完整的 JSON。在这种情况下，您无法运行该工具，因此应将失败情况报告给 Claude。工具结果的 content 不必是 JSON，但将原始字符串包装在一个单键 JSON 对象中，可以让 Claude 明确知道您收到了无效的 JSON，并保留原始输入以便调试：

{
  "INVALID_JSON": "<the unparseable input you received>"
}

将该包装对象序列化为字符串，作为工具结果内容块的 content 返回，并将 is_error 设置为 true：

{
  "type": "tool_result",
  "tool_use_id": "toolu_01A09q90qw90lq917835lq9",
  "is_error": true,
  "content": "{\"INVALID_JSON\": \"<the unparseable input you received>\"}"
}