はじめに / 対象と前提
Claude API のレスポンスを 1 文字ずつ画面に流したい、あるいは長い出力でタイムアウトさせたくない、という人向けの記事。ストリーミング自体は数行で動くが、tool use を混ぜた瞬間とエラー処理で一段難しくなる。そこを中心に書く。
- 想定読者:Claude API を非ストリーミングでは叩いたことがある Web エンジニア
- 前提知識:Python の基本、SSE(Server-Sent Events)が何かをざっくり知っている
- 環境:
- Python 3.14
-
anthropicPython 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 箇所から取る - 長い出力は、表示の要不要に関係なくストリーミングで受ける