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

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

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

websocket redis² nodejs² streaming¹⁶ 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 を優先する回避策

✦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, 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 { getRedisClient } from "./redis.js";
import { checkAndDeductBudget, type BudgetResult } from "./budget.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 history: Message[] = [];
  private isStreaming = false;
  private abortController: AbortController | null = null;
 
  constructor(userId: string, ws: WebSocket) {
    this.userId = userId;
    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 checkAndDeductBudget(this.userId);
    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 に使用量を記録（非同期）
      this.recordUsage(tokensUsed).catch(console.error);
 
    } catch (err: unknown) {
      if (err instanceof Error && err.name === "AbortError") {
        // ユーザーが停止した場合
        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));
    }
  }
 
  private async recordUsage(tokens: number) {
    const redis = getRedisClient();
    const key = `usage:${this.userId}:${new Date().toISOString().slice(0, 7)}`; // YYYY-MM
    await redis.incrby(key, tokens);
    await redis.expire(key, 60 * 60 * 24 * 45); // 45日間保持
  }
 
  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,    // ~約50回の短い会話
  pro: 500_000,   // 中程度のヘビーユーザー
  premium: 5_000_000, // 実質無制限に近い
};
 
export interface BudgetResult {
  allowed: boolean;
  used: number;
  limit: number;
  remaining: number;
}
 
export async function checkAndDeductBudget(
  userId: string,
  plan: JwtPayload["plan"] = "free"
): Promise<BudgetResult> {
  const redis = getRedisClient();
  const month = new Date().toISOString().slice(0, 7); // YYYY-MM
  const key = `budget:${userId}:${month}`;
  const limit = MONTHLY_LIMITS[plan];
 
  // Lua スクリプトでアトミックなチェック + インクリメント
  // （実際の消費トークン数は後から記録するため、ここでは「残り枠確認」のみ）
  const used = parseInt((await redis.get(key)) ?? "0", 10);
 
  if (used >= limit) {
    return { allowed: false, used, limit, remaining: 0 };
  }
 
  return { allowed: true, used, limit, remaining: limit - used };
}
 
export async function recordTokenUsage(userId: string, tokens: number): Promise<void> {
  const redis = getRedisClient();
  const month = new Date().toISOString().slice(0, 7);
  const key = `budget:${userId}:${month}`;
  await redis.incrby(key, tokens);
  await redis.expire(key, 60 * 60 * 24 * 45); // 45日保持
}

プランごとに上限を設け、月次の使用量を Redis に記録します。free ユーザーの 50,000 トークンは Claude Sonnet 4.6 であれば約 0.15ドル相当です。ここは自社のビジネスモデルに合わせて調整してください。

実践的なクライアント実装（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年8月31日まで）、生成速度を毎秒およそ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 によるユーザーごとの上限管理が実態からずれていきます。ずれた上限で弾かれたユーザーからの問い合わせは、コードの問題ではなく信頼の問題として返ってきます。私が停止経路にこだわるのは、金額よりもそちらの理由からです。

公式ドキュメントには書かれていない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 で Node.js プロセスを管理し、クラッシュ時の自動再起動を設定する
[ ] CloudWatch / Datadog でアクティブ WebSocket 接続数・Redis レイテンシー・Claude API エラー率を監視する
[ ] 1 ユーザーあたりの同時接続数を制限する（DDos・誤動作の防止）

// 同一ユーザーの重複接続を防ぐ
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, ws));
});

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

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

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

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

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