streaming

ストリーミングスキル

Server-Sent Events（SSE）を使用して、Claude API からのリアルタイムストリーミング応答を実装します。

このスキルを使用する場面

• リアルタイムのユーザーインターフェース
• 生成に時間がかかる処理
• 出力を段階的に表示する場合
• ストリーミング中のツール使用
• 拡張思考の可視化

SSE イベントの流れ

message_start
    → content_block_start
        → content_block_delta (repeated)
    → content_block_stop
    → (more blocks...)
→ message_delta
→ message_stop

基本的な実装

基本的なテキストストリーミング（Python）

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a short story."}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

イベントベースのストリーミング

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="")
            elif event.delta.type == "input_json_delta":
                # Tool input (accumulate, don't parse yet!)
                tool_input_buffer += event.delta.partial_json
        elif event.type == "content_block_stop":
            # Now safe to parse tool input
            if tool_input_buffer:
                tool_input = json.loads(tool_input_buffer)

TypeScript のストリーミング

import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();

const stream = client.messages.stream({
    model: 'claude-sonnet-5',
    max_tokens: 1024,
    messages: [{ role: 'user', content: 'Write a story.' }]
});

for await (const event of stream) {
    if (event.type === 'content_block_delta' &&
        event.delta.type === 'text_delta') {
        process.stdout.write(event.delta.text);
    }
}

const finalMessage = await stream.finalMessage();

イベントタイプのリファレンス

イベント	タイミング	データ
message_start	開始時	メッセージのメタデータ
content_block_start	ブロックの開始時	ブロックのタイプ、インデックス
content_block_delta	コンテンツのチャンク	差分コンテンツ
content_block_stop	ブロックの終了時	-
message_delta	メッセージの更新時	停止理由、使用量
message_stop	完了時	-

差分タイプ

差分タイプ	コンテンツ	タイミング
text_delta	.text	テキストコンテンツ
input_json_delta	.partial_json	ツール入力
thinking_delta	.thinking	拡張思考
signature_delta	.signature	思考の署名

ツール使用のストリーミング

重要なルール：ストリームの途中で JSON を解析しないでください！

# WRONG - Will fail on partial JSON!
for event in stream:
    if event.delta.type == "input_json_delta":
        tool_input = json.loads(event.delta.partial_json)  # FAILS!

# CORRECT - Accumulate then parse
tool_json_buffer = ""
for event in stream:
    if event.delta.type == "input_json_delta":
        tool_json_buffer += event.delta.partial_json
    elif event.type == "content_block_stop":
        if tool_json_buffer:
            tool_input = json.loads(tool_json_buffer)  # Safe now!
            tool_json_buffer = ""

ツールのストリーミングにおける完全なパターン

def stream_with_tools(client, messages, tools):
    current_block = None
    tool_input_buffer = ""

    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        messages=messages,
        tools=tools
    ) as stream:
        for event in stream:
            if event.type == "content_block_start":
                current_block = event.content_block
                tool_input_buffer = ""

            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    yield {"type": "text", "content": event.delta.text}
                elif event.delta.type == "input_json_delta":
                    tool_input_buffer += event.delta.partial_json

            elif event.type == "content_block_stop":
                if current_block.type == "tool_use":
                    yield {
                        "type": "tool_call",
                        "id": current_block.id,
                        "name": current_block.name,
                        "input": json.loads(tool_input_buffer)
                    }

拡張思考のストリーミング

thinking_content = ""
signature = ""

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=16000,
    thinking={"type": "adaptive"},
    messages=[{"role": "user", "content": "Solve this complex problem..."}]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "thinking_delta":
                thinking_content += event.delta.thinking
                # Optionally display thinking in UI
            elif event.delta.type == "signature_delta":
                signature = event.delta.signature
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="")

エラー処理

再試行可能なエラー

import time

RETRIABLE_ERRORS = [529, 429, 500, 502, 503]

def stream_with_retry(client, **kwargs):
    max_retries = 3
    base_delay = 1

    for attempt in range(max_retries):
        try:
            with client.messages.stream(**kwargs) as stream:
                for event in stream:
                    yield event
            return
        except anthropic.APIStatusError as e:
            if e.status_code in RETRIABLE_ERRORS and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
            else:
                raise

表示しない過負荷エラー

# CRITICAL: Check for error events even at HTTP 200!
for event in stream:
    if event.type == "error":
        if event.error.type == "overloaded_error":
            # Retry with backoff
            pass

接続管理

キープアライブの設定

import httpx

# Proper timeout configuration
http_client = httpx.Client(
    timeout=httpx.Timeout(
        connect=10.0,    # Connection timeout
        read=120.0,      # Read timeout (long for streaming!)
        write=30.0,      # Write timeout
        pool=30.0        # Pool timeout
    )
)

client = anthropic.Anthropic(http_client=http_client)

コネクションプーリング

http_client = httpx.Client(
    limits=httpx.Limits(
        max_keepalive_connections=20,
        max_connections=100,
        keepalive_expiry=30.0
    )
)

ベストプラクティス

推奨事項：

• 読み取りタイムアウトを 60 秒以上に設定します
• ツールの JSON を蓄積し、block_stop の後に解析します
• HTTP 200 の場合でもエラーイベントを処理します
• 再試行には指数バックオフを使用します

禁止事項：

• ストリーミング中に不完全な JSON を解析しないでください
• 短いタイムアウトを使用しないでください
• overloaded_error イベントを無視しないでください
• 接続を 5 分以上アイドル状態のままにしないでください

UI 統合パターン

async def stream_to_ui(websocket, prompt):
    """Stream Claude response to WebSocket client"""
    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        async for text in stream.text_stream:
            await websocket.send_json({
                "type": "chunk",
                "content": text
            })

        await websocket.send_json({
            "type": "complete",
            "usage": stream.get_final_message().usage
        })

関連項目

• [[llm-integration]] - API の基本
• [[tool-use]] - ツール呼び出し
• [[extended-thinking]] - 深い推論