はじめに / 対象と前提

Claude API でチャット UI の「文字がだんだん表示される」あれ、ストリーミング(SSE: Server-Sent Events)を自前で組んだことはあるだろうか。公式 SDK の client.messages.stream() は数行で動くが、型分岐を1つサボるだけで「tool_use の引数だけ乱れる」「たまに例外で落ちる」不具合を踏む。

前提。

Python 3.13、anthropic SDK 0.6x 系
Messages API を直接叩く構成(Bedrock/Vertex でも同様)
テキスト応答に加え tool_use も併用するケース

TL;DR

text_stream だけ見ると tool_use の引数を取りこぼす
SSE は message_start → content_block_start → content_block_delta(複数)→ content_block_stop → message_stop の順で来るが、delta は text_delta / input_json_delta / thinking_delta の3種類が混在する
input_json_delta は JSON の断片文字列。貯めてから最後に1回だけ json.loads する。逐次パースは壊れる
ping とエラー系イベントを拾わないと長時間ストリームでパーサーごと落ちる

手順 / 動かし方

最小構成、テキストだけをストリーミング表示する。

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY は環境変数から

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Pythonでフィボナッチ数列を書いて"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final_message = stream.get_final_message()

print()
print("usage:", final_message.usage)

text_stream はテキストのデルタだけを流す便利プロパティだが、tool_use を使うと ここには何も流れてこない。引数の組み立てを見るには生イベントを拾う必要がある。

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    tools=[{
        "name": "get_weather",
        "description": "指定した都市の天気を返す",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }],
    messages=[{"role": "user", "content": "東京の天気は?"}],
) as stream:
    partial_json_buf = ""
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                partial_json_buf = ""
        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
            elif event.delta.type == "input_json_delta":
                partial_json_buf += event.delta.partial_json
            elif event.delta.type == "thinking_delta":
                pass  # extended thinking 用
        elif event.type == "content_block_stop":
            if partial_json_buf:
                import json
                args = json.loads(partial_json_buf)
                print("\n[tool_use args]", args)
                partial_json_buf = ""

ポイントは input_json_delta を 貯めて content_block_stop が来てから1回だけ json.loads すること。

ハマりどころ

1. `event.delta.type` の分岐漏れで tool_use が空文字になる

content_block_delta は text_delta だけだと思い込んで event.delta.text にアクセスすると、input_json_delta のときは AttributeError で落ちる。必ず event.delta.type で分岐する。extended thinking 有効時は thinking_delta も来るので、想定外の type は無視するのが安全。

2. `partial_json` を逐次 `json.loads` しようとして壊れる

{"city": → "To → kyo"} のように JSON はトークン単位でバラバラに届く。届くたびにパースを試みるとエラーになる。リアルタイム表示が不要なら stream.get_final_message() で完成後の tool_use を取るほうが安全。組み立て過程を出したい場合だけ手動バッファリングが要る。

3. `ping` イベントとエラーイベントで例外を握りつぶさない

長文生成や extended thinking で数十秒かかるストリームでは、接続維持用の ping が定期的に挟まる。未知イベントとして例外を投げる実装だと、生成が長引くリクエストだけ再現性低くクラッシュする。overloaded_error もストリーム途中で error イベントとして飛んでくるので、ループを try/except で囲んでおく。

try:
    with client.messages.stream(...) as stream:
        for event in stream:
            if event.type == "ping":
                continue
            # 以下、通常の分岐
except anthropic.APIStatusError as e:
    print(f"stream error: {e.status_code}")

背景・補足

MessageStream は stream.current_message_snapshot で「今の時点までの Message」を取得できる。「今どのブロックが生成中か」だけ分かればよい場合は型分岐を自分で書くよりこちらが実装量が減る。内部実装はバージョン間で変わりうるので、SDK 更新時は動作確認しておくと安全。

まとめ

text_stream はテキスト専用。tool_use は event.type / event.delta.type 分岐が必須
input_json_delta は断片 JSON。content_block_stop まで貯めて json.loads
ping と error を無視すると長時間ストリームだけ再現性低く落ちる
リアルタイム表示が不要なら get_final_message() や current_message_snapshot が安全