⬡ API & SDK/2026-03-31上級

Claude API ストリーミング × リアルタイムチャットUI 本番実装ガイド

Claude APIのストリーミングレスポンスとServer-Sent Eventsを活用し、Next.js App RouterでリアルタイムチャットUIを本番品質で組み上げる手順を、SSEプロトコル・Tool Use統合・接続復旧・コスト最適化まで通しで整理します

streaming¹⁶ api³⁸ realtime³ chat-ui next-js sse² production⁸⁷

✦ プレミアム記事

ストリーミングが解く課題と複雑さ

Claude APIのストリーミング機能は、従来の「リクエスト→全応答待機→表示」という同期的なフローを、「リクエスト→部分的応答をリアルタイムで受け取り→UI更新」という体験へと変革します。特にチャットアプリケーションでは、ユーザーが「AIが考えている」様子をリアルタイムで目撃することで、エンゲージメントと信頼度が劇的に向上します。

しかし、ストリーミングは実装の複雑さをもたらします。接続の途中切断、タイムアウト、Tool Use（関数呼び出し）中の状態管理、レート制限の監視など、本番環境では非機能要件が急増します。

私は2014年から iOS/Android の個人開発を続けており、累計5,000万ダウンロードに到達するまでに、ネットワーク越しのリアルタイム体験を何度も実装してきました。Claude API ストリーミングは、その延長線上にありながらも、**「途中で AI の思考が中断する」「Tool Use を挟む」**という独特の難所を持っています。本ガイドは、その難所を実運用の観点から潰していく構成です。

Claude APIストリーミングの仕組み

Server-Sent Events（SSE）プロトコル

Claude APIのストリーミングはHTTP標準のServer-Sent Events（SSE）を採用しています。SSEはHTTP接続を保持したまま、サーバーから複数のイベントをクライアントへ継続的に送信する仕組みです。

SSEの特徴：

HTTP/1.1標準のテキストベースプロトコル
ブラウザのEventSourceで自動再接続（ハートビート機構あり）
WebSocketより実装が簡潔（但し双方向非対応）
ファイアウォール・プロキシ・CDLとの相性良好

Claude APIのイベントストリーム構造

Claude APIのストリーミングレスポンスは以下のイベント列で構成されます：

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

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "こんにちは"}}

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "。本日のお"}}

event: message_delta
data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 42}}

event: message_stop
data: {"type": "message_stop"}

各イベントの役割：

message_start：メッセージの開始。model、usage（入力トークン）を含む
content_block_start：テキスト/Tool Useのブロック開始
content_block_delta：テキストまたはTool Use入力のデルタ（差分）
content_block_stop：ブロック終了
message_delta：メッセージレベルの更新（stop_reason、出力トークン数）
message_stop：メッセージ完全終了（接続を安全にクローズ可）

実装上の重要ポイント

イベント順序への依存禁止: イベント到着順序を仮定した実装は脆弱です。常にindexフィールドに基づいてコンテンツを再構成してください。

Tool Useの段階: ストリーミング中にTool Use（tool_useコンテンツタイプ）が発生した場合、Tool Use完全性待機 → ツール実行 → 続きのストリーミングというフローになります。

✦

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

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦個人運用での実測値（P50/P95レイテンシ・1日あたりトークン消費・キャッシュヒット率）に基づいたストリーミング最適化の判断軸

✦途中切断・Tool Use 連鎖・レート制限が同時に起きる本番障害を、再接続キュー＋指数バックオフ＋会話履歴リプレイで吸収する設計

✦Cloudflare Workers × Edge ランタイムで Claude API ストリームを中継する際に、フロント側 EventSource のバッファ詰まりとデバウンスを避ける具体パターン

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

Next.js App Router でのストリーミングAPI実装

Route Handler でSSEエンドポイントを構築

// app/api/chat/route.ts
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});
 
export const runtime = "nodejs";
 
export async function POST(request: Request) {
  const { messages, systemPrompt } = await request.json();
 
  // レスポンスストリームを読み取り、SSEで返却
  const stream = await client.messages.stream({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 1024,
    system: systemPrompt || "You are a helpful assistant.",
    messages: messages,
  });
 
  // ReadableStream を使って SSE フォーマットで返却
  const readable = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          // チャンク全体をイベント形式でSSE送信
          const data = JSON.stringify(chunk);
          controller.enqueue(
            `event: ${chunk.type}\ndata: ${data}\n\n`
          );
        }
        controller.close();
      } catch (error) {
        controller.error(error);
      }
    },
  });
 
  return new Response(readable, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
      "X-Accel-Buffering": "no",
    },
  });
}

重要なレスポンスヘッダ：

Content-Type: text/event-stream：SSE形式を宣言
Cache-Control: no-cache：キャッシュ禁止（リアルタイム性保証）
X-Accel-Buffering: no：Nginx等のバッファリング無効化（重要）
Connection: keep-alive：接続持続

クライアント側：EventSource でストリーム受信

// lib/useChatStream.ts
"use client";
 
import { useCallback, useRef, useState } from "react";
 
interface ChatMessage {
  role: "user" | "assistant";
  content: string;
}
 
interface UseStreamResult {
  messages: ChatMessage[];
  isLoading: boolean;
  error: string | null;
  sendMessage: (userMessage: string) => Promise<void>;
}
 
export function useChatStream(): UseStreamResult {
  const [messages, setMessages] = useState<ChatMessage[]>([]);
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);
  const eventSourceRef = useRef<EventSource | null>(null);
 
  const sendMessage = useCallback(
    async (userMessage: string) => {
      if (!userMessage.trim()) return;
 
      // ユーザーメッセージをUIに追加
      setMessages((prev) => [...prev, { role: "user", content: userMessage }]);
      setIsLoading(true);
      setError(null);
 
      try {
        // ストリーミングリクエストを開始
        const systemPrompt =
          "You are a helpful AI assistant engaged in a real-time chat conversation.";
        const requestBody = {
          messages: [...messages, { role: "user", content: userMessage }],
          systemPrompt,
        };
 
        const response = await fetch("/api/chat", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(requestBody),
        });
 
        if (!response.ok) {
          throw new Error(`API error: ${response.status}`);
        }
 
        let assistantMessage = "";
 
        // SSEストリームをEventSourceで受信
        if (!response.body) throw new Error("No response body");
 
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let buffer = "";
 
        // アシスタント メッセージプレースホルダーを追加
        setMessages((prev) => [...prev, { role: "assistant", content: "" }]);
 
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
 
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split("\n");
 
          // 最後の不完全な行はバッファに残す
          buffer = lines.pop() || "";
 
          for (const line of lines) {
            if (line.startsWith("data: ")) {
              try {
                const data = JSON.parse(line.slice(6));
 
                // content_block_delta イベント時にテキストを追加
                if (
                  data.type === "content_block_delta" &&
                  data.delta?.type === "text_delta"
                ) {
                  assistantMessage += data.delta.text;
                  setMessages((prev) => {
                    const updated = [...prev];
                    updated[updated.length - 1] = {
                      ...updated[updated.length - 1],
                      content: assistantMessage,
                    };
                    return updated;
                  });
                }
 
                // message_stop でストリーム完了
                if (data.type === "message_stop") {
                  break;
                }
              } catch (e) {
                console.error("Failed to parse SSE data:", e);
              }
            }
          }
        }
 
        setIsLoading(false);
      } catch (err) {
        setError(
          err instanceof Error ? err.message : "Unknown error occurred"
        );
        setIsLoading(false);
        setMessages((prev) => prev.slice(0, -1)); // アシスタントメッセージを削除
      }
    },
    [messages]
  );
 
  return { messages, isLoading, error, sendMessage };
}

リアルタイムチャットUIコンポーネント

チャットメッセージコンポーネント

// components/ChatMessage.tsx
import React from "react";
import { ReactMarkdown } from "react-markdown";
 
interface ChatMessageProps {
  role: "user" | "assistant";
  content: string;
  isLoading?: boolean;
}
 
export function ChatMessage({
  role,
  content,
  isLoading = false,
}: ChatMessageProps) {
  const isUser = role === "user";
 
  return (
    <div
      className={`flex gap-4 mb-4 ${isUser ? "flex-row-reverse" : "flex-row"}`}
    >
      {/* アバター */}
      <div
        className={`flex-shrink-0 w-8 h-8 rounded-full flex items-center justify-center ${
          isUser ? "bg-blue-600" : "bg-gray-300"
        }`}
      >
        {isUser ? "👤" : "🤖"}
      </div>
 
      {/* メッセージバブル */}
      <div
        className={`flex-1 max-w-2xl rounded-lg px-4 py-2 ${
          isUser
            ? "bg-blue-600 text-white rounded-br-none"
            : "bg-gray-100 text-gray-900 rounded-bl-none"
        }`}
      >
        {isLoading ? (
          <div className="flex gap-2">
            <span className="inline-block w-2 h-2 bg-current rounded-full animate-bounce"></span>
            <span className="inline-block w-2 h-2 bg-current rounded-full animate-bounce delay-100"></span>
            <span className="inline-block w-2 h-2 bg-current rounded-full animate-bounce delay-200"></span>
          </div>
        ) : (
          <ReactMarkdown className="prose prose-sm">{content}</ReactMarkdown>
        )}
      </div>
    </div>
  );
}

チャットコンテナコンポーネント

// components/ChatContainer.tsx
"use client";
 
import { useRef, useEffect, useState } from "react";
import { useChatStream } from "@/lib/useChatStream";
import { ChatMessage } from "./ChatMessage";
import { ChatInput } from "./ChatInput";
 
export function ChatContainer() {
  const { messages, isLoading, error, sendMessage } = useChatStream();
  const scrollEndRef = useRef<HTMLDivElement>(null);
 
  // 自動スクロール：新メッセージが追加されたら最下部へ
  useEffect(() => {
    scrollEndRef.current?.scrollIntoView({ behavior: "smooth" });
  }, [messages]);
 
  return (
    <div className="flex flex-col h-screen bg-white">
      {/* メッセージリスト */}
      <div className="flex-1 overflow-y-auto p-4 space-y-4">
        {messages.length === 0 ? (
          <div className="flex items-center justify-center h-full text-gray-500">
            <p>メッセージを送信して会話を始めてください</p>
          </div>
        ) : (
          <>
            {messages.map((msg, idx) => (
              <ChatMessage
                key={idx}
                role={msg.role}
                content={msg.content}
                isLoading={isLoading && msg.role === "assistant" && msg === messages[messages.length - 1]}
              />
            ))}
            <div ref={scrollEndRef} />
          </>
        )}
      </div>
 
      {/* エラー表示 */}
      {error && (
        <div className="bg-red-100 border border-red-400 text-red-700 px-4 py-3 rounded relative">
          {error}
        </div>
      )}
 
      {/* 入力フィールド */}
      <ChatInput
        onSend={sendMessage}
        disabled={isLoading}
        placeholder="メッセージを入力..."
      />
    </div>
  );
}

エラーハンドリングとリトライ戦略

接続エラーの自動リトライ

// lib/retryWithExponentialBackoff.ts
interface RetryOptions {
  maxAttempts?: number;
  initialDelayMs?: number;
  maxDelayMs?: number;
  backoffMultiplier?: number;
}
 
export async function retryWithExponentialBackoff<T>(
  fn: () => Promise<T>,
  options: RetryOptions = {}
): Promise<T> {
  const {
    maxAttempts = 5,
    initialDelayMs = 1000,
    maxDelayMs = 30000,
    backoffMultiplier = 2,
  } = options;
 
  let lastError: Error | null = null;
 
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error instanceof Error ? error : new Error(String(error));
 
      // 最後の試行で失敗したら即座にエラー返却
      if (attempt === maxAttempts) {
        throw lastError;
      }
 
      // リトライ不可なエラー（4xx）はすぐに失敗
      if (
        error instanceof Error &&
        error.message.includes("400") ||
        error.message.includes("401") ||
        error.message.includes("403")
      ) {
        throw lastError;
      }
 
      // 指数バックオフでリトライ待機
      const delay = Math.min(
        initialDelayMs * Math.pow(backoffMultiplier, attempt - 1),
        maxDelayMs
      );
 
      // ジッターを追加してスターダスト（thundering herd）を回避
      const jitter = Math.random() * delay * 0.1;
 
      console.log(
        `Retry attempt ${attempt}/${maxAttempts} after ${Math.round(delay + jitter)}ms`
      );
 
      await new Promise((resolve) =>
        setTimeout(resolve, delay + jitter)
      );
    }
  }
 
  throw lastError || new Error("Unknown error occurred");
}

APIリクエストのタイムアウト管理

// lib/fetchWithTimeout.ts
interface FetchOptions extends RequestInit {
  timeoutMs?: number;
}
 
export async function fetchWithTimeout(
  url: string,
  options: FetchOptions = {}
): Promise<Response> {
  const { timeoutMs = 30000, ...fetchOptions } = options;
 
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
 
  try {
    const response = await fetch(url, {
      ...fetchOptions,
      signal: controller.signal,
    });
 
    return response;
  } catch (error) {
    if (error instanceof Error && error.name === "AbortError") {
      throw new Error(`Request timed out after ${timeoutMs}ms`);
    }
    throw error;
  } finally {
    clearTimeout(timeoutId);
  }
}

拡張されたチャットフック（エラーハンドリング付き）

// lib/useChatStreamWithRetry.ts
"use client";
 
import { useCallback, useState } from "react";
import { retryWithExponentialBackoff } from "./retryWithExponentialBackoff";
import { fetchWithTimeout } from "./fetchWithTimeout";
 
export function useChatStreamWithRetry() {
  const [messages, setMessages] = useState<Array<{ role: string; content: string }>>([]);
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);
 
  const sendMessage = useCallback(async (userMessage: string) => {
    if (!userMessage.trim()) return;
 
    setMessages((prev) => [...prev, { role: "user", content: userMessage }]);
    setIsLoading(true);
    setError(null);
 
    try {
      await retryWithExponentialBackoff(
        async () => {
          const response = await fetchWithTimeout("/api/chat", {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({
              messages: [...messages, { role: "user", content: userMessage }],
            }),
            timeoutMs: 60000,
          });
 
          if (!response.ok) {
            throw new Error(`HTTP ${response.status}: ${await response.text()}`);
          }
 
          let assistantMessage = "";
          setMessages((prev) => [...prev, { role: "assistant", content: "" }]);
 
          if (!response.body) throw new Error("No response body");
 
          const reader = response.body.getReader();
          const decoder = new TextDecoder();
          let buffer = "";
 
          while (true) {
            const { done, value } = await reader.read();
            if (done) break;
 
            buffer += decoder.decode(value, { stream: true });
            const lines = buffer.split("\n");
            buffer = lines.pop() || "";
 
            for (const line of lines) {
              if (line.startsWith("data: ")) {
                try {
                  const data = JSON.parse(line.slice(6));
 
                  if (data.type === "content_block_delta" && data.delta?.type === "text_delta") {
                    assistantMessage += data.delta.text;
                    setMessages((prev) => {
                      const updated = [...prev];
                      updated[updated.length - 1].content = assistantMessage;
                      return updated;
                    });
                  }
 
                  if (data.type === "message_stop") {
                    return;
                  }
                } catch (e) {
                  console.error("Parse error:", e);
                }
              }
            }
          }
        },
        { maxAttempts: 3, initialDelayMs: 2000 }
      );
 
      setIsLoading(false);
    } catch (err) {
      const errorMsg = err instanceof Error ? err.message : "Unknown error";
      setError(errorMsg);
      setIsLoading(false);
      setMessages((prev) => prev.slice(0, -1));
    }
  }, [messages]);
 
  return { messages, isLoading, error, sendMessage };
}

Tool Use × ストリーミング統合

Tool Use ストリーミングの構造

Tool Useが発生した場合、ストリーミングレスポンスは以下のパターンになります：

event: content_block_start
data: {"type": "content_block_start", "index": 1, "content_block": {"type": "tool_use", "id": "tool_use_1", "name": "search", "input": {}}}

event: content_block_delta
data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": "{\"query\""}}

event: content_block_delta
data: {"type": "content_block_delta", "index": 1, "delta": {"type": "input_json_delta", "partial_json": ": \"Claude API streaming\""}}

event: content_block_stop
data: {"type": "content_block_stop", "index": 1}

event: message_stop
data: {"type": "message_stop"}

Tool Use対応のチャットフック

// lib/useChatWithTools.ts
"use client";
 
import { useState, useCallback } from "react";
 
interface ToolDefinition {
  name: string;
  description: string;
  input_schema: {
    type: string;
    properties: Record<string, unknown>;
    required: string[];
  };
}
 
interface ContentBlock {
  type: "text" | "tool_use";
  text?: string;
  id?: string;
  name?: string;
  input?: Record<string, unknown>;
}
 
export function useChatWithTools(tools: ToolDefinition[]) {
  const [messages, setMessages] = useState<Array<{ role: string; content: unknown }>>([]);
  const [isLoading, setIsLoading] = useState(false);
 
  const sendMessage = useCallback(
    async (userMessage: string) => {
      setMessages((prev) => [...prev, { role: "user", content: userMessage }]);
      setIsLoading(true);
 
      try {
        // ストリーミングレスポンスを受信
        const response = await fetch("/api/chat-with-tools", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            messages: [...messages, { role: "user", content: userMessage }],
            tools,
          }),
        });
 
        let contentBlocks: ContentBlock[] = [];
        let currentBlock: Partial<ContentBlock> | null = null;
        let toolInputBuffer = "";
 
        if (!response.body) throw new Error("No response body");
 
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let buffer = "";
 
        setMessages((prev) => [...prev, { role: "assistant", content: contentBlocks }]);
 
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
 
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split("\n");
          buffer = lines.pop() || "";
 
          for (const line of lines) {
            if (!line.startsWith("data: ")) continue;
 
            try {
              const data = JSON.parse(line.slice(6));
 
              if (data.type === "content_block_start") {
                currentBlock = { type: data.content_block.type };
                if (data.content_block.type === "tool_use") {
                  currentBlock.id = data.content_block.id;
                  currentBlock.name = data.content_block.name;
                  currentBlock.input = {};
                  toolInputBuffer = "";
                }
              } else if (data.type === "content_block_delta") {
                if (data.delta.type === "text_delta" && currentBlock?.type === "text") {
                  currentBlock.text = (currentBlock.text || "") + data.delta.text;
                } else if (
                  data.delta.type === "input_json_delta" &&
                  currentBlock?.type === "tool_use"
                ) {
                  toolInputBuffer += data.delta.partial_json;
                  try {
                    currentBlock.input = JSON.parse(toolInputBuffer);
                  } catch {
                    // JSON未完成、次のチャンク待機
                  }
                }
              } else if (data.type === "content_block_stop") {
                if (currentBlock) {
                  contentBlocks.push(currentBlock as ContentBlock);
                  currentBlock = null;
                }
              }
 
              if (data.type === "message_stop") {
                // Tool Use検出：ツール実行 → 続きのストリーム
                const toolUses = contentBlocks.filter((b) => b.type === "tool_use");
                if (toolUses.length > 0) {
                  // ツール実行API呼び出し
                  const toolResults = await Promise.all(
                    toolUses.map(async (tu) => {
                      const result = await executeToolLocally(tu.name!, tu.input!);
                      return { type: "tool_result", tool_use_id: tu.id, content: result };
                    })
                  );
 
                  // ツール結果を含めて続きのストリーミング
                  const continueMessages = [
                    ...messages,
                    { role: "user", content: userMessage },
                    { role: "assistant", content: contentBlocks },
                    { role: "user", content: toolResults },
                  ];
 
                  // 再帰的にストリーミング続行...
                }
                break;
              }
            } catch (e) {
              console.error("Parse error:", e);
            }
          }
        }
 
        setIsLoading(false);
      } catch (error) {
        setIsLoading(false);
        setMessages((prev) => prev.slice(0, -1));
      }
    },
    [messages, tools]
  );
 
  return { messages, isLoading, sendMessage };
}
 
// ツール実行関数（ローカル実装の例）
async function executeToolLocally(
  name: string,
  input: Record<string, unknown>
): Promise<string> {
  if (name === "search") {
    const query = input.query as string;
    // 実装: Webサーチ、DB検索等
    return `Search results for "${query}": ...`;
  }
  return "Tool not implemented";
}

レート制限とコスト最適化

トークン使用量のトラッキング

// lib/tokenTracker.ts
interface TokenUsage {
  inputTokens: number;
  outputTokens: number;
  totalTokens: number;
  estimatedCostUSD: number;
}
 
const TOKEN_PRICES = {
  "claude-3-5-sonnet-20241022": {
    input: 0.003, // $3 per 1M input tokens
    output: 0.015, // $15 per 1M output tokens
  },
};
 
export function calculateTokenCost(
  model: string,
  inputTokens: number,
  outputTokens: number
): TokenUsage {
  const prices = TOKEN_PRICES[model as keyof typeof TOKEN_PRICES] || {
    input: 0.003,
    output: 0.015,
  };
 
  const inputCost = (inputTokens / 1_000_000) * prices.input;
  const outputCost = (outputTokens / 1_000_000) * prices.output;
 
  return {
    inputTokens,
    outputTokens,
    totalTokens: inputTokens + outputTokens,
    estimatedCostUSD: inputCost + outputCost,
  };
}
 
// 使用例
const usage: TokenUsage = calculateTokenCost(
  "claude-3-5-sonnet-20241022",
  15000,
  5000
);
console.log(`Cost: $${usage.estimatedCostUSD.toFixed(4)}`);
// 出力: Cost: $0.0825

レート制限検出とキューイング

// lib/rateLimitQueue.ts
interface QueuedRequest {
  id: string;
  fn: () => Promise<Response>;
  retries: number;
}
 
export class RateLimitQueue {
  private queue: QueuedRequest[] = [];
  private processing = false;
  private requestsPerMinute = 50; // Claude API: 50 RPM for tier-1
  private requestTimestamps: number[] = [];
 
  async enqueue(fn: () => Promise<Response>): Promise<Response> {
    return new Promise((resolve, reject) => {
      const request: QueuedRequest = {
        id: Math.random().toString(36),
        fn,
        retries: 0,
      };
 
      this.queue.push(request);
      this.process().catch(reject);
    });
  }
 
  private async process(): Promise<void> {
    if (this.processing) return;
    this.processing = true;
 
    while (this.queue.length > 0) {
      const request = this.queue.shift()!;
      const now = Date.now();
 
      // 1分以内のリクエスト数を確認
      this.requestTimestamps = this.requestTimestamps.filter(
        (ts) => now - ts < 60000
      );
 
      if (this.requestTimestamps.length >= this.requestsPerMinute) {
        // レート制限に達した：待機
        const oldestRequest = this.requestTimestamps[0];
        const waitMs = 60000 - (now - oldestRequest);
        console.log(`Rate limit reached. Waiting ${waitMs}ms...`);
        await new Promise((resolve) => setTimeout(resolve, waitMs));
      }
 
      try {
        const response = await request.fn();
 
        // 429（Too Many Requests）の場合
        if (response.status === 429) {
          const retryAfter =
            response.headers.get("Retry-After") || "60";
          const delayMs = parseInt(retryAfter) * 1000;
 
          console.warn(
            `Rate limited. Retrying after ${delayMs}ms...`
          );
 
          await new Promise((resolve) =>
            setTimeout(resolve, delayMs)
          );
 
          // キューの先頭に戻す
          if (request.retries < 3) {
            request.retries++;
            this.queue.unshift(request);
          } else {
            throw new Error("Max retries exceeded for rate limit");
          }
        } else {
          this.requestTimestamps.push(now);
        }
      } catch (error) {
        console.error("Request failed:", error);
        throw error;
      }
    }
 
    this.processing = false;
  }
}
 
// 使用例
const queue = new RateLimitQueue();
const response = await queue.enqueue(() =>
  fetch("/api/chat", {
    method: "POST",
    body: JSON.stringify({ /* ... */ }),
  })
);

本番デプロイのベストプラクティス

ストリーミング対応の Cloudflare Workers ルーター

// wrangler.toml 設定例
[env.production]
name = "my-chat-app-prod"
routes = [
  { pattern = "https://example.com/api/chat*", zone_name = "example.com" }
]
 
[build]
command = "npm run build"
cwd = "."
 
# Streaming API用のタイムアウト延長
limits = { cpu_seconds = 30 }

キャッシュとストリーミングの共存

// cache-worker.ts
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // ストリーミングAPI（/api/chat）はキャッシュをバイパス
    if (request.url.includes("/api/chat")) {
      return fetch(request, {
        cf: {
          cacheEverything: false,
          cacheTtl: 0,
        },
      });
    }
 
    // 静的資産はキャッシュ
    return fetch(request, {
      cf: {
        cacheEverything: true,
        cacheTtl: 86400, // 24h
      },
    });
  },
};

本番ログとモニタリング

// lib/logging.ts
interface StripeLogEntry {
  timestamp: string;
  level: "info" | "warn" | "error";
  event: string;
  userId?: string;
  duration?: number;
  tokenCount?: number;
  error?: string;
}
 
export class ChatLogger {
  private logs: StripeLogEntry[] = [];
 
  log(entry: Omit<StripeLogEntry, "timestamp">) {
    const logEntry: StripeLogEntry = {
      timestamp: new Date().toISOString(),
      ...entry,
    };
 
    this.logs.push(logEntry);
 
    // ログサービスに送信（CloudflareログpushやDatadog等）
    if (process.env.NODE_ENV === "production") {
      this.sendToLoggingService(logEntry);
    }
  }
 
  private sendToLoggingService(entry: StripeLogEntry) {
    // 実装例：CloudflareログpushへのPOST
    fetch("https://logs.example.com/api/logs", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(entry),
    }).catch((e) => console.error("Logging failed:", e));
  }
}
 
export const logger = new ChatLogger();

使用例：

logger.log({
  level: "info",
  event: "stream_completed",
  userId: "user_123",
  duration: 3500,
  tokenCount: 1200,
});

公式ドキュメントに書かれていない運用知見

ここからは、Anthropic 公式ドキュメントには明記されていない、私自身が個人開発の現場でストリーミングを運用して得た知見をまとめます。教科書通りの実装でも、本番でハマるポイントはこのあたりに集中します。

接続復旧は「再ストリーム」ではなく「会話履歴リプレイ」で組む

「途中で切れたら続きから再開する」という発想で実装すると、Claude API の仕様上どうしても無理が出ます。私の運用では、次の判断基準で割り切っています。

出力トークンが 概ね 50 トークン未満で切断 された場合 → クライアント側でメッセージごと破棄し、ユーザーには何も表示せず無音で再リクエスト
50 トークン以上が表示済みで切断された場合 → 表示済み部分を assistant メッセージとして会話履歴に挿入し、システムプロンプトに「直前の応答が途中で切れたため、続きを書いてください」と短く付け加えて再ストリーム

この方式に切り替えてから、私が運用しているチャット系プロトタイプでは、ユーザーが切断に気づく確率が体感で 3 割→1 割未満まで下がりました。

Tool Use 連鎖は「2 段目以降のキャンセル可能性」を必ず実装する

Tool Use が 1 リクエストで完結することは本番ではほぼなく、多くの場合は 2〜3 段の連鎖になります。途中でユーザーが「もうこの方向じゃない」と思ったとき、現在進行中の Tool Use を安全にキャンセルできる経路を最初から組み込んでおく必要があります。

実装の骨格としては、サーバー側 Route Handler 内で AbortController を messages.stream() に渡し、クライアントからの DELETE リクエストでサーバー側の AbortController.abort() を呼ぶ構造にします。途中までの Tool Use 結果は破棄し、それまでに到着していた content_block_delta のテキスト部分だけを「ここまでの応答」として履歴に残します。これを忘れると、ユーザーは「止めるボタンが効かないアプリ」という印象を持ち、リピート率がはっきり下がります。

レイテンシは P50 ではなく P95 で監視する

ストリーミングの体感品質は、初トークン到着までの TTFB（Time To First Byte） で決まります。私の壁紙・癒し系アプリ群（AdMob 経由のマネタイズが中心）でも、ユーザーは「途中の遅さ」より「最初の沈黙」のほうを強く嫌います。

私が実際に運用している Cloudflare Workers + Claude API のエッジ中継構成での実測値は、おおむね次のレンジに収まっています。

TTFB P50: 約 380〜450 ms（同一リージョン、システムプロンプト約 600 トークン）
TTFB P95: 約 900〜1,200 ms（リトライ 1 回込み）
接続維持率（90 秒以上のストリーム）: 98.7%
中断時の自動再接続成功率（上述のリプレイ方式）: 94%

P50 だけで監視していると、P95 の沈黙が混じるユーザーの 5% が静かに離脱します。初トークンに 1.2 秒以上かかったセッションは、専用のログラインで赤くマークし、原因を Anthropic ステータス・自前ネットワーク・プロンプトサイズの 3 軸で分けて記録するのがおすすめです。

コストは「キャッシュヒット率」と「停止トークン分布」の 2 指標で抑える

ストリーミング運用で青ざめるのは、想定の 2〜3 倍の請求が月末に届いたときです。私が安定運用で意識している指標は次の 2 つです。

キャッシュヒット率: 直近 7 日の同一プロンプトベース部分について、Anthropic のプロンプトキャッシュが効いた割合。50% を切り続けたらシステムプロンプトの構造を疑う
停止トークン分布: message_delta.stop_reason の end_turn / max_tokens / tool_use / stop_sequence の比率。max_tokens 切断が 15% を超えるとユーザーから「途中で切れる」というクレームが増えるので、max_tokens を上げるか、応答テンプレートを短く誘導する

この 2 指標は、Claude Code から wrangler tail で流れるアクセスログに直接 console.info で吐き出し、別途 Workers Analytics Engine に書き込んで日次集計しています。Anthropic の Usage ダッシュボードは月次のざっくり値しか出ないので、自前で日次まで落とさないと判断が遅れます。

Cloudflare Workers × EventSource のバッファ詰まりを避ける

エッジ中継で Claude API ストリームを受け、フロントの EventSource に流す構成にすると、チャンクが 4KB 単位でまとめて飛ぶ現象に当たることがあります。これは Workers の Response ストリームが内部でバッファされるためで、放置すると「ぬるっと一気に表示される」体験になり、ストリーミングの意味が薄れます。

私は次の対策で安定させています。

レスポンスヘッダに Cache-Control: no-cache, no-transform と X-Accel-Buffering: no を必ず付ける
TransformStream を使い、controller.enqueue() の直後に await new Promise((r) => setTimeout(r, 0)) を 1 ティック挟む
5〜10 トークンごとに、event: ping\ndata: {}\n\n のハートビートを別途送る（プロキシ側のアイドルクローズ回避にも有効）

このあたりは、廣川（私）が運営している壁紙・癒し系アプリのオンライン辞書機能を Claude ベースに置き換えたとき、最後まで残った地味な詰まりでした。1997 年に独学でインターネットに触れた頃から、こういう「プロトコルの隙間」を埋める作業がいちばん楽しい瞬間です。

ストリーミングを止める判断軸

最後に、「ストリーミングをやめて非ストリーム応答に倒すべき」シーンも書き残しておきます。

1 回の応答が 300 トークン未満で安定している（タイトル生成・要約短文）→ ストリーミングの体感メリットがほぼゼロ、レイテンシだけ悪化する
バックエンドで応答全文に対して 後処理（JSON Schema 検証・正規化） をかける必要がある → 非ストリーム + 結果を返すほうが実装も品質も安定
オフラインで結果をキャッシュして何度も配る用途（FAQ ボット、固定応答）→ 一度生成して KV に格納、ストリーム不要

「全部ストリーミング」は実装コストとデバッグコストが高くつきます。ユーザーが沈黙を嫌がる画面だけにストリーミングを残すのが、長く運用するうえで結局いちばん安いという結論に至りました。

本番運用で押さえるべき要点

Claude API ストリーミング × リアルタイムチャット UI を実運用に乗せるとき、私が毎回チェックしているのは次の点です。

プロトコル: SSE の構造と index ベースの再構成、message_delta.stop_reason の網羅処理
接続管理: 50 トークン境界での再ストリーム判定、AbortController でのキャンセル経路、ハートビート
Tool Use: 連鎖の各段でのキャンセル可能性、中間結果の履歴保存
可観測性: TTFB P95、接続維持率、キャッシュヒット率、停止トークン分布を日次で見られる状態
エッジ中継: Cloudflare Workers での X-Accel-Buffering: no、TransformStream の 1 ティック遅延、ハートビート