Claude API を Python で本番運用する

Claude API を触り始めたばかりの頃は、anthropic.Anthropic().messages.create(...) が返ってくるところまで書けば「API 連携できた」と思えるものです。ですが実運用に載せると、そこから先の世界の方がずっと広いことに気付きます。突然の 429、ネットワーク切断、ポツポツ返ってくる 503、ハングアップするリクエスト。これらとどう付き合うかが、本番サービスの品質を決めます。

この記事は、私が Claude API を Python のバックエンドに組み込んで運用してきた中で、最終的に落ち着いた設計を整理したものです。Anthropic SDK の基本的な使い方は別記事に譲って、ここではその外側、信頼性に関わる部分だけを深掘りします。

本番で最初に壊れるのはリトライ設計、次にタイムアウト

Claude API を組み込んだサービスで障害が起きるとき、原因は大きく分けて 3 つです。

リトライが雑で、偶発的なエラーを吸収できていない
タイムアウトが大雑把で、片方向ハング時にサービス全体を引きずる
レートリミットを正しく読めておらず、突発的な 429 で雪崩が起きる

この 3 つに目を向けると、SDK を単純にラップするだけの実装では足りないことが見えてきます。SDK は正常系を綺麗に扱うのが仕事ですが、異常系は呼び出し側で設計する必要があります。

リトライ戦略は「何を対象にするか」から決める

リトライ設計の一番のポイントは、「何を対象にリトライするか」を曖昧にしないことです。Claude API で遭遇するエラーは大きく次のカテゴリに分けられます。

リトライすべきもの：429（レートリミット）、503（一時的な過負荷）、接続エラー、読み取りタイムアウト
リトライしてはいけないもの：400 系の大半（リクエストが根本的に不正）、401（認証）、403（権限）、context overflow
ケースバイケース：500（サーバー側）、504（ゲートウェイタイムアウト）

この分類を無視して「エラーなら 3 回リトライ」と書いてしまうと、認証が切れた瞬間にリトライだけで 3 倍の負荷を出す、というような事故が起きます。

私の基本形はこうです。

import random
import time
from anthropic import Anthropic, APIError, APIStatusError, APIConnectionError, APITimeoutError
 
client = Anthropic(timeout=60.0, max_retries=0)
 
RETRYABLE_STATUS = {429, 503, 504, 529}
MAX_ATTEMPTS = 5
BASE_DELAY = 1.0
MAX_DELAY = 30.0
 
def call_with_retry(payload: dict) -> dict:
    attempt = 0
    while True:
        attempt += 1
        try:
            return client.messages.create(**payload)
 
        except APIStatusError as e:
            if e.status_code not in RETRYABLE_STATUS or attempt >= MAX_ATTEMPTS:
                raise
            sleep_for = _retry_delay(e, attempt)
 
        except (APIConnectionError, APITimeoutError) as e:
            if attempt >= MAX_ATTEMPTS:
                raise
            sleep_for = _backoff(attempt)
 
        time.sleep(sleep_for)
 
def _backoff(attempt: int) -> float:
    # full jitter
    cap = min(BASE_DELAY * 2 ** (attempt - 1), MAX_DELAY)
    return random.uniform(0, cap)
 
def _retry_delay(err: APIStatusError, attempt: int) -> float:
    # Retry-After があれば最優先
    retry_after = getattr(err.response, "headers", {}).get("retry-after")
    if retry_after:
        try:
            return min(float(retry_after), MAX_DELAY)
        except ValueError:
            pass
    return _backoff(attempt)

SDK の組み込み max_retries はあえて 0 にして、自前で握っています。理由は 2 つあって、ひとつは Retry-After ヘッダを自分で読んでログに残したいから。もうひとつは、リトライに関するロジックを「どこで何回寝ているか」を一箇所に集約したいからです。SDK 任せにすると、障害時にどの層で何秒寝ているかが分からなくなります。

429 を見たら、まず「誰がなぜ出したか」を確認する

429 は Anthropic 側からの「今は待って」というシグナルですが、原因は複数あります。アカウント全体のトークンレート、モデル別の RPM、同一キーでの同時接続数など、異なるレイヤーで出ます。

ここで効くのが、エラーメッセージと Retry-After ヘッダをログに構造化して残す習慣です。

except APIStatusError as e:
    log.warning({
        "event": "anthropic_status_error",
        "status": e.status_code,
        "message": getattr(e, "message", None),
        "request_id": getattr(e, "request_id", None),
        "retry_after": getattr(e.response, "headers", {}).get("retry-after"),
        "attempt": attempt,
    })

リクエスト ID (request_id) は、Anthropic のサポートに問い合わせる時に決定的に重要な情報です。普段から残していないと、いざトラブった時に手がかりが消えます。

私は過去、夜中に突発的な 429 が出るようになった時、request_id を時系列で眺めて、特定のモデルだけ急に重くなっているのを見つけました。アプリ側は何も変えていなかったので、原因が自分たち側にないと早期に結論できたのは、このログのおかげです。

connect timeout と read timeout を分離して設計する

Python の httpx は接続タイムアウトと読み取りタイムアウトを分けて指定できます。Claude API のように、リクエストの処理時間がプロンプトによって大きく変わる相手では、この分離が決定的に重要です。

接続タイムアウトは「TCP が繋がらない」状態に対する見切りなので、短くて構いません。私の現場では 5 秒が基本です。一方、読み取りタイムアウトは「Claude が考えている最中」も含むので、長めに取ります。モデルや max_tokens によりますが、私は 60 〜 180 秒で調整しています。

import httpx
from anthropic import Anthropic
 
http_client = httpx.Client(
    timeout=httpx.Timeout(
        connect=5.0,
        read=120.0,
        write=10.0,
        pool=5.0,
    ),
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
)
 
client = Anthropic(http_client=http_client, max_retries=0)

もう一つ重要なのは、timeout=60.0 のように単一値で指定した場合、その値が connect にもそのまま効いてしまうことです。connect=60 は長すぎて、DNS や TCP 層の異常を早く検出できません。必ず分離しておきましょう。

ストリーミング応答を使う場合は、さらに注意が必要です。ストリーミングでは「最初のバイトが返ってくる時間」と「応答が完了する時間」が別の概念になります。read timeout はチャンク間のアイドル時間として効くので、ストリーミングなら 60 秒で十分なケースが多いです。非ストリーミングで長文を出すなら 120 〜 180 秒が安心、という使い分けになります。

タイムアウトとリトライの相互作用を設計する

ここがつまずきやすいポイントです。read timeout を 120 秒、リトライを 5 回に設定すると、最悪ケースで 10 分以上待つ設計になってしまいます。上位のリクエストには数秒〜数十秒の SLA が設定されていることが多いので、それを超える時間を API 層で費やすのは事故の元です。

私は次のルールで設計するようにしています。

全体の上限時間を、呼び出し側の SLA の 70% 以内に収める
1 回あたりの read timeout × 最大リトライ回数が、全体上限を超えないように逆算する
ユーザー向け同期処理では max_attempts=3 を上限にする（5 回以上は非同期キュー向けの設計）

コードで表現するとこうなります。

import time
 
def call_with_budget(payload: dict, total_budget_s: float) -> dict:
    start = time.monotonic()
    attempt = 0
    while True:
        attempt += 1
        remaining = total_budget_s - (time.monotonic() - start)
        if remaining <= 0:
            raise TimeoutError("budget exhausted")
 
        per_attempt_timeout = min(remaining, 30.0)
        try:
            return client.with_options(timeout=per_attempt_timeout).messages.create(**payload)
        except (APIConnectionError, APITimeoutError) as e:
            if attempt >= MAX_ATTEMPTS or (time.monotonic() - start) >= total_budget_s:
                raise
            time.sleep(_backoff(attempt))

with_options(timeout=...) で毎回のタイムアウトを動的に決め、残り予算を超えないように調整します。この形にしておくと、上位 API の SLA を壊しません。

同時並列実行の設計 — 気付かないうちにレートを使い切らないために

バッチ処理や複数ユーザー同時対応で Claude を並列に叩く場合、無制限の並列は必ず 429 を引きます。asyncio.gather で 50 リクエストを同時発射する、みたいな実装は本番では通りません。

私は asyncio.Semaphore で並列数に明示的な上限を掛け、さらに「同時並列数」と「秒間リクエスト数」の両方を管理するようにしています。

import asyncio
from anthropic import AsyncAnthropic
 
client = AsyncAnthropic()
sem = asyncio.Semaphore(10)  # 同時並列上限
 
async def call_async(payload: dict):
    async with sem:
        return await client.messages.create(**payload)

Semaphore だけでは秒間レートまでは制御できないので、強めのバーストが予想される場合はトークンバケットを挟みます。簡易実装でよければこれで十分です。

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last = time.monotonic()
        self.lock = asyncio.Lock()
 
    async def acquire(self):
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last = now
            if self.tokens < 1:
                wait = (1 - self.tokens) / self.rate
                await asyncio.sleep(wait)
                self.tokens = 0
            else:
                self.tokens -= 1

Anthropic の RPM 上限を 70 〜 80% で使う程度に設定しておくと、他のクライアントが共有している場合でも安全です。

エラーハンドリングの記録粒度を揃える

本番運用で差が出るのは、エラーが起きた時の「記録の粒度」です。SDK の例外はそのまま投げず、メタデータを付けた上で構造化ログに落とします。

def call_safely(payload: dict):
    request_id = None
    try:
        res = client.messages.create(**payload)
        return res
    except APIStatusError as e:
        request_id = getattr(e, "request_id", None)
        log.error({
            "event": "claude_api_status_error",
            "status": e.status_code,
            "request_id": request_id,
            "payload_model": payload.get("model"),
            "payload_max_tokens": payload.get("max_tokens"),
            "input_tokens_estimate": _estimate_tokens(payload),
        })
        raise
    except (APIConnectionError, APITimeoutError) as e:
        log.error({
            "event": "claude_api_network_error",
            "error_type": type(e).__name__,
        })
        raise

request_id、使用モデル、max_tokens、おおよその入力トークン数は必ず残します。この 4 つがあれば、後から「どのリクエストが重かったか」「どのモデルが不安定だったか」を遡れます。

実運用で遭遇した失敗パターン 3 つ

ここまでは設計の話でしたが、実際に運用する中で学んだことも共有します。

失敗1：ストリーミングで接続切断に気付かない。ストリーミング応答中にネットワークが切れると、クライアントによっては即座にエラーにならず、イベントがゆっくり止まるだけの挙動になります。この場合、ループ側でアイドル時間を計測して自前で切る必要があります。

async def stream_with_idle_guard(payload, idle_limit_s=20.0):
    last = time.monotonic()
    async with client.messages.stream(**payload) as stream:
        async for event in stream:
            now = time.monotonic()
            if now - last > idle_limit_s:
                raise TimeoutError("stream idle exceeded")
            last = now
            yield event

失敗2：大量プロンプトで入力トークン過多エラーが出始める。ユーザー生成データをそのまま Claude に渡している設計だと、ある日急に context オーバーフローが頻発します。入力サイズは API を叩く前に必ず上限チェックを入れ、超えそうならサマライズや分割で受けます。

失敗3：Retry-After を無視したリトライ地獄。429 を受けた時、指数バックオフだけに頼って Retry-After を読まないと、サーバ側の指示と噛み合わないリトライが連打されます。Retry-After が来ていたらそれを最優先で尊重するのが正解です。

全体を振り返ってに代えて

Claude API を本番で安定運用するコツは、派手な最適化ではなく地味な防御の積み重ねです。タイムアウトを connect と read に分ける、Retry-After を読む、request_id を残す、全体予算を逆算します。これだけの話を徹底するだけで、ユーザーから見た安定性は大きく変わります。

今日のあなたのコードに最初に足すべきは、恐らく「request_id をログに残すこと」です。そこから順に仕込んでいくと、障害が起きた時の調査時間がまず短くなります。

Claude API を Python で本番運用する — レート制限・リトライ・タイムアウトの設計