ストリーミングレスポンスの実装 — Claude API でリアルタイム応答を実現する

ストリーミングを初めて自分のアプリに組み込んだときの驚きは、いまでもよく覚えています。それまでは応答が完成するまで画面が数十秒沈黙し、試してくれた方から「固まったのかと思って閉じました」と言われたこともありました。トークンが1つずつ流れ始めた途端、処理時間そのものは変わらないのに、体感はまるで別物になります。個人開発では応答待ちの数十秒が離脱に直結するだけに、この差は無視できません。ここでは Python / TypeScript でストリーミングを組み込む手順と、動かして初めて見える注意点を、実装で確かめた順に並べました。

ストリーミングとは

ストリーミングを使用すると、完全な応答を待つ代わりに、Claude の応答をトークンごとにリアルタイムで受信できます。これはユーザーに即座のフィードバックを提供し、反応の良いアプリケーションを構築するために不可欠です。

ℹ️

ストリーミングは特に以下の場合に価値があります: - チャットアプリケーション (応答が生成されているのを表示) - 長編コンテンツ (ニュース記事、レポート) - リアルタイムダッシュボード - 帯域幅が限られたモバイルアプリケーション - 認識される応答時間を改善

Server-Sent Events (SSE) を理解する

ストリーミングレスポンスは Server-Sent Events (SSE) プロトコルを使用します。このプロトコルはサーバーからクライアントへイベントストリームとしてデータを送信します。

SSE フォーマット

data: {"type":"content_block_start","index":0,"content_block":{"type":"text"}}

data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" world"}}

data: {"type":"message_stop"}

各イベントは、イベントの種類を示す type フィールドを持つ JSON オブジェクトです。

Python ストリーミング実装

基本的なストリーミング例

import anthropic
 
client = anthropic.Anthropic()
 
messages = [
    {
        "role": "user",
        "content": "Write a short story about a robot learning to paint."
    }
]
 
# stream=True を使用してストリーミングを有効にする
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=messages
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()  # 完全な応答後の改行

モデル名には 2026年7月時点の既定である claude-sonnet-5 を指定しています。2026年6月30日に公開された Sonnet 5 は導入価格（100万トークンあたり入力 $2・出力 $10、2026年8月31日まで）で利用でき、Opus 4.8 と比べて入出力とも4割ほど安く、ストリーミングを多用するチャット用途でコストを抑えやすい選択です。長時間の複雑なエージェント処理には claude-opus-4-8 を検討してください。

詳細なイベント処理

より細かく制御するために、個々のイベントを処理できます:

import anthropic
 
client = anthropic.Anthropic()
 
messages = [
    {
        "role": "user",
        "content": "Explain quantum computing in simple terms."
    }
]
 
# イベントを個別に処理
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=messages
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            print(f"Starting content block {event.index}")
 
        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
 
        elif event.type == "content_block_stop":
            print(f"\nFinished content block {event.index}")
 
        elif event.type == "message_stop":
            print("\nStream finished")

ストリーミング中の完全な応答の収集

import anthropic
 
client = anthropic.Anthropic()
 
def stream_message(user_input: str) -> str:
    """Stream a message and return the full response"""
    full_response = ""
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_input}]
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)
 
    print()  # 最終改行
    return full_response
 
# 関数を使用
result = stream_message("What are the benefits of renewable energy?")
print(f"\nFull response ({len(result)} characters)")

TypeScript/JavaScript ストリーミング

公式 SDK を使用

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function streamResponse() {
  const stream = client.messages.stream({
    model: "claude-sonnet-5",
    max_tokens: 1024,
    messages: [
      {
        role: "user",
        content: "Write a haiku about the moon."
      }
    ]
  });
 
  // テキスト イベントを処理
  stream.on("text", (text: string) => {
    process.stdout.write(text);
  });
 
  // 最終メッセージを取得
  const finalMessage = await stream.finalMessage();
  console.log("\n\nFinal message:", finalMessage);
}
 
streamResponse();

手動イベント処理

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function processStreamEvents() {
  const stream = await client.messages.create({
    model: "claude-sonnet-5",
    max_tokens: 1024,
    stream: true,
    messages: [
      {
        role: "user",
        content: "Describe the water cycle."
      }
    ]
  });
 
  for await (const event of stream) {
    switch (event.type) {
      case "content_block_start":
        console.log(`Starting block ${event.index}`);
        break;
 
      case "content_block_delta":
        if (event.delta.type === "text_delta") {
          process.stdout.write(event.delta.text);
        }
        break;
 
      case "content_block_stop":
        console.log(`\nFinished block ${event.index}`);
        break;
 
      case "message_stop":
        console.log("Stream complete");
        break;
    }
  }
}
 
processStreamEvents();

ウェブベースのストリーミング (フロントエンド)

Fetch API を使用したストリーミング

async function streamFromAPI(userMessage) {
  const response = await fetch("https://api.anthropic.com/v1/messages", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "x-api-key": process.env.ANTHROPIC_API_KEY,
      "anthropic-version": "2023-06-01"
    },
    body: JSON.stringify({
      model: "claude-sonnet-5",
      max_tokens: 1024,
      stream: true,
      messages: [
        { role: "user", content: userMessage }
      ]
    })
  });
 
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let fullResponse = "";
 
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
 
    const chunk = decoder.decode(value);
    const lines = chunk.split("\n");
 
    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const eventData = line.slice(6);
        try {
          const event = JSON.parse(eventData);
 
          if (event.type === "content_block_delta") {
            if (event.delta.type === "text_delta") {
              const text = event.delta.text;
              fullResponse += text;
              // リアルタイムで UI を更新
              document.getElementById("response").textContent = fullResponse;
            }
          }
        } catch (e) {
          // 無効な JSON をスキップ
        }
      }
    }
  }
 
  return fullResponse;
}

React ストリーミングコンポーネント

import { useState } from "react";
 
export function StreamingChat() {
  const [input, setInput] = useState("");
  const [response, setResponse] = useState("");
  const [loading, setLoading] = useState(false);
 
  const handleSubmit = async (e) => {
    e.preventDefault();
    setLoading(true);
    setResponse("");
 
    try {
      const fetchResponse = await fetch("/api/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message: input })
      });
 
      const reader = fetchResponse.body.getReader();
      const decoder = new TextDecoder();
      let fullText = "";
 
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
 
        const chunk = decoder.decode(value);
        const lines = chunk.split("\n");
 
        for (const line of lines) {
          if (line.startsWith("data: ")) {
            try {
              const event = JSON.parse(line.slice(6));
              if (event.type === "content_block_delta" &&
                  event.delta.type === "text_delta") {
                fullText += event.delta.text;
                setResponse(fullText);
              }
            } catch (e) {
              // 無効な JSON をスキップ
            }
          }
        }
      }
    } finally {
      setLoading(false);
    }
  };
 
  return (
    <div>
      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="Enter your message..."
          disabled={loading}
        />
        <button type="submit" disabled={loading}>
          {loading ? "Streaming..." : "Send"}
        </button>
      </form>
      <div id="response">{response}</div>
    </div>
  );
}

ストリームイベントの種類と拾い方

サポートされているイベント

# メッセージ開始イベント
{
  "type": "message_start",
  "message": {
    "id": "msg_123",
    "type": "message",
    "role": "assistant",
    "stop_reason": null
  }
}
 
# コンテンツ ブロック開始
{
  "type": "content_block_start",
  "index": 0,
  "content_block": {
    "type": "text"
  }
}
 
# テキスト デルタ (トークン受信)
{
  "type": "content_block_delta",
  "index": 0,
  "delta": {
    "type": "text_delta",
    "text": "Hello"
  }
}
 
# コンテンツ ブロック停止
{
  "type": "content_block_stop",
  "index": 0
}
 
# メッセージ デルタ (最終統計)
{
  "type": "message_delta",
  "delta": {
    "stop_reason": "end_turn"
  },
  "usage": {
    "output_tokens": 42
  }
}
 
# メッセージ停止
{
  "type": "message_stop"
}

ストリームでのエラー処理

ストリームエラー処理

import anthropic
 
client = anthropic.Anthropic()
 
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
 
except anthropic.APIError as e:
    print(f"API Error: {e}")
    # レート制限、認証エラー などを処理
 
except anthropic.APIConnectionError as e:
    print(f"Connection Error: {e}")
    # ネットワーク エラーを処理
 
except anthropic.RateLimitError as e:
    print(f"Rate Limited: {e}")
    # 指数バックオフを実装

タイムアウト処理

async function streamWithTimeout(
  userMessage: string,
  timeoutMs: number = 30000
) {
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
 
  try {
    const stream = client.messages.stream({
      model: "claude-sonnet-5",
      max_tokens: 1024,
      messages: [{ role: "user", content: userMessage }],
      signal: controller.signal
    });
 
    for await (const event of stream) {
      if (event.type === "content_block_delta" &&
          event.delta.type === "text_delta") {
        console.log(event.delta.text);
      }
    }
  } catch (error) {
    if (error.name === "AbortError") {
      console.error("Stream timeout");
    } else {
      throw error;
    }
  } finally {
    clearTimeout(timeoutId);
  }
}

UI/UX ベストプラクティス

段階的な表示

応答がそのまま表示されるようにします:

<div id="chat">
  <div class="message user">Your question</div>
  <div class="message assistant" id="response">
    <!-- レスポンスがここに表示され、トークンごと -->
  </div>
</div>

ローディングインジケータ

function showLoadingIndicator() {
  const response = document.getElementById("response");
  response.innerHTML = '<span class="loading">Thinking...</span>';
}
 
function updateResponse(token) {
  const response = document.getElementById("response");
  // 最初のトークンで読み込みインジケータを削除
  if (response.textContent.includes("Thinking")) {
    response.textContent = token;
  } else {
    response.textContent += token;
  }
}

ネットワークの問題への対応

import anthropic
import time
 
def stream_with_retry(prompt, max_retries=3):
    client = anthropic.Anthropic()
 
    for attempt in range(max_retries):
        try:
            with client.messages.stream(
                model="claude-sonnet-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            ) as stream:
                for text in stream.text_stream:
                    yield text
            return  # 成功
 
        except anthropic.APIConnectionError as e:
            if attempt == max_retries - 1:
                raise  # 最後の試み
            wait_time = 2 ** attempt  # 指数バックオフ
            print(f"Retry in {wait_time}s...")
            time.sleep(wait_time)

パフォーマンスを引き出すヒント

長い応答には常にストリーミングを使用 — より良い UX と認識される遅延の短縮
デバウンスを実装 — 高速なストリームで各トークンで UI を更新しない
接続プーリングを使用 — 複数のストリームで HTTP 接続を再利用
ストリームヘルスを監視 — エラーと再試行失敗を追跡
合理的なタイムアウトを設定 — ハングしている接続を防止

⚠️

ユーザーに表示する前に完全な応答をバッファリングしないでください。最良のユーザーエクスペリエンスのためにトークンが到着したらすぐにストリーミングしてください。

実装例: ストリーミングチャットアプリケーション

import anthropic
import json
 
def chat_with_streaming(conversation_history: list[dict]):
    """メモリ付きチャット会話をストリーム"""
    client = anthropic.Anthropic()
 
    # システム コンテキストを追加
    messages = [
        {
            "role": "user",
            "content": "You are a helpful assistant."
        }
    ] + conversation_history
 
    full_response = ""
 
    print("Assistant: ", end="", flush=True)
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=messages
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)
 
    print()  # 新しい行
    return full_response
 
# 使用例
history = []
while True:
    user_input = input("You: ")
    if user_input.lower() == "quit":
        break
 
    history.append({"role": "user", "content": user_input})
    assistant_response = chat_with_streaming(history)
    history.append({"role": "assistant", "content": assistant_response})

次のステップ

チャットアプリケーションにストリーミングを実装
ストリーミングを使用してリアルタイムダッシュボードを構築
ストリーミング実装にエラー復旧を追加
高速ストリーム用の UI 更新を最適化

実装を次に進めるなら、App Router での組み込み手順をまとめた Claude API のSSEストリーミングをNext.js App Routerに実装するが続きとして読みやすいはずです。エラーも出さずに止まるストリームの検知と途中再開については、Claude API のストリーミングが「エラーも出さずに」止まるときに運用の詳細をまとめています。

ストリーミングは体験が華やかな反面、個人開発で私自身が実際に組んでみると、つまずくのはたいてい「途中で切れた接続の後始末」でした。Dolice Labs のアプリでも、UI を凝る前にまず再接続とエラー時のフォールバックを固めるようにしています。見栄えは後から足せますが、壊れ方の設計は最初に決めておくほうが楽です。実装では、最後に受け取ったイベントのインデックスを保持しておき、切断時はそこから再開できるようにすると、ユーザーには一瞬の引っかかり程度で復帰でき、体験の毀損を最小限に抑えられます。

検証時に確認すべきこと

本番相当の負荷をかけた状態で再現できるか
ログだけでなくメトリクスにエラー率を出しているか
失敗時の人間への通知が遅延なく届くか

ストリーミングレスポンスの実装 — Claude API でリアルタイム応答を実現する

ストリーミングとは

Server-Sent Events (SSE) を理解する

SSE フォーマット

Python ストリーミング実装

基本的なストリーミング例

詳細なイベント処理

ストリーミング中の完全な応答の収集

TypeScript/JavaScript ストリーミング

公式 SDK を使用

手動イベント処理

ウェブベースのストリーミング (フロントエンド)

Fetch API を使用したストリーミング

React ストリーミングコンポーネント

ストリームイベントの種類と拾い方

サポートされているイベント

ストリームでのエラー処理

ストリームエラー処理

タイムアウト処理

UI/UX ベストプラクティス

段階的な表示

ローディングインジケータ

ネットワークの問題への対応

パフォーマンスを引き出すヒント

実装例: ストリーミングチャットアプリケーション

次のステップ

検証時に確認すべきこと

お読みいただきありがとうございます

関連記事

ストリーミングレスポンスの実装 — Claude API でリアルタイム応答を実現する

ストリーミングとは

Server-Sent Events (SSE) を理解する

SSE フォーマット

Python ストリーミング実装

基本的なストリーミング例

詳細なイベント処理

ストリーミング中の完全な応答の収集

TypeScript/JavaScript ストリーミング

公式 SDK を使用

手動イベント処理

ウェブベースのストリーミング (フロントエンド)

Fetch API を使用したストリーミング

React ストリーミング コンポーネント

ストリーム イベントの種類と拾い方

サポートされているイベント

ストリームでのエラー処理

ストリーム エラー処理

タイムアウト処理

UI/UX ベストプラクティス

段階的な表示

ローディング インジケータ

ネットワークの問題への対応

パフォーマンスを引き出すヒント

実装例: ストリーミング チャット アプリケーション

次のステップ

検証時に確認すべきこと

お読みいただきありがとうございます

関連記事

React ストリーミングコンポーネント

ストリームイベントの種類と拾い方

ストリームエラー処理

ローディングインジケータ

実装例: ストリーミングチャットアプリケーション