メッセージツールインフラストラクチャ

きめ細かいツールストリーミング

レイテンシが重要なアプリケーション向けに、サーバー側でのJSONバッファリングなしでツール入力をストリーミングします。

この機能はZero Data Retention（ZDR）の対象です。組織がZDR契約を締結している場合、この機能を通じて送信されたデータは、APIレスポンスが返された後に保存されることはありません。

きめ細かいツールストリーミングは、すべてのモデルおよびすべてのプラットフォームで利用可能です。これにより、バッファリングやJSON検証なしでツール使用パラメータ値のストリーミングが可能になり、大きなパラメータの受信開始までの「latency」（レイテンシ）が短縮されます。

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

きめ細かいツールストリーミングの使用方法

きめ細かいツールストリーミングは、Claude API、Claude Platform on AWS、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はlines_of_textパラメータが有効なJSONかどうかを検証するためのバッファリングを行わずに、長い詩の行をツール呼び出しmake_fileにストリーミングできます。つまり、パラメータ全体がバッファリングされて検証されるのを待つことなく、パラメータが到着するたびにストリームを確認できます。

きめ細かいツールストリーミングでは、サーバーが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バッファリングなしでツール入力をストリーミングします。

きめ細かいツールストリーミングの使用方法

以下は、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)

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>"
}

次のステップ

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

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

ツール呼び出しの処理

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

ツールリファレンス

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

Was this page helpful?

きめ細かいツールストリーミングの使用方法

ツール入力デルタの蓄積

ツールレスポンスでの無効なJSONの処理

次のステップ

きめ細かいツールストリーミングの使用方法

ツール入力デルタの蓄積

ツールレスポンスでの無効なJSONの処理

次のステップ

きめ細かいツールストリーミングの使用方法

ツール入力デルタの蓄積

ツールレスポンスでの無効なJSONの処理

次のステップ

きめ細かいツールストリーミングの使用方法

ツール入力デルタの蓄積

ツールレスポンスでの無効なJSONの処理

次のステップ