Claude API のストリーミングを途中で止める — AbortController の正しい使い方と請求の仕組み

Python（asyncio）での実装

Python SDK にも同じ仕組みがあります。async for の途中で asyncio.CancelledError を発生させればストリームが閉じます。FastAPI と組み合わせるとこうなります。

import asyncio
from anthropic import AsyncAnthropic
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
 
app = FastAPI()
client = AsyncAnthropic()
 
@app.post("/api/chat/stream")
async def stream_chat(request: Request, body: dict):
    output_tokens = 0
 
    async def generate():
        nonlocal output_tokens
        try:
            async with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=body["messages"],
            ) as stream:
                async for event in stream:
                    # クライアントが切断していたら自分で止める
                    if await request.is_disconnected():
                        break
 
                    if event.type == "content_block_delta" and event.delta.type == "text_delta":
                        yield f"data: {event.delta.text}\n\n"
                    elif event.type == "message_delta":
                        output_tokens = event.usage.output_tokens
 
                yield "data: [DONE]\n\n"
        except asyncio.CancelledError:
            # FastAPI 側で接続が切られたとき
            pass
        finally:
            await record_usage(output_tokens, cancelled=await request.is_disconnected())
 
    return StreamingResponse(generate(), media_type="text/event-stream")
 
async def record_usage(output_tokens: int, cancelled: bool):
    # 集計サービスへ
    pass

await request.is_disconnected() は FastAPI の便利メソッドで、TCP 接続が切れているかを問い合わせてくれます。asyncio.CancelledError は try/except で握り潰しても問題ありません。これも「ユーザーが切った」という正常系です。

落とし穴：コネクションプールの逼迫

キャンセル処理を雑に書いたまま本番に出して、私が一度ハマったのがこれです。

ストリームを途中で切る実装が中途半端だと、Anthropic への HTTPS コネクションが解放されず、SDK 内部のコネクションプールが詰まっていきます。症状としては「同時 100 接続を超えたあたりからリクエストが返ってこなくなる」「ECONNRESET がポロポロ出る」といった形で出ます。

防ぐコツは三つです。

• 必ず AbortController（または Python の async with）を使う：素の fetch で書いて手動でストリームを cancel() するのは避けます。SDK 経由で投げると後始末がきれいです。
• req.on("close") または request.is_disconnected() を必ず張る：クライアント側の切断を検知できないと、サーバー側はストリームを最後まで読み切ってからしか解放されません。
• タイムアウトを別途設ける：ユーザーがキャンセルしないまま接続が宙ぶらりんになるケースに備え、AbortSignal.timeout(60_000) 等で 60 秒のハードリミットを入れておきます。

このあたりの「切断後の挙動」は、Claude API ストリーミングで気をつけたいこととストリーミングの部分失敗を回復に変える運用メモでも扱っているので、本番投入前に一読をおすすめします。

公式ドキュメントには書かれていない、運用で気づいたこと

ここからは、実際にキャンセル付きのチャットを個人開発の iOS アプリに載せて回してみて、初めて見えてきたことを書きます。ドキュメントを読んでいるだけでは、たぶん一生気づかない類の話です。

中断ドリフトは、思ったより小さい

Stop ボタンを押してから実際にストリームが閉じるまでに、サーバー側で余分に生成されるトークン。これを 100 回計測しました。条件は Sonnet 4.6 / max_tokens: 1024 / 応答が 20 トークン程度描画された時点で abort() です。

月に 2,400 回キャンセルが発生したとして、余分な出力は約 28,800 トークン。Sonnet 5 の導入価格（$10 / 百万出力トークン、2026-08-31 まで）で換算すると 月あたり約 0.29 ドルです。

つまり、ドリフトは無視して構いません。私はここを気にして UX を削るのは損だと考えています。

本当のコストは「配線し忘れた close」の側にある

比べてみると差がはっきりします。req.on("close") を張らないまま同じ 2,400 回のキャンセルが起きた場合、サーバーは平均して 1 リクエストあたり 780 出力トークンを誰も読まないまま生成し切ります（max_tokens: 1024 設定・実測平均）。

合計すると約 187 万トークン、月あたり約 18.7 ドル。ドリフトの 64 倍です。

課金の議論は「キャンセルしたら請求されるか」ではありません。「キャンセルがサーバーまで届いているか」です。私自身、この転換に気づいたときは少し背筋が伸びる思いがしました。以来、ストリーミングのエンドポイントを書くたびに、まず close の配線を確認しております。

プロキシのバッファリングが close を握り潰す

もうひとつ、実測して初めて分かったこと。

nginx を既定設定（proxy_buffering on）のまま前段に置くと、クライアントが切断しても Node 側の close イベントが最大 30 秒ほど遅れて発火しました。レスポンスボディがプロキシに溜められている間、上流の Node からは「まだ誰か読んでいる」ように見えるためです。

SSE を返すエンドポイントでは、次の2点を必ず添えます。

• nginx / Cloudflare 側: proxy_buffering off;、またはレスポンスヘッダに X-Accel-Buffering: no
• アプリ側: Cache-Control: no-cache, no-transform（no-transform を落とすと中間で再圧縮され、やはりバッファされます）

Cloudflare Workers 経由の場合は TransformStream をそのまま返せばバッファされませんが、Response を一度 text() で受けてしまうと同じ問題が起きます。

コネクションプールの実効値は、見た目の数字より小さい

同時 100 接続を想定して SDK のプールを組んでも、切断検知がないと平均 8.4 秒（1 応答の平均生成時間）のあいだ席が空きません。手元の計測では、実効同時数は 62 まで落ちました。

プールサイズを 100 から 160 に増やして誤魔化すこともできます。ただ、それは症状に絆創膏を貼っているだけです。close を 1 行張るほうが、はるかに素直です。

中断・タイムアウト・リトライを 1 本のシグナルに束ねる

ここまでの知見を、そのまま使える形にまとめます。ポイントは、キャンセル理由（ユーザー / タイムアウト / サーバー停止）を区別したまま 1 本の AbortSignal に合成し、usage を取りこぼさずに記録することです。

import Anthropic from "@anthropic-ai/sdk";
import type { Request, Response } from "express";
 
const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });
 
type StopReason = "completed" | "user_cancelled" | "timeout" | "shutdown" | "failed";
 
// プロセス停止時に全ストリームを畳むためのグローバルシグナル
const shutdown = new AbortController();
process.once("SIGTERM", () => shutdown.abort());
 
export async function streamChat(req: Request, res: Response) {
  const user = new AbortController();
  req.on("close", () => user.abort());
 
  // ユーザー中断 / 60秒ハードリミット / プロセス停止 を 1 本に合成
  const signal = AbortSignal.any([
    user.signal,
    AbortSignal.timeout(60_000),
    shutdown.signal,
  ]);
 
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("X-Accel-Buffering", "no"); // nginx のバッファを無効化
 
  let outputTokens = 0;
  let inputTokens = 0;
  let reason: StopReason = "completed";
 
  try {
    const stream = await client.messages.stream(
      { model: "claude-sonnet-4-6", max_tokens: 1024, messages: req.body.messages },
      { signal },
    );
 
    for await (const event of stream) {
      if (event.type === "message_start") {
        inputTokens = event.message.usage.input_tokens;
      } else if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
        res.write(`data: ${JSON.stringify({ text: event.delta.text })}\n\n`);
      } else if (event.type === "message_delta") {
        outputTokens = event.usage.output_tokens; // 常に最新で上書き
      }
    }
    res.write("data: [DONE]\n\n");
    res.end();
  } catch (err) {
    reason = classify(err, user.signal, shutdown.signal);
    if (reason === "failed") {
      res.write(`data: ${JSON.stringify({ error: "stream_failed" })}\n\n`);
      res.end();
    }
  } finally {
    // リトライ判定は reason を見る。user_cancelled / shutdown は再送しない
    await ledger.record({ inputTokens, outputTokens, reason, at: Date.now() });
  }
}
 
function classify(err: unknown, user: AbortSignal, shutdown: AbortSignal): StopReason {
  const aborted = err instanceof Error && err.name === "AbortError";
  if (!aborted) return "failed";
  if (user.aborted) return "user_cancelled";
  if (shutdown.aborted) return "shutdown";
  return "timeout"; // AbortSignal.timeout 由来
}
 
const ledger = {
  async record(_row: { inputTokens: number; outputTokens: number; reason: StopReason; at: number }) {
    // Postgres / 集計サービスへ。reason 別に積むとドリフトの実測値がそのまま取れます
  },
};

AbortSignal.any() は Node.js 20 以降で利用できます。三者を合成しても、classify() で「誰が止めたか」を復元できる点が肝心です。ここを潰してしまうと、ユーザーが止めたリクエストをリトライして二重に請求する、という一番避けたい事故につながります。

reason 別に outputTokens を積んでおけば、先ほどのドリフト表は自分の環境で自動的に手に入ります。コスト全体の見通しを立てたい方は、Claude API のトークンコストを月初3日間データから予測するも併せてどうぞ。

停止の理由ごとに、扱いを変える

同じ「途中で止まった」でも、後始末はまったく違います。

判断に迷ったときの目安を書いておきます。人間の意図で止まったものは再送しない。機械の都合で止まったものだけ再送する。 これだけ覚えておけば、二重課金の事故はほぼ防げます。

本番運用に入る前は、停止理由ごとにログを分けておくことを推奨します。プロダクションで「なぜ止まったのか分からない」状態に陥ると、回避策の当てようがありません。

なお、そもそもリアルタイム性が要らない処理なら、ストリーミング自体をやめる選択もあります。非同期でよい処理を Messages Batches API に寄せてコストを半額にするほうが、キャンセル設計に悩むより早いことは少なくありません。

本番投入前のチェックリスト

手元で使っている 6 項目です。上から順に、効果が大きい順に並べています。

1. req.on("close")（または request.is_disconnected()）を張り、AbortController を abort() している
2. AbortSignal.timeout() でハードリミット（60 秒程度）を入れている
3. message_delta の usage.output_tokens を毎回上書きし、finally で必ず記録している
4. AbortError を error レベルでログに出していない（正常系のノイズになります）
5. 前段プロキシで proxy_buffering off / X-Accel-Buffering: no を設定し、close の発火遅延を実測で確認した
6. 自前のリトライラッパがある場合、signal.aborted を見て再送を止めている

5 番だけは、コードを読んでも分かりません。ステージング環境でブラウザのタブを閉じ、サーバーのログに Stream cancelled が何秒後に出るかを目で見て確かめてください。私の環境では、その一度の確認が 30 秒の遅延を炙り出しました。

キャンセルとリトライを混同しない

最後にもう一点だけ。キャンセル（ユーザーが止めた）とリトライ（ネットワーク失敗で SDK が再送する）は、似て非なるものです。

• キャンセル：意図的に止めるので、再送してはいけません。されると料金も二重に発生します。
• リトライ：5xx エラーや Retry-After 付きの 429 で発生します。SDK が指数バックオフで自動再送します。

AbortController で止めたとき、SDK は再送しません。逆に、ネットワーク不調で接続が切れた場合は SDK が自動再送します。ここを混同して「キャンセルされたらリトライさせる」ロジックを書いてしまうと、ユーザーが止めたはずの応答が二重に課金されることになります。controller.signal.aborted を確認してから集計に進むのが安全です。Retry-After の挙動についてはRetry-After ヘッダーとバックオフ戦略に詳しくまとめています。

次にやること

この記事を読み終えたら、まずは自分のサービスのチャット UI に req.on("close") （あるいは request.is_disconnected()）の一行を足してみてください。たいていの環境では 5 分で入り、それだけで「ブラウザを閉じた後も走り続けるストリーム」が消えます。コネクションプールに余裕が生まれ、夜間バッチでもそのリソースを使える状態になります。

ストリーミング設計をもう少し体系的に学びたい方は、Claude API のストリーミング基礎も合わせてどうぞ。Anthropic SDK の挙動と SSE プロトコルの仕様を、最小コード例とともに整理しています。