細粒度ツールストリーミング

This feature is eligible for Zero Data Retention (ZDR). When your organization has a ZDR arrangement, data sent through this feature is not stored after the API response is returned.

細粒度ツールストリーミングは、すべてのモデルとすべてのプラットフォームで一般利用可能です。これにより、ツール使用パラメータ値のストリーミングがバッファリングやJSON検証なしで可能になり、大きなパラメータの受信を開始するまでのレイテンシが削減されます。

細粒度ツールストリーミングを使用する場合、無効または部分的なJSON入力を受け取る可能性があります。コード内でこれらのエッジケースに対応するようにしてください。

細粒度ツールストリーミングの使用方法

細粒度ツールストリーミングは、すべてのモデルとすべてのプラットフォーム（Claude API、Amazon Bedrock、Google Vertex AI、Microsoft Foundry）で利用可能です。これを使用するには、細粒度ストリーミングを有効にしたいユーザー定義ツールでeager_input_streamingをtrueに設定し、リクエストでストリーミングを有効にします。

APIで細粒度ツールストリーミングを使用する方法の例を次に示します。

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-7",
    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:
        pass
    final_message = stream.get_final_message()

print(final_message.usage)

この例では、細粒度ツールストリーミングにより、Claudeは長い詩の行をツール呼び出しmake_fileにストリーミングでき、lines_of_textパラメータが有効なJSONであるかを検証するためにバッファリングする必要がありません。これは、パラメータ全体がバッファリングされて検証されるのを待つことなく、パラメータがストリーミングされるのをリアルタイムで見ることができることを意味します。

細粒度ツールストリーミングでは、ツール使用チャンクがより速くストリーミング開始され、多くの場合、より長く、単語の区切りが少なくなります。これはチャンキング動作の違いによるものです。

例：

細粒度ストリーミングなし（15秒の遅延）:

Chunk 1: '{"'
Chunk 2: 'query": "Ty'
Chunk 3: 'peScri'
Chunk 4: 'pt 5.0 5.1 '
Chunk 5: '5.2 5'
Chunk 6: '.3'
Chunk 8: ' new f'
Chunk 9: 'eatur'
...

細粒度ストリーミングあり（3秒の遅延）:

Chunk 1: '{"query": "TypeScript 5.0 5.1 5.2 5.3'
Chunk 2: ' new features comparison'

細粒度ストリーミングはパラメータをバッファリングやJSON検証なしで送信するため、結果のストリームが有効なJSON文字列で完了することは保証されません。特に、停止理由max_tokensに達した場合、ストリームはパラメータの途中で終了し、不完全な可能性があります。通常、max_tokensに達した場合を処理するための特定のサポートを記述する必要があります。

ツール入力デルタの蓄積

tool_useコンテンツブロックがストリーミングされるとき、初期のcontent_block_startイベントにはinput: {}（空のオブジェクト）が含まれます。これはプレースホルダーです。実際の入力は一連のinput_json_deltaイベントとして到着し、各イベントはpartial_json文字列フラグメントを含みます。コードはこれらのフラグメントを連結し、ブロックが閉じたら結果を解析する必要があります。

蓄積契約：

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（文字列）の間の型の不一致は設計によるものです。空のオブジェクトはコンテンツ配列のスロットをマークします。デルタ文字列は実際の値を構築します。

PythonおよびTypeScript SDKは、この蓄積を自動的に実行する高レベルのストリームヘルパー（stream.get_final_message()、stream.finalMessage()）を提供します。ブロックが閉じる前に部分的な入力に反応する必要がある場合（進捗インジケータのレンダリングや下流リクエストの早期開始など）にのみ、上記の手動パターンを使用してください。

ツール応答での無効なJSONの処理

細粒度ツールストリーミングを使用する場合、モデルから無効または不完全なJSONを受け取る可能性があります。この無効なJSONをエラー応答ブロック内でモデルに返す必要がある場合、適切な処理を確保するためにJSONオブジェクトでラップできます（合理的なキーを使用）。例えば：

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

このアプローチは、モデルがコンテンツが無効なJSONであることを理解するのに役立ちながら、デバッグ目的で元の形式が正しくないデータを保持します。

無効なJSONをラップする場合、ラッパーオブジェクト内の有効なJSON構造を維持するために、無効なJSON文字列内のクォートまたは特殊文字を適切にエスケープしてください。

次のステップ

メッセージのストリーミング

サーバー送信イベントとストリームイベントタイプの完全なリファレンス。

ツール呼び出しの処理

ツールを実行し、必要なメッセージ形式で結果を返します。

ツールリファレンス

Anthropicスキーマツールとそのバージョン文字列の完全なディレクトリ。

import json
import anthropic

client = anthropic.Anthropic()

tool_inputs = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-7",
    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:
        if (
            event.type == "content_block_start"
            and event.content_block.type == "tool_use"
        ):
            tool_inputs[event.index] = ""
        elif (
            event.type == "content_block_delta"
            and event.delta.type == "input_json_delta"
        ):
            tool_inputs[event.index] += event.delta.partial_json
        elif event.type == "content_block_stop" and event.index in tool_inputs:
            parsed = json.loads(tool_inputs[event.index])
            print(f"Tool input: {parsed}")