⬡ API & SDK/2026-04-26上級

Claude API × WebSocket × Redis Pub/Sub でスケーラブルなリアルタイム AI チャットサーバーを構築する — Node.js 本番設計・マルチユーザー対応・コスト制御の実装

Claude API × WebSocket × Redis Pub/Sub でリアルタイム AI チャットサーバーを本番運用する設計と実装。SSE との判断基準、マルチインスタンス対応、停止遅延がユーザーごとの予算精度をどう崩すかまで、計測コードつきで扱います。

websocketredisnodejsstreaming¹⁶ realtime³ production⁸⁷ api³⁹

✦ プレミアム記事

チャット UI でユーザーが送信ボタンを押してから、AI の返答が少しずつ画面に流れ始める瞬間があります。あの体験を SSE（Server-Sent Events）で実装している方は多いのですが、「ユーザーが途中でメッセージを送り直せるようにしたい」「接続を切らずに複数ターンを管理したい」「入力中インジケーターを双方向で表示したい」という要件が加わった途端、SSE では対応しきれない壁にぶつかります。

私自身、モバイルアプリのバックエンドを設計するときに同じ壁にぶつかりました。SSE は HTTP レスポンスの拡張でしかなく、サーバーからクライアントへの一方向通信です。クライアントからサーバーへのメッセージは、別途 REST API の呼び出しが必要になります。セッションを管理するために状態をどこかに持つ必要が生まれ、複数のインスタンスに水平スケールすると途端に複雑になります。

WebSocket は最初から双方向通信を前提に設計されています。一本の接続で送受信が完結するので、会話の文脈管理がシンプルになります。そして Redis Pub/Sub を組み合わせると、サーバーが複数台になっても特定のユーザー宛てのメッセージを正しいインスタンスに届けられます。

以下は、Claude API のストリーミングを WebSocket に載せ、Redis Pub/Sub で複数インスタンスに広げるまでの設計と実装です。動くコードと、実際に運用してから気づいた前提を添えました。

SSE と WebSocket の使い分け — 設計判断の基準

まずどちらを選ぶべきかの判断基準を整理します。この判断を誤ると後から大きなリファクタが発生します。

SSE で十分なケース:

単発の質問 → 回答のみ（ユーザーが途中で割り込まない）
サーバーからの一方向通知（ニュースフィード、進捗バー）
インフラが HTTP/2 に対応していて、接続数の制限が問題にならない環境です
バックエンドが Cloudflare Workers や Lambda など WebSocket を使いづらい環境

WebSocket が必要なケース:

ユーザーが AI の生成中に「止めて」「別の話題に切り替えて」ができる双方向制御
「ユーザーが入力中」インジケーターをリアルタイムで AI 側に伝えたい
1 つの接続で複数ターンの会話を管理し、ネットワーク確立コストを削減したい
複数ユーザーが同じセッションを共有するコラボレーション機能

今回は後者のケースを想定します。

アーキテクチャ全体設計

構成の全体像は次のとおりです。

[Mobile/Web Client]
      |  WebSocket (ws://)
      |
[Node.js WebSocket Server (複数台)]
      |  Claude API Streaming (HTTPS)
      |  Redis Pub/Sub (セッション間メッセージルーティング)
      |
[Redis] --- [Claude API (Anthropic)]

各コンポーネントの役割は以下のとおりです。

Node.js WebSocket Server: クライアントの接続を受け付け、Claude API にストリーミングリクエストを送り、返ってきたデルタをクライアントへ転送します
Redis（Pub/Sub）: サーバーが複数台になったときに、あるインスタンスに届いたメッセージを正しいインスタンスのユーザーへ届けるためのルーター
Redis（Hash/String）: ユーザーごとのトークン使用量と会話履歴の管理

依存パッケージは以下のとおりです（Node.js 20 以上を推奨）。

npm install ws @anthropic-ai/sdk ioredis jsonwebtoken dotenv
npm install -D typescript @types/ws @types/node tsx

✦

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

この記事の続きを読む

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

この記事で得られること

✦停止要求から生成が止まるまでの遅延を測る計装コードと、遅延100ms／500ms／2,000ms が月3,000回の中断でいくらの差になるかの試算

✦Redis Pub/Sub に停止指示だけを乗せると再起動中のインスタンスで消える理由と、ローカル abort を優先する回避策

✦掲載コードを模擬サーバーと Redis で実際に動かして見つけた二つの穴——停止が AbortError にならない問題と、記録キーの不一致で月間上限が一度も働かない問題——と、予約→精算の修正版

✦WebSocket サーバー・Claude API ストリーム変換・Redis ルーティング・JWT 予算制御の完全な実装コード（Node.js / TypeScript）

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

✦

この記事を購入する

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

または

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

WebSocket サーバーの基本実装

まずは単一インスタンスの WebSocket サーバーを実装します。後でRedis による水平スケールを追加します。

// src/server.ts
import { WebSocketServer, WebSocket } from "ws";
import { createServer } from "http";
import { verifyToken, type JwtPayload } from "./auth.js";
import { ChatSession } from "./session.js";
 
const httpServer = createServer();
const wss = new WebSocketServer({ server: httpServer });
 
// ユーザーIDとWebSocket接続のマップ
const sessions = new Map<string, ChatSession>();
 
wss.on("connection", async (ws: WebSocket, req) => {
  // クエリパラメータから JWT トークンを取得
  const url = new URL(req.url ?? "/", "http://localhost");
  const token = url.searchParams.get("token");
 
  if (!token) {
    ws.close(4001, "Authentication required");
    return;
  }
 
  let payload: JwtPayload;
  try {
    payload = verifyToken(token);
  } catch {
    ws.close(4001, "Invalid token");
    return;
  }
 
  const { userId } = payload;
  const session = new ChatSession(userId, payload.plan, ws);
  sessions.set(userId, session);
 
  console.log(`✅ Connected: userId=${userId}`);
 
  ws.on("message", async (data) => {
    try {
      const message = JSON.parse(data.toString());
      await session.handleMessage(message);
    } catch (err) {
      ws.send(JSON.stringify({ type: "error", message: "Invalid message format" }));
    }
  });
 
  ws.on("close", () => {
    sessions.delete(userId);
    session.cleanup();
    console.log(`🔌 Disconnected: userId=${userId}`);
  });
 
  ws.on("error", (err) => {
    console.error(`WebSocket error for userId=${userId}:`, err);
    sessions.delete(userId);
    session.cleanup();
  });
});
 
const PORT = process.env.PORT ?? 8080;
httpServer.listen(PORT, () => {
  console.log(`🚀 WebSocket server listening on port ${PORT}`);
});

接続確立時に JWT を検証し、ユーザーIDをキーにしてセッションを管理するのがポイントです。接続ごとに状態を持つため、切断時のクリーンアップを確実に実装する点が肝心です。

Claude API ストリーミングをWebSocketに変換する

次に、ChatSession クラスの実装です。ここが最も重要な部分です。

// src/session.ts
import { WebSocket } from "ws";
import Anthropic from "@anthropic-ai/sdk";
import { checkBudget, recordTokenUsage, type BudgetResult } from "./budget.js";
import type { JwtPayload } from "./auth.js";
 
const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});
 
interface Message {
  role: "user" | "assistant";
  content: string;
}
 
export class ChatSession {
  private userId: string;
  private ws: WebSocket;
  private plan: JwtPayload["plan"];
  private history: Message[] = [];
  private isStreaming = false;
  private abortController: AbortController | null = null;
 
  constructor(userId: string, plan: JwtPayload["plan"], ws: WebSocket) {
    this.userId = userId;
    this.plan = plan;
    this.ws = ws;
  }
 
  async handleMessage(message: { type: string; content?: string }) {
    if (message.type === "chat") {
      await this.handleChat(message.content ?? "");
    } else if (message.type === "stop") {
      this.stopStreaming();
    }
  }
 
  private async handleChat(userContent: string) {
    if (this.isStreaming) {
      this.send({ type: "error", message: "Already processing a message" });
      return;
    }
 
    // トークンバジェットチェック（後述）
    const budget: BudgetResult = await checkBudget(this.userId, this.plan);
    if (!budget.allowed) {
      this.send({
        type: "error",
        message: `Monthly limit reached. Used: ${budget.used} / ${budget.limit} tokens`,
      });
      return;
    }
 
    this.history.push({ role: "user", content: userContent });
    this.isStreaming = true;
    this.abortController = new AbortController();
 
    let assistantContent = "";
 
    try {
      // ストリーミング開始を通知
      this.send({ type: "stream_start" });
 
      const stream = await anthropic.messages.stream(
        {
          model: "claude-sonnet-4-6",
          max_tokens: 2048,
          system: "あなたは親切で誠実なAIアシスタントです。",
          messages: this.history.map((m) => ({
            role: m.role,
            content: m.content,
          })),
        },
        { signal: this.abortController.signal }
      );
 
      for await (const chunk of stream) {
        if (chunk.type === "content_block_delta") {
          const delta = chunk.delta;
          if (delta.type === "text_delta") {
            assistantContent += delta.text;
            // デルタをWebSocketで即時送信
            this.send({ type: "delta", content: delta.text });
          }
        }
      }
 
      // ファイナルメッセージからトークン使用量を取得
      const finalMessage = await stream.finalMessage();
      const tokensUsed = finalMessage.usage.input_tokens + finalMessage.usage.output_tokens;
 
      // 履歴に追加
      this.history.push({ role: "assistant", content: assistantContent });
 
      // ストリーム完了を通知（トークン使用量も含める）
      this.send({
        type: "stream_end",
        tokensUsed,
        remainingBudget: budget.remaining - tokensUsed,
      });
 
      // Redis に使用量を記録（予算の参照と同じキーに書く。非同期）
      recordTokenUsage(this.userId, tokensUsed).catch(console.error);
 
    } catch (err: unknown) {
      // SDK が投げるのは APIUserAbortError で、err.name は "Error" のままです（後述）
      if (err instanceof Anthropic.APIUserAbortError) {
        // ユーザーが停止した場合
        this.history.push({ role: "assistant", content: assistantContent });
        this.send({ type: "stream_stopped", content: assistantContent });
      } else {
        console.error("Claude API error:", err);
        this.send({ type: "error", message: "AI response failed. Please try again." });
        // エラー時は不完全なアシスタントメッセージを履歴から除外
      }
    } finally {
      this.isStreaming = false;
      this.abortController = null;
    }
  }
 
  stopStreaming() {
    if (this.abortController && this.isStreaming) {
      this.abortController.abort();
    }
  }
 
  private send(data: object) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
  }
 
  cleanup() {
    this.stopStreaming();
  }
}

重要なのは abortController の使い方です。ユーザーが「停止」メッセージを送ったとき、abort() を呼ぶと stream の for-await ループが AbortError で中断されます。不完全な応答をそのまま履歴に残すか破棄するかはビジネス要件によりますが、ここでは生成済みの部分を保存しています。

Redis Pub/Sub によるマルチインスタンス対応

サーバーが1台なら sessions マップだけで十分です。ところが本番ではロードバランサー下で複数インスタンスが動きます。ユーザー A の WebSocket が Pod 1 に接続しているとき、Pod 2 がそのユーザーへのメッセージを受け取っても届けられません。

Redis Pub/Sub がこの問題を解決します。

// src/pubsub.ts
import { getRedisClient, getSubscriberClient } from "./redis.js";
 
type MessageHandler = (message: object) => void;
const localHandlers = new Map<string, MessageHandler>();
 
// サブスクライバーの初期化（サーバー起動時に1回だけ）
export async function initPubSub() {
  const subscriber = getSubscriberClient();
 
  // 全ユーザー向けチャンネルをパターンサブスクライブ
  await subscriber.psubscribe("user:*:messages");
 
  subscriber.on("pmessage", (_pattern, channel, message) => {
    // channel 例: "user:user123:messages"
    const userId = channel.split(":")[1];
    const handler = localHandlers.get(userId);
    if (handler) {
      try {
        handler(JSON.parse(message));
      } catch (err) {
        console.error("PubSub message parse error:", err);
      }
    }
  });
 
  console.log("✅ Redis Pub/Sub initialized");
}
 
// このインスタンスにいるユーザーのハンドラーを登録
export function registerUserHandler(userId: string, handler: MessageHandler) {
  localHandlers.set(userId, handler);
}
 
export function unregisterUserHandler(userId: string) {
  localHandlers.delete(userId);
}
 
// 任意のインスタンスからユーザー宛てにメッセージを送る
export async function publishToUser(userId: string, message: object) {
  const publisher = getRedisClient();
  await publisher.publish(`user:${userId}:messages`, JSON.stringify(message));
}

ChatSession の constructor でハンドラーを登録し、cleanup で解除するように修正します。

// ChatSession の constructor に追加
import { registerUserHandler, unregisterUserHandler } from "./pubsub.js";
 
constructor(userId: string, ws: WebSocket) {
  this.userId = userId;
  this.ws = ws;
 
  // Pub/Sub ハンドラーを登録（別インスタンスからのメッセージも受け取れる）
  registerUserHandler(userId, (message) => {
    this.send(message);
  });
}
 
cleanup() {
  this.stopStreaming();
  unregisterUserHandler(this.userId);
}

これで、どのインスタンスから publishToUser("user123", {...}) を呼んでも、user123 の WebSocket が接続しているインスタンスまでメッセージが届くようになります。

JWT 認証とユーザーごとのコスト制御

プロダクション環境で最も重要な実装のひとつが、ユーザーごとの使用量制限です。制限なしで公開すると、1ユーザーが大量のトークンを消費し、API 費用が爆発します。

// src/auth.ts
import jwt from "jsonwebtoken";
 
export interface JwtPayload {
  userId: string;
  plan: "free" | "pro" | "premium";
  iat: number;
  exp: number;
}
 
const JWT_SECRET = process.env.JWT_SECRET ?? "change-me-in-production";
 
export function verifyToken(token: string): JwtPayload {
  return jwt.verify(token, JWT_SECRET) as JwtPayload;
}
 
export function generateToken(userId: string, plan: JwtPayload["plan"]): string {
  return jwt.sign({ userId, plan }, JWT_SECRET, { expiresIn: "24h" });
}

// src/budget.ts
import { getRedisClient } from "./redis.js";
import type { JwtPayload } from "./auth.js";
 
// プランごとの月間トークン上限
const MONTHLY_LIMITS: Record<JwtPayload["plan"], number> = {
  free: 50_000,    // 短い会話で数十回ぶん
  pro: 500_000,
  premium: 5_000_000,
};
 
export interface BudgetResult {
  allowed: boolean;
  used: number;
  limit: number;
  remaining: number;
}
 
// 参照と記録で必ず同じキーを使うため、キー生成は一か所にまとめます
function monthKey(userId: string): string {
  return `budget:${userId}:${new Date().toISOString().slice(0, 7)}`; // YYYY-MM
}
 
export async function checkBudget(
  userId: string,
  plan: JwtPayload["plan"]
): Promise<BudgetResult> {
  const redis = getRedisClient();
  const used = parseInt((await redis.get(monthKey(userId))) ?? "0", 10);
  const limit = MONTHLY_LIMITS[plan];
  return { allowed: used < limit, used, limit, remaining: Math.max(0, limit - used) };
}
 
export async function recordTokenUsage(userId: string, tokens: number): Promise<void> {
  const redis = getRedisClient();
  const key = monthKey(userId);
  await redis.incrby(key, tokens);
  await redis.expire(key, 60 * 60 * 24 * 45); // 45日保持
}

プランごとに上限を設け、月次の使用量を Redis に記録します。free ユーザーの 50,000 トークンは、Claude Sonnet 4.6（入力 $3 / 出力 $15 per 1M、公式料金表で確認）なら、すべて入力なら約 $0.15、すべて出力なら約 $0.75 です。実際の会話はその間に収まります。ここは自社のビジネスモデルに合わせて調整してください。

実践的なクライアント実装（TypeScript）

クライアント側（React/React Native 想定）の実装例です。接続の再試行ロジックは、最初に作り込んでおきたい部分です。

// client/useChatWebSocket.ts
import { useState, useEffect, useRef, useCallback } from "react";
 
interface ChatMessage {
  role: "user" | "assistant";
  content: string;
}
 
export function useChatWebSocket(token: string) {
  const [messages, setMessages] = useState<ChatMessage[]>([]);
  const [currentResponse, setCurrentResponse] = useState("");
  const [isStreaming, setIsStreaming] = useState(false);
  const [isConnected, setIsConnected] = useState(false);
  const wsRef = useRef<WebSocket | null>(null);
  const retryCount = useRef(0);
  const maxRetries = 5;
 
  const connect = useCallback(() => {
    if (wsRef.current?.readyState === WebSocket.OPEN) return;
 
    const ws = new WebSocket(
      `${process.env.NEXT_PUBLIC_WS_URL}?token=${token}`
    );
 
    ws.onopen = () => {
      console.log("✅ WebSocket connected");
      setIsConnected(true);
      retryCount.current = 0;
    };
 
    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
 
      switch (data.type) {
        case "stream_start":
          setIsStreaming(true);
          setCurrentResponse("");
          break;
 
        case "delta":
          setCurrentResponse((prev) => prev + data.content);
          break;
 
        case "stream_end":
          setIsStreaming(false);
          setMessages((prev) => [
            ...prev,
            { role: "assistant", content: currentResponse },
          ]);
          setCurrentResponse("");
          break;
 
        case "stream_stopped":
          setIsStreaming(false);
          setMessages((prev) => [
            ...prev,
            { role: "assistant", content: data.content + " [中断]" },
          ]);
          setCurrentResponse("");
          break;
 
        case "error":
          setIsStreaming(false);
          console.error("Server error:", data.message);
          break;
      }
    };
 
    ws.onclose = (event) => {
      setIsConnected(false);
      wsRef.current = null;
 
      // 認証エラーはリトライしない
      if (event.code === 4001) return;
 
      if (retryCount.current < maxRetries) {
        const delay = Math.min(1000 * 2 ** retryCount.current, 30000); // 指数バックオフ
        console.log(`Retrying in ${delay}ms (attempt ${retryCount.current + 1})`);
        setTimeout(() => {
          retryCount.current++;
          connect();
        }, delay);
      }
    };
 
    ws.onerror = (err) => {
      console.error("WebSocket error:", err);
    };
 
    wsRef.current = ws;
  }, [token]);
 
  useEffect(() => {
    connect();
    return () => {
      wsRef.current?.close();
    };
  }, [connect]);
 
  const sendMessage = useCallback(
    (content: string) => {
      if (wsRef.current?.readyState !== WebSocket.OPEN) {
        console.warn("WebSocket not connected");
        return;
      }
      setMessages((prev) => [...prev, { role: "user", content }]);
      wsRef.current.send(JSON.stringify({ type: "chat", content }));
    },
    []
  );
 
  const stopGeneration = useCallback(() => {
    wsRef.current?.send(JSON.stringify({ type: "stop" }));
  }, []);
 
  return { messages, currentResponse, isStreaming, isConnected, sendMessage, stopGeneration };
}

指数バックオフでの再接続（1s → 2s → 4s → 8s → 16s → 30s 上限）は、ネットワークが不安定な環境ほど効いてきます。モバイルアプリではバックグラウンド復帰時に接続が切れていることが多いので、この処理を怠ると UI 上は繋がっているように見えてメッセージが届かない問題が起きます。

よくある落とし穴と対策

実装中に遭遇しやすい問題を3つ挙げます。

1. 「会話履歴が溜まりすぎてコンテキストウィンドウを超える」

長時間のセッションでは history 配列が膨れ上がり、Claude API への入力トークンが増え続けます。200K コンテキストでもやがて限界に達します。

// 会話履歴を最新N件に制限するヘルパー
function trimHistory(history: Message[], maxMessages = 20): Message[] {
  if (history.length <= maxMessages) return history;
  // 最初のシステムコンテキスト（あれば）は保持して、中間を切り捨て
  const recent = history.slice(-maxMessages);
  return recent;
}

定期的にサーバー側でサマリーを作成し、古い履歴を1つのアシスタントメッセージに圧縮するアプローチも有効です。

2. 「ストリーム中に接続が切れると不完全な履歴が残る」

ストリーミング中にクライアントが切断すると、ws.on("close") が発火します。この時点で abortController.abort() を呼ぶと Claude API の生成も中断されますが、生成済みの部分テキストをどう扱うかが問題です。

私のアプローチは「部分的なアシスタントメッセージは履歴に残す（[未完了] マーカー付き）」です。ユーザーが再接続したときに会話の流れが分かるようにするためです。

ws.on("close", () => {
  if (session.isStreaming) {
    // 生成中断時、部分テキストを履歴に保存
    session.abortAndSavePartial();
  }
  sessions.delete(userId);
  session.cleanup();
});

3. 「Redis 接続がタイムアウトして WebSocket サーバー全体が応答不能になる」

Redis が一時的に落ちたとき、Redis クライアントの接続待ちがブロッキングになり WebSocket サーバー全体が固まるケースがあります。

// src/redis.ts
import Redis from "ioredis";
 
let client: Redis | null = null;
 
export function getRedisClient(): Redis {
  if (!client) {
    client = new Redis({
      host: process.env.REDIS_HOST ?? "localhost",
      port: parseInt(process.env.REDIS_PORT ?? "6379"),
      retryStrategy: (times) => {
        if (times > 3) return null; // 3回失敗したらリトライ諦め
        return Math.min(times * 200, 2000);
      },
      enableOfflineQueue: false, // Redis 未接続時に即エラーを返す（キューに溜めない）
      connectTimeout: 3000,
      commandTimeout: 2000,
    });
 
    client.on("error", (err) => {
      console.error("Redis error:", err);
      // アプリを落とさず、個別コマンドのエラーとして処理
    });
  }
  return client;
}

enableOfflineQueue: false は見落としやすい設定です。Redis が落ちているとき、コマンドをキューに溜めずに即エラーを返すことで、WebSocket ハンドラーが Redis の回復を待ってブロッキングすることを防げます。

停止ボタンを押してから、課金が実際に止まるまで

WebSocket に載せ替えるとき、私が最初に期待していたのは「速くなること」でした。

結果から言えば、最初のトークンが届くまでの時間はほとんど変わりません。その区間を支配しているのは Claude API 側が生成を始めるまでの時間で、トランスポートを差し替えても短くなりません。期待していた方向とは違いました。

変わったのは、止めたいときに止まるかどうかです。

個人開発の壁紙アプリで問い合わせチャットのバックエンドを触っていたときのことです。SSE と REST を組み合わせた構成では、ユーザーが停止ボタンを押してから生成が本当に止まるまでに、別接続の往復が挟まります。その往復の間も、Claude API は文章を作り続けています。作られた分は、当然ながら課金されます。

「止めたつもり」を計測する

推測で語る前に、自分の環境の数字を持つことです。停止を要求した瞬間から、ストリームが実際に終わるまでに何トークン届いたかを記録します。

// src/abort-metrics.ts
type StopTrace = { requestedAt: number; tokensAfterStop: number };
 
const traces = new Map<string, StopTrace>();
 
// stop メッセージを受け取った時点で記録を開始する
export function markStopRequested(streamId: string) {
  traces.set(streamId, { requestedAt: Date.now(), tokensAfterStop: 0 });
}
 
// delta を WebSocket に流すたびに呼ぶ（停止要求後の分だけ数える）
export function countDeltaAfterStop(streamId: string, text: string) {
  const t = traces.get(streamId);
  if (!t) return;
  // 概算で構いません。正確な値は message_delta の usage で取れます
  t.tokensAfterStop += Math.ceil(text.length / 4);
}
 
// ストリームが実際に終了した時点で確定させる
export function finalizeStop(streamId: string): StopTrace | null {
  const t = traces.get(streamId);
  if (!t) return null;
  traces.delete(streamId);
  console.info(
    JSON.stringify({
      event: "stop_latency",
      streamId,
      latencyMs: Date.now() - t.requestedAt,
      tokensAfterStop: t.tokensAfterStop,
    })
  );
  return t;
}

latencyMs と tokensAfterStop の2つを、まず1週間ぶんログに溜めてみてください。この2つが、以降のすべての判断の土台になります。

遅延はそのまま金額に変換できる

上のフックで測った停止遅延は、単価と生成速度が分かれば金額に直せます。

前提を明示しておきます。出力トークンの単価を $10 / 1M（公式料金表に載っている Claude Sonnet 5 の出力単価。2026年10月に確認）、生成速度を毎秒およそ50トークン、ユーザーによる中断を月3,000回として試算しています。生成速度はモデルと出力内容で変わりますので、実際にはご自身のログの値に置き換えてください。

停止経路	停止遅延の目安	1回あたりの無駄トークン	月3,000回での追加コスト
WebSocket 上で abort() を直接呼ぶ	約100ms	約5	約 $0.15
REST の cancel エンドポイントを別途叩く	約500ms	約25	約 $0.75
前段プロキシのバッファリングを挟む	約2,000ms	約100	約 $3.00
停止を配線していない（残りを生成しきる）	—	約400	約 $12.00

金額だけを見れば、月に十数ドルの差です。これを理由に WebSocket を選ぶ必要はありません。

効いてくるのは、その先です。停止したはずのトークンが課金され続けると、前述の TokenBudget によるユーザーごとの上限管理が実態からずれていきます。ずれた上限で弾かれたユーザーからの問い合わせは、コードの問題ではなく信頼の問題として返ってきます。私が停止経路にこだわるのは、金額よりもそちらの理由からです。

掲載コードを実際に動かして、二つの穴が見つかりました

以前この記事に載せていたコードでは、checkAndDeductBudget という名前の関数が「確認して差し引く」ことをしておらず、記録は別のキーに書かれていました。名前を信じて読み流していたのは私です。今回、手元の模擬 SSE サーバー（Node.js 22 / @anthropic-ai/sdk 0.133.0）と ioredis-mock で実行して、はじめて気づきました。

穴 1: 停止しても AbortError にならないのです。 5 つ目のデルタで abort() を呼んだところ、投げられた例外は次のとおりでした。

確認した項目	結果
err.name	"Error"
err.constructor.name	APIUserAbortError
err instanceof Anthropic.APIUserAbortError	true

err.name === "AbortError" で分岐していた以前のコードでは、停止のたびに「AI response failed」というエラー側へ落ち、stream_stopped も部分テキストの保存も実行されませんでした。判定は instanceof Anthropic.APIUserAbortError に改めています。

穴 2: 月間上限が一度も働いていませんでした。 以前の記録先は usage: で始まるキー、参照先は budget: で始まるキーでした。1 回 2,000 トークンの会話を 100 回（合計 200,000 トークン）記録してみると、usage: 側には 200000 と積み上がる一方、budget: 側は空のまま。上限 50,000 の free プランでも、弾かれる回は一度も来ませんでした。上のコードでは、キー生成を monthKey に一本化してあります。

穴 3: 確認と記録のあいだに隙間がありました。 キーをそろえても、「読んで、あとで足す」構成では、同時に届いた要求が全員通ります。残り枠 50,000 に対し、10,000 トークンぶんの要求を 20 件同時に投げた結果は次のとおりです。

方式	通過した件数	最終的な使用量
読んでから、あとで記録（以前の掲載コード）	20 件	200,000
先に予約して、戻り値で判定	5 件	50,000

同じユーザーが複数の端末やインスタンスから同時に送る場面は、本番では珍しくありません。INCRBY の戻り値は一件ごとに順番が付くので、Lua スクリプトがなくても、「先に枠を予約し、超えたら戻す」だけで判定が分かれます。

// src/budget.ts（追記）予約 → 精算
export async function reserveBudget(
  userId: string,
  plan: JwtPayload["plan"],
  reserveTokens: number
): Promise<{ ok: boolean }> {
  const redis = getRedisClient();
  const key = monthKey(userId);
  const after = await redis.incrby(key, reserveTokens); // 戻り値で順番が決まる
  if (after > MONTHLY_LIMITS[plan]) {
    await redis.decrby(key, reserveTokens); // 超えたぶんは返す
    return { ok: false };
  }
  await redis.expire(key, 60 * 60 * 24 * 45);
  return { ok: true };
}
 
// 実績が分かったら差分だけ足し引きする（停止・エラー時は届いたぶんで精算）
export async function settleBudget(
  userId: string,
  reserved: number,
  actual: number
): Promise<void> {
  const redis = getRedisClient();
  await redis.incrby(monthKey(userId), actual - reserved);
}

予約量は、max_tokens に入力の概算を足した値で十分です。finally で必ず settleBudget を呼ぶ形にしておけば、停止やエラーでも枠が戻らなくなる事態を避けられます。この精算は、ここまでの試験では未検証です。実際に組み込む際は、ご自身の環境で二本同時に走らせて確かめていただければと思います。

表示された数字ではなく、書き込まれたキーを見ます。 この線引きは、次に似た関数を書くときも崩さないようにしています。

公式ドキュメントには書かれていない3つの前提

計測を進めるうちに、はっきりした形では文書化されていない前提が3つあることに気づきました。

1. close ハンドラで abort() を呼ばなければ、生成は続きます。

クライアントの接続が切れても、Claude API 側のストリームは自動では止まりません。届け先を失ったトークンが生成され、そのまま課金されます。前掲の ws.on("close") で abortController.abort() を必ず呼んでいるのは、履歴の整合性のためだけではありません。

2. Redis Pub/Sub は、届かなかったメッセージを保持しません。

Pub/Sub は購読者がいなければメッセージを捨てます。停止指示を Pub/Sub だけに乗せていると、購読側インスタンスが再起動している数秒の間に、その指示は静かに消えます。生成は止まりません。

対策はシンプルです。同一インスタンスに接続がある場合は直接 abort() を呼び、Pub/Sub は他インスタンス宛ての経路としてのみ使います。

async function requestStop(userId: string) {
  const local = sessions.get(userId);
  if (local) {
    local.abortController?.abort(); // 同一インスタンス: 確実な経路
    return;
  }
  // 他インスタンスにいる場合のみ Pub/Sub へ
  await publisher.publish(`user:${userId}`, JSON.stringify({ type: "stop" }));
}

3. ロードバランサーのアイドルタイムアウトは、WebSocket にも適用されます。

AWS の ALB は既定で60秒のアイドルタイムアウトを持ちます。これは WebSocket の常時接続にも効きます。ユーザーが黙って考えている間に接続が切られ、次の送信で「なぜか届かない」状態になります。

ハートビートの間隔は、必ずアイドルタイムアウトより短く設定してください（60秒設定なら30秒間隔が安全側です）。同様に、Nginx を前段に置く場合は proxy_read_timeout を明示的に伸ばす必要があります。

私はこの確認を、一度取りこぼしてから最初のチェック項目に置くようにしました。原因が分からないまま「たまにメッセージが届かない」と言われる時間は、できれば繰り返したくないものです。

状況別の推奨

状況	推奨する構成	理由
単発の質問と回答だけ・中断なし	SSE のまま	WebSocket の運用コストに見合う利点がありません
中断あり・単一インスタンスで足りる	WebSocket のみ（Redis なし）	停止経路がプロセス内で完結し、最も確実です
中断あり・複数インスタンス	WebSocket + Redis Pub/Sub	ローカル優先・Pub/Sub は他インスタンス向けの補助に
ユーザーごとの厳密な予算管理が必要	WebSocket + 停止遅延の常時計測	上限の精度は停止経路の速さで決まります

本番デプロイのチェックリスト

[ ] JWT_SECRET を本番専用の強固なランダム文字列へ変更しましたか（32 文字以上を推奨します）
[ ] WebSocket サーバーの前段に Nginx または ALB を置き、SSL/TLS を終端していますか（wss:// のみに限定します）
[ ] ANTHROPIC_API_KEY を環境変数で安全に管理し、コードへ直書きしていませんか
[ ] Redis に requirepass でパスワードを設定しましたか
[ ] インスタンスの分散には、ロードバランサーの WebSocket スティッキーセッション（同一クライアントを同一インスタンスへ）または Redis Pub/Sub のどちらかを採用しましたか
[ ] PM2 または systemd でプロセスを管理し、クラッシュ時に自動で再起動する設定にしましたか
[ ] ハートビートを有効にし、切れたまま残った接続を定期的に掃除していますか
[ ] CloudWatch / Datadog などで、アクティブ接続数・Redis のレイテンシー・Claude API のエラー率を監視していますか
[ ] 1 ユーザーあたりの同時接続数を制限しましたか（過負荷や誤動作の防止になります）

チェックリストを一通り眺めて気づいたのは、項目の多くが「何かが起きてから」ではなく「起きる前に見える状態にしておく」ためのものだということです。接続数もエラー率も、グラフがなければ異常に気づく手段がありません。最初のデプロイ前に、監視の画面だけでも先に用意しておくと安心できるかもしれません。

// 同一ユーザーの重複接続を防ぐ
wss.on("connection", async (ws, req) => {
  // ...（JWT 検証後）
 
  if (sessions.has(userId)) {
    // 既存セッションを切断して新しい接続に切り替え
    const existingSession = sessions.get(userId)!;
    existingSession.ws.close(4002, "New connection established");
    existingSession.cleanup();
  }
 
  // 新しいセッションを登録
  sessions.set(userId, new ChatSession(userId, payload.plan, ws));
});

全体を振り返って: 今日できる一歩

今回実装した構成を振り返ると、WebSocket + Redis Pub/Sub の組み合わせは「スケーラブルなリアルタイム AI チャット」の標準的なパターンといえます。複数インスタンスへの水平スケール、双方向の会話制御、ユーザーごとのコスト制限という3つの課題を、それぞれ明確な責務で解決しています。

まずはシングルインスタンスで Redis なしの最小版を動かして、接続→チャット→切断のサイクルを確認してみてください。そこから Redis を足してスケールさせるのが、最も確実な順序です。

会話履歴の永続化（Redis 以外のストレージへの保存）や、プロンプトキャッシュ機能によるコスト削減については、Claude API ストリーミング × リアルタイムチャットUI 本番実装ガイドやClaude API のプロンプトキャッシュで月額コストを半分にした実装メモも合わせてご覧ください。

本番環境で予期しないエラーに遭遇したときは、Claude API プロダクション耐障害設計パターンが参考になります。

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

Claude Lab は広告なしで運営しており、サーバー費用などの運営コストはメンバーシップのご支援で賄っています。実装コード・ベンチマーク・本番設計パターンなど、実務でお役立ていただける記事を毎日更新しています。もし読んでよかったと感じていただけましたら、ぜひご覧ください。

✦コピー&ペーストで使える実装コード付き
✦毎日新しい上級ガイドを追加
✦¥580/月または ¥1,480 の永久アクセス

メンバーシップを見る →