SSEイベント種別をSDKなしで生パースする方法
Claude APIのストリーミングが送るSSEイベント種別を、公式SDKを使わずcURLと標準ライブラリだけで読み解く実装手順をまとめる。
はじめに — 前提とゴール
Claude APIのストリーミング応答は、SDKを使えばイベントの型が自動でパースされます。ただし公式SDKが無い言語や、依存を増やしたくない環境では、cURLで受けた生のHTTPストリームを自分でパースする必要が出てきます。本稿は curl と標準ライブラリの json だけで、6種のコアイベントと content_block_delta の4つのサブタイプを組み立てる手順を書きます。
実装に必要な前提は次の3つだけです。ANTHROPIC_API_KEY が環境変数に設定済みであること、curl が使えること、パース側はPython 3(標準ライブラリのみ)を使うことです。他言語でも標準の文字列処理とJSONパーサーがあれば同じ手順で実装できます。ここで組む処理は「1行読んで data: かどうか判定し、JSONの type で分岐する」だけなので、外部ライブラリを1つも持たない環境や、依存を最小限に抑えたいバッチジョブへの組み込みにも向いています。
ステップ1 — curlでイベントストリームを取得する
まずストリーミングを有効にしたリクエストを送ります。ポイントは -N オプションです。これを付けないとcurlが応答をバッファリングしてからまとめて出力するため、ストリーミングの意味がなくなります。
curl -N https://api.anthropic.com/v1/messages \
-H "content-type: application/json" \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-d '{
"model": "claude-opus-5-5",
"max_tokens": 256,
"stream": true,
"messages": [{"role": "user", "content": "Hello"}]
}'このリクエストが返す生のストリームは、次のような event: / data: の組が連続する形式です。
event: message_start
data: {"type":"message_start","message":{"id":"msg_01...","type":"message","role":"assistant","content":[],"model":"claude-opus-5-5","usage":{"input_tokens":10,"output_tokens":1}}}
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
event: content_block_stop
data: {"type":"content_block_stop","index":0}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":15}}
event: message_stop
data: {"type":"message_stop"}各イベントは event: 行のSSEイベント名と、data: 行のJSONに含まれる type フィールドの両方でイベント種別を示します。両方を見なくても、data: のJSONだけで種別判定は完結します。
ステップ2 — イベント種別ごとに分岐して処理する
1つのストリームは、次の順で流れます。この間に ping イベントが任意の数だけ挟まることもあります。
1つのストリームの流れ
- 1
message_start
空の
contentを持つMessageオブジェクトが届きます。 - 2
コンテンツブロックの繰り返し
ブロックごとに
content_block_start、1件以上のcontent_block_delta、content_block_stopが並びます。indexは最終的なcontent配列の添字に対応します。 - 3
message_delta(1回以上)
メッセージ全体の変化(
stop_reasonやusage)が届きます。 - 4
message_stop
ストリームの終端です。
パーサーがそれぞれのイベント種別で何をするかを一覧にすると次のとおりです。
| イベント種別 | パーサーが取るべき処理 | 状態遷移 | 落とし穴 |
|---|---|---|---|
message_start | パーサーが取るべき処理message.id と初期usageを保持 | 状態遷移ストリーム開始 | 落とし穴ここでusageを最終値と誤読しない |
content_block_start | パーサーが取るべき処理content_block を該当indexの初期状態としてコピー | 状態遷移ブロック生成 | 落とし穴tool_useはinputが空オブジェクトで始まる |
content_block_delta | パーサーが取るべき処理delta.typeでさらに分岐 | 状態遷移ブロック更新 | 落とし穴delta.typeを見ずにtext決め打ちしない |
content_block_stop | パーサーが取るべき処理蓄積した文字列やinputを確定・パース | 状態遷移ブロック確定 | 落とし穴tool_use入力はここで初めて1回だけパースする |
message_delta | パーサーが取るべき処理usageを最新値で上書き | 状態遷移メッセージ更新 | 落とし穴加算すると累積値を二重計上する |
message_stop | パーサーが取るべき処理ストリーム終了処理 | 状態遷移終了 | 落とし穴後続イベントを待ち続けない |
ping | パーサーが取るべき処理何もせず読み飛ばす | 状態遷移変化なし | 落とし穴エラーと誤判定しない |
error | パーサーが取るべき処理error.typeをログしてループを抜ける | 状態遷移異常終了 | 落とし穴ストリームの途中でも届く |
分岐は1つの関数にまとめておくと、新しいコンテンツブロック種別が増えたときに行を1本足すだけで済みます。stateに持つものも最小です。ブロックごとの辞書(content_block_start の内容をそのままコピーしたもの)、input_json_delta の文字列バッファ、最新の usage があれば足ります。
次のパーサーは、この方針をそのまま実装したものです。content_block_start の内容をindex単位でコピーし、deltaで該当フィールドを上書き・追記していきます。
import json
import sys
blocks = {} # index -> 組み立て中のブロック
partial = {} # index -> input_json_delta の蓄積
usage = {}
for raw_line in sys.stdin:
line = raw_line.rstrip("\n")
if not line.startswith("data:"):
continue
payload = line[len("data:"):].strip()
if not payload:
continue
event = json.loads(payload)
etype = event.get("type")
if etype == "ping":
continue
if etype == "error":
print("stream error:", event["error"]["type"], file=sys.stderr)
break
if etype == "content_block_start":
blocks[event["index"]] = dict(event["content_block"])
elif etype == "content_block_delta":
idx, delta = event["index"], event["delta"]
block = blocks[idx]
if delta["type"] == "text_delta":
block["text"] += delta["text"]
elif delta["type"] == "input_json_delta":
partial[idx] = partial.get(idx, "") + delta["partial_json"]
elif delta["type"] == "thinking_delta":
block["thinking"] += delta["thinking"]
elif delta["type"] == "signature_delta":
block["signature"] = delta["signature"]
elif etype == "content_block_stop":
idx = event["index"]
if idx in partial:
blocks[idx]["input"] = json.loads(partial.pop(idx))
print(idx, blocks[idx])
elif etype == "message_delta":
usage = event["usage"] # 累積値なので上書きする
# message_start / message_stop / 未知の type は何もしない
print("usage:", usage)curlの出力をそのままこのスクリプトへパイプします。
curl -N https://api.anthropic.com/v1/messages \
-H "content-type: application/json" \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-d '{"model":"claude-opus-5-5","max_tokens":256,"stream":true,"messages":[{"role":"user","content":"Hello"}]}' \
| python3 parse_stream.pyAPIキーを使わずに分岐を試すには、イベント列をファイルに保存して標準入力から流します。テキスト、tool_use、web_search_tool_result の3ブロックと ping、message_delta を含むイベント列を fixture.sse に保存し、このパーサーで再生します。fixture.sse の中身は次のとおりです。
event: message_start
data: {"type":"message_start","message":{"id":"msg_01","type":"message","role":"assistant","content":[],"model":"claude-opus-5-5","usage":{"input_tokens":472,"output_tokens":2}}}
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
event: ping
data: {"type": "ping"}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me check."}}
event: content_block_stop
data: {"type":"content_block_stop","index":0}
event: content_block_start
data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_01T1x1fJ34qAmk2tNTrN7Up6","name":"get_weather","input":{}}}
event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":""}}
event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"location\":"}}
event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":" \"San Francisco, CA\"}"}}
event: content_block_stop
data: {"type":"content_block_stop","index":1}
event: content_block_start
data: {"type":"content_block_start","index":2,"content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_014","content":[]}}
event: content_block_stop
data: {"type":"content_block_stop","index":2}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":89}}
event: message_stop
data: {"type":"message_stop"}再生する手順と出力が次です。
python3 parse_stream.py < fixture.sse0 {'type': 'text', 'text': 'Let me check.'}
1 {'type': 'tool_use', 'id': 'toolu_01T1x1fJ34qAmk2tNTrN7Up6', 'name': 'get_weather', 'input': {'location': 'San Francisco, CA'}}
2 {'type': 'web_search_tool_result', 'tool_use_id': 'srvtoolu_014', 'content': []}
usage: {'output_tokens': 89}tool_use の input は {} で始まったあと、3つの input_json_delta の連結を1回パースした結果に置き換わっています。id と name は content_block_start で確定済みのため、content_block_stop の時点で「どのツールに何を渡すか」が1つの辞書に揃います。web_search_tool_result はdeltaなしで確定します。fixtureでは content が空配列ですが、実際の応答では content_block_start の時点で検索結果が入っています。
ステップ3 — deltaの4種類を組み立てる
content_block_delta イベントは、ブロックの種類に応じた delta.type を持ちます。蓄積のしかたが型ごとに違うため、パーサー側の分岐も型ごとに変えます。
delta.typeごとの蓄積先
text_delta
delta.textを連結すると元の文字列に戻ります。input_json_delta
delta.partial_jsonは部分的なJSON文字列です。content_block_stopまで蓄積して1回だけパースします。thinking_delta
delta.thinkingを連結します。signature_delta
thinkingブロックの整合性検証に使う署名です。
content_block_stopの直前に届きます。
テキストブロックの例です。
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ello frien"}}ツール呼び出しでは、tool_use.input の最終形はオブジェクトですが、デルタとして届くのは部分的なJSON文字列です。現行モデルは input の1つのキーと値をまとめて送るため、ツール呼び出しの途中でイベントの間隔が空くことがあります。
event: content_block_start
data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_01T1x1fJ34qAmk2tNTrN7Up6","name":"get_weather","input":{}}}
event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"location\":"}}
event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":" \"San Francisco, CA\"}"}}thinkingブロックは thinking_delta で本文が届き、signature_delta で閉じます。display: "omitted" を設定した場合は本文が流れません。ブロックが開き、thinking が空文字列の thinking_delta が届き、signature_delta が1回届いて閉じます。「thinking_delta に必ずテキストが入っている」前提で実装すると、display を変えた途端に空の文字列だけを受け取ることになります。
thinkingブロックの content_block_start も、textブロックと同じく空の器から始まります。thinking と signature の両方が空文字列で始まり、thinking_delta が本文を、signature_delta が署名を埋めていきます。
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"1071 = 2 x 462 + 147"}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"EqQBCgIYAhIM1gbcDa9GJwZA2b3hGgxBdjrkzLoky3dl1pkiMOYds..."}}deltaを1件も送らないブロックもある
Web検索やコード実行などのサーバーツールを使うと、content_block_delta を1件も送らないコンテンツブロックが混ざります。ツール呼び出し自体は server_tool_use というタイプで、通常の tool_use と同じく input_json_delta の連なりで入力が組み上がります。その検索結果を返す web_search_tool_result ブロックは事情が違います。
event: content_block_start
data: {"type":"content_block_start","index":2,"content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_014hJH82Qum7Td6UV8gDXThB","content":[{"type":"web_search_result","title":"...","url":"...","encrypted_content":"...","page_age":null}]}}
event: content_block_stop
data: {"type":"content_block_stop","index":2}このブロックは content_block_start の時点で中身がすべて入っており、直後に content_block_stop が来るだけです。「1つのコンテンツブロックは必ずdeltaを持つ」前提でdeltaからブロックを組み立てると、このブロックだけ中身が空のまま確定します。deltaが0件でも正常なブロックがあることを、最初から想定に入れておきます。
サーバー側フォールバックが混ざるとどうなるか
fallbacks パラメーターを付けたリクエストがストリーミング中に拒否され、サーバー側フォールバックが発動すると、通常のイベントに加えて特別なコンテンツブロックが割り込みます。モデルの切り替わりごとに、content_block_start と content_block_stop だけが対になった fallback というタイプのブロックが挟まり、その間に content_block_delta は1件も来ません。
fallbacksはベータ機能で、リクエストにはベータヘッダー server-side-fallback-2026-07-01 が要ります。Amazon Bedrock、Google Cloud、Microsoft Foundry、Message Batches APIでは使えません。
deltaが届くブロックと届かないブロック
deltaが届く
text・tool_use・server_tool_use・thinking です。content_block_start の内容に、後続のdeltaで中身を足して確定します。
deltaが届かない
web_search_tool_result は content_block_start に中身が入っています。fallback はモデルの切り替わり位置を示すだけで、中身を持ちません。
message_start についても追加の情報があります。既定では空の content を持つだけですが、thinking-binding-controls-2026-08-01 というベータヘッダーを付けたリクエストでは input_transformations という配列も一緒に届きます。サーバー側フォールバックが起きた場合、最後の message_delta にもこの配列が再度含まれ、実際に応答したモデルの情報が入ります。
パーサーが再開処理に渡せる情報
自前パーサーの役割は、リトライ手順そのものを実装することではなく、再開に使える内容がどこまで届いていたかを正確に切り出すことです。tool_use とthinkingのブロックは途中までの内容から復元できないため、content_block_stop まで届いていない場合は捨てます。再開の起点にできるのは、直近のtextブロックまでです。SDKのヘルパー側で未知のイベントがどう扱われるかは、Anthropic SDK 1.14のstream()に未知のイベントが届く条件にあります。
世代別の再開手順(assistantメッセージとして続けるか、userメッセージで指示するか)と、error.type ごとにリトライを判断する分類は、パース処理とは別のレイヤーの話です。Claude APIストリーミングのエラー復旧に、その2点をまとめています。
よくあるつまずき
ツール入力のJSONを content_block_delta が届くたびにパースしようとすると、途中の文字列は不完全なJSONなので必ず例外になります。content_block_stop まで文字列を蓄積してから1回だけパースする実装にします。
message_delta の usage にあるトークン数は累積値です。イベントごとに加算していくと、実際の消費量より大きい数字になります。最後に届いた message_delta の値をそのまま使います。
ping イベントは、ストリームの途中に任意の数だけ挟まるイベントです。エラーではないので、data: に type: "ping" が来たら何もせず読み飛ばします。
未知のイベント種別が来る可能性は、バージョニングポリシーで予告されています。未知の type は無視して継続する設計にしておくと、新しい種別が追加されてもパーサーが落ちません。
パース対象を data: 行だけに絞ると実装は単純になりますが、event: 行と data: 内の type が一致しない実装ミスには気づきにくくなります。デバッグ時は両方をログに残しておくと原因の切り分けが早くなります。ツール呼び出しのストリーミングでinvalid jsonに遭遇した場合の対処はツールストリーミングでinvalid jsonが来たときの直し方にまとめています。
実装を書き終えたら、text・tool_use・server_tool_use・web_search_tool_result・thinking・fallbackの6種類の content_block タイプでそれぞれ動作確認します。サーバーツールを使うリクエストと使わないリクエストの両方を、テストケースに含めておくのが実務的です。
まとめ
SSEの生パースは、data: 行のJSONの type だけで分岐できます。ブロックは content_block_start の内容をコピーして始め、deltaで埋めていく形にしておくと、テキスト・ツール入力・thinkingを同じ流れで扱えます。deltaを送らないブロックの存在を前提にしておけば、サーバーツールやフォールバックが混ざるストリームでも分岐が崩れません。Agent SDKでのストリーミング出力を有効にする手順はAgent SDKのストリーミング出力を有効にするにまとめています。