はじめに / 対象と前提

Claude API のレスポンスを 1 文字ずつ画面に流したい、あるいは長い出力でタイムアウトさせたくない、という人向けの記事。ストリーミング自体は数行で動くが、tool use を混ぜた瞬間とエラー処理で一段難しくなる。そこを中心に書く。

想定読者:Claude API を非ストリーミングでは叩いたことがある Web エンジニア
前提知識:Python の基本、SSE(Server-Sent Events)が何かをざっくり知っている
環境:
- Python 3.14
- anthropic Python SDK(2026 年 9 月時点の最新版)
- モデル:claude-sonnet-5
- macOS 15

イベント名やフィールド名は Messages API のストリーミング仕様に沿っている。SDK のバージョンが離れている場合は、手元の型定義と見比べてほしい。

TL;DR

テキストを流すだけなら client.messages.stream() と text_stream で足りる
tool use の引数は input_json_delta の断片文字列で届く。断片ごとに json.loads すると落ちるので、content_block_stop まで溜めてからパースする
ストリーミングは HTTP 200 を返した後に失敗する。ステータスコードだけ見ていると取りこぼす
usage は message_start と message_delta の 2 箇所に分かれて届く

手順 / 動かし方

1. 最小構成(テキストを流すだけ)

pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "SSE を3行で説明して"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final = stream.get_final_message()

print()
print(final.stop_reason, final.usage)

with を抜けると接続が閉じる。get_final_message() は、断片を SDK 側で組み立て直した完成形の Message を返すので、非ストリーミングと同じ形で後続処理に渡せる。

print に flush=True を付け忘れると、バッファに溜まって最後にまとめて出る。「ストリーミングが効いていない」と勘違いしやすいので最初に確認する。

2. イベントの流れを把握する

ストリームの中身は、以下の順でイベントが届く。

途中に ping イベントが挟まることがある。また content_block_delta の中身は delta.type で種類が変わる。

delta.type	中身	出るタイミング
`text_delta`	`text`	通常のテキスト
`input_json_delta`	`partial_json`	tool use の引数
`thinking_delta`	`thinking`	拡張思考を有効にしたとき
`signature_delta`	`signature`	thinking ブロックの終わり

3. tool use を含むストリームを自前で処理する

生イベントを自分で回す版。ブロックの index をキーにして断片を溜めるのが要点。

import json
import anthropic

client = anthropic.Anthropic()

tools = [{
    "name": "get_weather",
    "description": "指定した都市の現在の天気を返す",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

blocks = {}       # index -> {"type", "name", "id", "buf"}
tool_calls = []

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "東京の天気は?"}],
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            cb = event.content_block
            blocks[event.index] = {
                "type": cb.type,
                "name": getattr(cb, "name", None),
                "id": getattr(cb, "id", None),
                "buf": "",
            }
        elif event.type == "content_block_delta":
            d = event.delta
            if d.type == "text_delta":
                print(d.text, end="", flush=True)
            elif d.type == "input_json_delta":
                blocks[event.index]["buf"] += d.partial_json
        elif event.type == "content_block_stop":
            b = blocks[event.index]
            if b["type"] == "tool_use":
                args = json.loads(b["buf"]) if b["buf"] else {}
                tool_calls.append({"id": b["id"], "name": b["name"], "input": args})

print(tool_calls)

期待する出力の形は次のとおり(id は毎回変わる)。

[{'id': 'toolu_...', 'name': 'get_weather', 'input': {'city': '東京'}}]

ハマりどころ

1. input_json_delta を都度パースして落ちる

最初に書いたコードは、delta が来るたびに json.loads(d.partial_json) していた。結果はこうなる。

json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)

原因:partial_json は「JSON の一部分の文字列」で、単体では JSON として成立しない。{"ci → ty": " → 東京"} のように、キーの途中でも容赦なく切れる。最初の delta が空文字のこともある。

回避策:

index ごとにバッファへ連結し、content_block_stop で 1 回だけパースする
引数なしのツールはバッファが空のまま終わるので、json.loads("") を避けて {} にフォールバックする
並列ツール実行では複数の tool_use ブロックが来る。バッファを 1 本の変数で持つと混ざるので、必ず index で分ける

自前で組み立てる必要がなければ、stream.get_final_message() の content から tool_use ブロックを取るのが一番安全。進捗表示のために途中経過を見たいときだけ、生イベントを触ればいい。

2. HTTP 200 の後に error イベントで失敗する

非ストリーミングなら、過負荷時は 529、レート制限は 429 が返ってくるので、ステータスコードで分岐できる。ストリーミングは違う。レスポンスヘッダーで 200 を返し、本文を流し始めてから、途中でエラーになることがある。

event: error
data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}

原因:SSE は最初にヘッダーを送ってしまうので、その後に起きた障害をステータスコードでは伝えられない。

回避策:for ループごと try で囲み、途中まで受け取ったテキストの扱いを決めておく。

received = []
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "..."}],
    ) as stream:
        for text in stream.text_stream:
            received.append(text)
except anthropic.APIConnectionError as e:
    print("接続断:", e)
except anthropic.APIStatusError as e:
    print("API エラー:", e.status_code, e.message)
except anthropic.APIError as e:
    print("ストリーム中のエラー:", e)

注意点は 2 つ。

例外クラスは具体的なものから先に書く。APIError は基底クラスなので、先頭に置くと全部そこで捕まる
SDK の自動リトライは接続確立までが対象。ストリームの途中で切れた分は自分で再送する必要がある。そのとき received をそのまま画面に残すと、再送後の出力と二重になる。「捨てて最初からやり直す」か「途中までを assistant メッセージとして渡して続きを書かせる」かを先に決めておく

3. usage を取り違える

コスト集計のために message_start の usage だけを記録していたら、出力トークンが常にごく小さい値になっていた。

原因:usage は 2 箇所に分かれて届く。

イベント	取れるもの
`message_start`	`input_tokens`(キャッシュ関連の値もここ)。`output_tokens` は開始時点の値
`message_delta`	`output_tokens` の累積値と `stop_reason`

message_delta の output_tokens は差分ではなく累積なので、足し込むと多重計上になる。

回避策:

usage = {"input": 0, "output": 0}
stop_reason = None

for event in stream:
    if event.type == "message_start":
        usage["input"] = event.message.usage.input_tokens
    elif event.type == "message_delta":
        usage["output"] = event.usage.output_tokens   # 上書き
        stop_reason = event.delta.stop_reason

stop_reason も message_start の時点では null で、message_delta で初めて確定する。ここでも get_final_message() を使えば、両方が埋まった状態で取れる。

背景・補足

長い出力は非ストリーミングだと SDK に止められる

max_tokens を大きくして非ストリーミングで呼ぶと、SDK がリクエスト送信前にエラーを出すことがある。応答に 10 分以上かかりうる呼び出しでは、途中のネットワーク機器にアイドル接続を切られる恐れがあるため、ストリーミングが強く推奨されている。

「画面に流す必要はないが出力は長い」という場合は、ストリーミングで受けて get_final_message() だけ使えばいい。呼び出し側のコードは非ストリーミングとほぼ同じ形で書ける。

非同期版

FastAPI などから使うなら AsyncAnthropic に替える。

client = anthropic.AsyncAnthropic()

async with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "..."}],
) as stream:
    async for text in stream.text_stream:
        yield text

クライアント(ブラウザ)が切断したときに async with を抜けるようにしておかないと、誰も読まない出力のトークン代を払い続けることになる。

まとめ

テキスト表示だけなら text_stream、完成形が欲しければ get_final_message()
tool use の引数は断片で届く。index ごとに溜めて content_block_stop でパースする
エラーは HTTP 200 の後にも来る。ループ全体を try で囲み、途中まで受け取った分の扱いを決めておく
usage は message_start(入力)と message_delta(出力の累積)の 2 箇所から取る
長い出力は、表示の要不要に関係なくストリーミングで受ける