Claude APIのメッセージをストリーミングする

Claude API API 6分で読めます

ストリーミングを使うと、Claude が回答を生成しながら少しずつ結果を受け取れます。全文が出来上がるまで待つ必要がないため、チャットUIのように「文字が流れて表示される」体験を作れ、長い応答でも体感の待ち時間を短くできます。本記事では Claude API(Messages API)のストリーミングの仕組みと実装、届くイベントの種類、注意点を解説します。これは Anthropic 公式ドキュメントの非公式日本語訳です。最新の仕様は公式ドキュメントをご確認ください。

ストリーミングの概要

ストリーミングは、Messages API のリクエストで stream: true を指定すると有効になります。応答は Server-Sent Events(SSE)という仕組みで、生成の進行に合わせて小さなイベントが順番に送られてきます。受け取り側はイベントを逐次処理し、テキストの断片(デルタ)をつなぎ合わせて最終的な回答を組み立てます。非ストリーミングでは1回のHTTPレスポンスで完成した本文が返るのに対し、ストリーミングでは接続を開いたまま断片が流れ続ける、という違いです。

SSEイベントの流れ

ストリームでは、おおむね次の順序でイベントが届きます。

これらの合間に、接続維持のための ping イベントが挟まることがあります。実際に画面へ表示する文字は、主に content_block_deltatext_delta を連結して得ます。

SDKでの実装

生のSSEを自分でパースすることもできますが、公式SDK(Python / TypeScript)を使うと、イベントの解釈や断片の連結を SDK が肩代わりしてくれるため実装が簡単になります。Python では client.messages.stream(...) をコンテキストマネージャで開き、text_stream を反復するだけでテキストの断片を順に受け取れます。

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "俳句を1つ詠んでください。"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

TypeScript SDK でも client.messages.stream() が用意され、on('text', ...) のようなイベントハンドラや for await でのイテレーションに対応します。ストリーム終了後に完成したメッセージ全体を取得することもできます。

ツール使用時のストリーミング

ツール使用(tool use)を伴う応答でもストリーミングできます。この場合、ツールに渡す入力(JSON)は一度に届かず、content_block_deltainput_json_delta として少しずつ流れてきます。断片は途中では不完全なJSONなので、content_block_stop までを蓄積してから結合し、そこで初めて完全なJSONとしてパースします。SDKを使えばこの蓄積・結合も自動的に行われ、確定したツール入力を受け取れます。ストリーミングはツールの引数が大きい場合でも、生成の進捗をユーザーに見せられる利点があります。

注意点とエラー処理

ストリーミングでは接続を長く開いたままにするため、途中でネットワークが切れたり、error イベントが届いたりする可能性があります。堅牢に実装するには、(1)未知のイベント種別が来ても無視して処理を続ける(将来イベントが追加されても壊れないように前方互換で書く)、(2)error イベントや接続断を検知して適切に再試行・通知する、(3)最終的な stop_reason を確認して、max_tokens による途中終了などを見分ける、を意識してください。長時間かかる大きなリクエストではストリーミングが実質的に推奨される場面もあります。表示だけでなくログや後段処理に使う場合は、ストリーム終了後に組み立てた完成メッセージを正として扱うのが安全です。