SSEイベント種別をSDKなしで生パースする方法

はじめに — 前提とゴール

Claude APIのストリーミング応答は、SDKを使えばイベントの型が自動でパースされます。ただし公式SDKが無い言語や、依存を増やしたくない環境では、cURLで受けた生のHTTPストリームを自分でパースする必要が出てきます。本稿は curl と標準ライブラリの json だけで、6種のコアイベントと content_block_delta の3つのサブタイプを組み立てる手順を書きます。

実装に必要な前提は次の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",
    "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","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 — イベント種別ごとに分岐して処理する

ストリーム全体は、message_start → コンテンツブロックごとの content_block_start / content_block_delta の連続 / content_block_stop → 1つ以上の message_delta → message_stop という順で流れます。この間に ping イベントが任意の数だけ挟まることもあります。

パーサーがそれぞれのイベント種別で何をすべきかを一覧にすると次のとおりです。

この表を1つの分岐関数として実装しておくと、後から新しいコンテンツブロック種別が追加された場合でも、既存の行に手を入れずに新しい行を1本足すだけで対応できるようになります。

state側に持たせる最小フィールドも種別ごとに違います。message_start は message.id と message.model、content_block_start はindex単位の content_block.type と初期値、content_block_delta は蓄積用の文字列バッファ、message_delta は最新の usage だけを持てば足ります。逆に言うと、それ以外のフィールドをstateに抱え込む必要はありません。content_block_start の段階で content_block.type を見ておけば、後続の content_block_delta がどの delta.type を送ってくるかを事前に予測でき、想定外の delta.type が来たときだけ分岐を追加すればよいという設計にできます。

data: 行だけを読み、JSONの type で分岐する最小のパーサーは次のとおりです。

import json
import sys
 
current_text = ""
tool_json_parts = {}
 
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
    elif etype == "error":
        print("stream error:", event["error"]["type"], file=sys.stderr)
        break
    elif etype == "content_block_delta":
        delta = event["delta"]
        if delta["type"] == "text_delta":
            current_text += delta["text"]
        elif delta["type"] == "input_json_delta":
            idx = event["index"]
            tool_json_parts.setdefault(idx, "")
            tool_json_parts[idx] += delta["partial_json"]
        elif delta["type"] == "thinking_delta":
            pass  # thinkingテキストを蓄積したい場合はここに追加する
    elif etype == "content_block_stop":
        idx = event["index"]
        if idx in tool_json_parts:
            tool_input = json.loads(tool_json_parts[idx])
            print("tool input:", tool_input)
    elif etype in ("message_start", "content_block_start", "message_delta", "message_stop"):
        pass
    else:
        # 未知のイベント種別は無視して継続する(バージョニングポリシーで随時追加される)
        continue
 
print(current_text)

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","max_tokens":256,"stream":true,"messages":[{"role":"user","content":"Hello"}]}' \
  | python3 parse_stream.py

ステップ3 — deltaの3種類を組み立てる

content_block_delta イベントは、ブロックの種類に応じて3つの delta.type を持ちます。それぞれ蓄積のしかたが異なるため、パーサー側の分岐も型ごとに変える必要があります。

テキストブロックは text_delta で、delta.text をそのまま連結するだけで元の文字列に戻ります。

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ello frien"}}

ツール呼び出しは input_json_delta です。tool_use.input の最終形はオブジェクトですが、デルタとして送られてくるのは部分的なJSON文字列です。文字列を蓄積してから content_block_stop を受け取った時点で1回だけパースします。現行モデルは input の1つのキーと値をまとめて送るため、ツール呼び出しの途中でイベントの間隔が空くことがあります。

event: content_block_delta
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"location\": \"San Fra"}}}

ツール呼び出しでは、content_block_start の時点で id と name が確定しています。input はこの段階では空オブジェクトで、続く input_json_delta の連なりで中身が埋まっていきます。

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\"}"}}

先のパーサーは input の中身しか蓄積していませんが、content_block_start の id と name も同じ辞書に記録しておくと、content_block_stop の時点で「どのツールに何を渡すか」が1つのオブジェクトとして揃います。

thinkingブロックは thinking_delta です。display: "omitted" を設定している場合、thinking_delta は1件も送られず、ブロックの開始直後に signature_delta が1回送られて閉じます。この違いを踏まえずに「必ず thinking_delta が来る」前提で実装すると、display 設定を変えた途端にパーサーが空振りします。

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"1071 = 2 x 462 + 147"}}

デルタを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 が来るだけで content_block_delta は1件も発生しません。「1つのコンテンツブロックは必ずdeltaを持つ」という前提でパーサーを組むと、このブロックの通過時にタイムアウト待ちのような挙動になります。ブロックの種類によってはdeltaが0件でも正常、というケースを最初から想定に入れておきます。

サーバー側フォールバックが混ざるとどうなるか

fallbacks パラメーターを付けたリクエストがストリーミング中に拒否され、サーバー側フォールバックが発動すると、通常の6種類のイベントに加えて特別なコンテンツブロックが割り込みます。モデルの切り替わりごとに、content_block_start と content_block_stop だけが対になった fallback というタイプのブロックが挟まり、その間に content_block_delta は1件も来ません。

さらに、最初の message_start は既定では空の content を持つだけですが、thinking-binding-controls-2026-08-01 というベータヘッダーを付けたリクエストでは input_transformations という配列も一緒に届きます。サーバー側フォールバックが起きた場合、最後の message_delta にもこの配列が再度含まれ、実際に応答したモデルの情報が入ります。

自前のパーサーで content_block_start の content_block.type を見ずに「content_block_delta が1件も来ないのはおかしい」という前提で組んでいると、fallback ブロックの通過時にハングしたりエラー扱いにしたりします。デルタが0件のコンテンツブロックはエラーではなく正常なフォールバック境界として扱う分岐を用意しておきます。

パーサーが再開処理に渡せる情報

自前パーサーの役割は、ネットワーク断やエラー種別ごとのリトライ手順そのものを実装することではなく、再開に使える「完全なブロック」がどこまで届いていたかを正確に切り出すことです。再開の起点になるのは content_block_stop まで届いた完全なブロックだけで、content_block_delta の途中で切れたブロックはその時点で破棄し、手前の完了ブロックまでをパーサーの出力として渡します。

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":""}}

パーサー側でブロックの最終形をオブジェクトとして組み立てたいときは、content_block_start の content_block をそのブロックの初期状態としてコピーし、その後に届く delta で該当フィールドを上書き・追記していく実装にすると、テキスト・ツール入力・thinkingのどのブロックにも同じロジックで対応できます。

世代別の再開手順(assistantメッセージとして続けるか、userメッセージで指示するか)と、error.type ごとにリトライすべきか判断する分類は、パース処理そのものとは別のレイヤーの話です。本稿はSSEの生パース実装に絞り、その2点はClaude APIストリーミングのエラー復旧にまとめています。

よくあるつまずき

ツール入力のJSONを content_block_delta が届くたびにパースしようとすると、途中の文字列は不完全なJSONなので必ず例外になります。content_block_stop まで文字列を蓄積してから1回だけパースする実装にします。

message_delta の usage にあるトークン数は累積値です。イベントごとに加算していくと、実際の消費量より大きい数字になります。最後に届いた message_delta の値をそのまま使います。

ping イベントは接続維持のための空イベントで、エラーではありません。data: に type: "ping" が来たら何もせず読み飛ばします。

未知のイベント種別が来る可能性は仕様上明記されています。新しいイベント種別が追加されてもクラッシュしない設計(未知の type は無視して継続する)にしておくと、APIのバージョンアップでパーサーが壊れません。

パース対象を data: 行だけに絞ると実装は単純になりますが、event: 行と data: 内の type が一致しない実装ミスに気づきにくくなります。デバッグ時は両方をログに残しておくと原因の切り分けが早くなります。ツール呼び出しのストリーミングでinvalid jsonに遭遇した場合の対処はツールストリーミングでinvalid jsonが来たときの直し方にまとめています。

実装を書き終えたら、text・tool_use・server_tool_use・web_search_tool_result・thinking・fallbackの6種類の content_block タイプでそれぞれ動作確認しておくと、本番で未知のブロックに遭遇したときの切り分けが早くなります。web検索やコード実行などのサーバーツールを使うリクエストと、使わないシンプルなリクエストの両方をテストケースに含めておくのが実務的です。

まとめ

公式SDKを使わずにClaude APIのストリーミング応答をパースする実装は、data: 行のJSONから type を読み、message_start / content_block_start / content_block_delta / content_block_stop / message_delta / message_stop の6種と ping / error を分岐処理するだけで組み立てられます。content_block_delta はさらに text_delta / input_json_delta / thinking_delta の3種類に分かれ、ツール入力だけは蓄積してから1回パースする必要があります。累積カウンターの usage と、随時追加される未知のイベント種別への耐性を押さえておけば、生パース実装として実用に足ります。サーバー側フォールバックの fallback ブロックや、サーバーツールが返す web_search_tool_result のようにdeltaを1件も送らないブロックがある点も、実装時にあらかじめ想定しておくとハングやタイムアウトの誤検知を防げます。Agent SDKでのストリーミング出力を有効にする手順はAgent SDKのストリーミング出力を有効にするにまとめています。