LLM APIの429・529エラー対処｜リトライ実装とエラー一覧

LLM API の 429（自分の上限超過）と 529（提供側の混雑）は、公式 SDK が初めから自動で送り直してくれます。最初にやることはリトライ処理を書くことではなく、SDK の max_retries と timeout を自分の用途に合わせて設定することです。

この記事では、Anthropic と OpenAI の公式 Python SDK を手元の模擬サーバーにつなぎ、429・529・タイムアウトのときに SDK が何回送り直し、何秒待つかを数えました。その結果をもとに、ステータス別のエラー一覧表、最悪の待ち時間の計算、retry-after を優先する待ち方、並列実行とストリーミングの扱いを、呼ぶ側の視点でまとめます。

確認環境（確認日 2026-10-06）：Ubuntu 24.04、Python 3.11.15、anthropic 1.11.0、openai 3.24.0（どちらも公式 Python SDK）。実際の API は呼ばず、この記事に全文を載せた模擬サーバー（Python 標準ライブラリ製）に SDK をつないで実行しました。エラーの種類と意味は、Anthropic と OpenAI の公式ドキュメントを今回開いて確認した範囲です。

この記事の結論

• 公式 SDK は初めから 2 回まで自動で送り直す。 対象は接続エラー・408・409・429・500 以上。まず max_retries と timeout を設定する。
• 自前のリトライを外側に重ねない。 重ねると回数が掛け算になる（3 回 × 3 回で 9 回送ることを確認）。自前で待ち方を決めたいときは max_retries=0 にしてから書く。
• 最悪の待ち時間は timeout ×（max_retries + 1）に、送り直す前の待ち時間を足した値。 初期値のままだと 10 分 × 3 回で 30 分を超える。
• 429 は自分の使い方を見直す合図、529 は提供側が混んでいる合図。 429 は retry-after に従い、529 は間隔を空けて送り直す。
• retry-after の無い 429 は、待っても通らないことがある。 利用額の上限や残高切れの可能性があるので、メッセージを読んで設定を見直す。

ステータス別エラー一覧の早見表

Anthropic の API が返すエラーです（公式ドキュメント「Errors」に記載のもの）。

OpenAI の API が返すエラーです（公式ドキュメント「Error codes」に記載のもの）。

529 は Anthropic の API が使うステータスで、OpenAI では同じ「提供側の混雑」が 503 として返ります。

Python SDK が投げる例外との対応です。自動リトライの対象は両社の SDK ドキュメントの記載で、「1 回で返る」と書いた行と 429・529・時間切れの行は今回の実行でも確かめました。

まず設定する2つ：max_retries と timeout

両社の SDK は同じ作りで、初期値も同じです。

クライアント全体に設定する書き方と、1 回の呼び出しだけ変える書き方があります。掲載のコードは模擬サーバーにつなぐため api_key="dummy" と base_url を指定しています。実際の API を呼ぶときは、この 2 つの指定を外し、公式ドキュメントの手順で API キーを設定してください。

# settings.py  まず設定する 2 つ: max_retries と timeout
import anthropic, httpx2, openai

BASE = "http://127.0.0.1:10320/rate2"          # 模擬サーバー（最初の 2 回は 429 を返す）

# クライアント全体の設定
a_client = anthropic.Anthropic(
    api_key="dummy", base_url=BASE,
    max_retries=3,                              # 初期値は 2。0 で自動リトライを止める
    timeout=httpx2.Timeout(60.0, connect=5.0),  # 初期値は 10 分。接続だけ 5 秒に短く
)
o_client = openai.OpenAI(api_key="dummy", base_url=f"{BASE}/v1", max_retries=3, timeout=60.0)

# 1 回の呼び出しだけ変える（with_options）
message = a_client.with_options(max_retries=5, timeout=20.0).messages.create(
    model="mock", max_tokens=16, messages=[{"role": "user", "content": "hi"}])
print("anthropic:", message.content[0].text, "/ request id:", message._request_id)

# 応答ヘッダーを見たいときは with_raw_response
raw = o_client.chat.completions.with_raw_response.create(
    model="mock", messages=[{"role": "user", "content": "hi"}])
print("openai:", raw.parse().choices[0].message.content, "/ request-id ヘッダー:", raw.headers.get("request-id"))

実行結果：

anthropic: OK / request id: req_mock
openai: OK / request-id ヘッダー: req_mock

設定の目安は次のとおりです（数値は例です）。

Anthropic の SDK は、ストリーミングを使わないリクエストが 10 分ほどを超えそうだと見積もると、送る前に ValueError を出します（SDK ドキュメント「Long requests」）。max_tokens を大きくするときはストリーミングに切り替えます。

SDKが何回送り直すかを数える

模擬サーバー（全文は記事の後半）に、SDK から同じリクエストを送り、サーバーが受け取った回数・間隔・最後に返った例外を記録しました。

# probe.py  公式 SDK が何回送り直し、どれだけ待つかを模擬サーバーで数える
import json, time, urllib.request
import anthropic, openai

BASE = "http://127.0.0.1:10320"

def call_anthropic(scenario, **options):
    client = anthropic.Anthropic(api_key="dummy", base_url=f"{BASE}/{scenario}", **options)
    return client.messages.create(model="mock", max_tokens=16,
                                  messages=[{"role": "user", "content": "hi"}])

def call_openai(scenario, **options):
    client = openai.OpenAI(api_key="dummy", base_url=f"{BASE}/{scenario}/v1", **options)
    return client.chat.completions.create(model="mock",
                                          messages=[{"role": "user", "content": "hi"}])

def measure(label, call, scenario, **options):
    urllib.request.urlopen(f"{BASE}/_reset").read()
    start = time.monotonic()
    try:
        call(scenario, **options)
        result = "成功"
    except Exception as e:                      # 例外の型を見るための確認用
        result = type(e).__name__
    elapsed = time.monotonic() - start
    stats = json.load(urllib.request.urlopen(f"{BASE}/_stats"))
    times = [row[0] for row in stats["log"]]
    gaps = [round(b - a, 1) for a, b in zip(times, times[1:])]
    print(f"{label:<30} 受信 {len(times)} 回  {elapsed:4.1f} 秒  間隔 {gaps}  → {result}")

print("anthropic", anthropic.__version__, "/ openai", openai.__version__)
for name, call in (("anthropic", call_anthropic), ("openai", call_openai)):
    print(f"--- {name}")
    measure("429 が 2 回（retry-after: 2）", call, "rate2")
    measure("429 が続く（retry-after 無し）", call, "cap429")
    measure("529 が続く", call, "over529")
    measure("529 が続く max_retries=4", call, "over529", max_retries=4)
    measure("529 が続く max_retries=0", call, "over529", max_retries=0)
    measure("400", call, "bad400")
    measure("401", call, "auth401")
    measure("413", call, "big413")
    measure("5 秒かかる応答 timeout=1", call, "slow", timeout=1.0)
    measure("同上 max_retries=0", call, "slow", timeout=1.0, max_retries=0)

実行結果：

anthropic 1.11.0 / openai 3.24.0
--- anthropic
429 が 2 回（retry-after: 2）      受信 3 回   4.1 秒  間隔 [2.0, 2.0]  → 成功
429 が続く（retry-after 無し）        受信 3 回   1.5 秒  間隔 [0.5, 1.0]  → RateLimitError
529 が続く                        受信 3 回   1.4 秒  間隔 [0.4, 1.0]  → OverloadedError
529 が続く max_retries=4          受信 5 回   6.7 秒  間隔 [0.5, 0.9, 1.8, 3.5]  → OverloadedError
529 が続く max_retries=0          受信 1 回   0.0 秒  間隔 []  → OverloadedError
400                            受信 1 回   0.0 秒  間隔 []  → BadRequestError
401                            受信 1 回   0.0 秒  間隔 []  → AuthenticationError
413                            受信 1 回   0.0 秒  間隔 []  → RequestTooLargeError
5 秒かかる応答 timeout=1             受信 3 回   4.3 秒  間隔 [1.4, 1.9]  → APITimeoutError
同上 max_retries=0               受信 1 回   1.0 秒  間隔 []  → APITimeoutError
--- openai
429 が 2 回（retry-after: 2）      受信 3 回   4.1 秒  間隔 [2.0, 2.0]  → 成功
429 が続く（retry-after 無し）        受信 3 回   1.4 秒  間隔 [0.4, 0.9]  → RateLimitError
529 が続く                        受信 3 回   1.4 秒  間隔 [0.4, 1.0]  → InternalServerError
529 が続く max_retries=4          受信 5 回   6.8 秒  間隔 [0.4, 0.8, 1.7, 3.9]  → InternalServerError
529 が続く max_retries=0          受信 1 回   0.0 秒  間隔 []  → InternalServerError
400                            受信 1 回   0.0 秒  間隔 []  → BadRequestError
401                            受信 1 回   0.0 秒  間隔 []  → AuthenticationError
413                            受信 1 回   0.0 秒  間隔 []  → APIStatusError
5 秒かかる応答 timeout=1             受信 3 回   4.3 秒  間隔 [1.5, 1.8]  → APITimeoutError
同上 max_retries=0               受信 1 回   1.0 秒  間隔 []  → APITimeoutError

読み取れることは 4 つです。

1. retry-after があると、SDK はその秒数どおりに待つ。 「retry-after: 2」を 2 回返すと、間隔はちょうど 2.0 秒ずつになり、3 回目で成功しました。呼ぶ側のコードは何も書いていません。
2. retry-after が無いと、0.5 秒・1 秒・2 秒・4 秒と倍々で待つ。 実際の間隔は少し短め（0.4、0.9、1.8、3.5 秒）で、これは SDK が待ち時間を最大 25% 短くするばらつき（ジッター）を入れているためです。
3. 400・401・413 は 1 回で返る。 直さないと通らないエラーなので、SDK は送り直しません。
4. タイムアウトも送り直しの対象。 timeout=1 のとき、1 秒で打ち切る → 待つ → 送り直す、を 3 回行い、合計 4.3 秒かかりました。

最悪の待ち時間を計算する

呼び出しが戻ってくるまでの最悪の時間は、次の式で見積もれます。

最悪の待ち時間 = timeout ×（max_retries + 1）+ 送り直す前の待ち時間の合計

送り直す前の待ち時間は、retry-after が無い場合で最大 0.5 秒、1 秒、2 秒、4 秒、8 秒、以降は 8 秒ずつです。

先ほどの実行結果でも、timeout=1・max_retries=2 のとき 1 × 3 = 3 秒に待ち時間（0.4 秒と 0.9 秒）が加わって 4.3 秒でした。初期値のまま画面から呼ぶと、利用者が最大で約 30 分待つ計算になります。 人が待つ処理では、timeout を先に短くします。

timeout は全体の締め切りではない

timeout に秒数を 1 つ渡した場合、それは「次のデータが届くまで待つ時間」として働きます。3 秒かけて少しずつ返す応答を timeout=1.0 で受けたところ、時間切れにならず最後まで受け取れました。

# drip.py  timeout は「全体の締め切り」ではなく、通信の区切りごとの待ち時間
import time
import anthropic

client = anthropic.Anthropic(api_key="dummy", base_url="http://127.0.0.1:10320/drip",
                             timeout=1.0, max_retries=0)
start = time.monotonic()
message = client.messages.create(model="mock", max_tokens=16,
                                 messages=[{"role": "user", "content": "hi"}])
print(f"timeout=1.0 でも成功: {message.content[0].text} / {time.monotonic() - start:.1f} 秒")

実行結果：

timeout=1.0 でも成功: OK / 2.4 秒

「必ず 30 秒以内に戻す」のような全体の締め切りが必要なときは、SDK の timeout とは別に、呼ぶ側で開始時刻からの経過時間を見て打ち切ります（後述のコードの deadline がその例です）。

自前のリトライを重ねるとどうなるか

SDK の自動リトライを知らずに、外側に for 文のリトライを書くと、回数が掛け算になります。

# stacked.py  SDK の自動リトライの外側に、自前のリトライを重ねるとどうなるか
import json, time, urllib.request
import anthropic

BASE = "http://127.0.0.1:10320"
client = anthropic.Anthropic(api_key="dummy", base_url=f"{BASE}/over529")  # max_retries は初期値の 2

start = time.monotonic()
for attempt in range(3):                      # 自前で 3 回まで試す
    try:
        client.messages.create(model="mock", max_tokens=16,
                               messages=[{"role": "user", "content": "hi"}])
        break
    except anthropic.APIStatusError as e:
        print(f"自前の {attempt + 1} 回目: {type(e).__name__}")
        time.sleep(1)

stats = json.load(urllib.request.urlopen(f"{BASE}/_stats"))
print("サーバーが受け取った回数:", stats["count"]["over529"])
print(f"かかった時間: {time.monotonic() - start:.1f} 秒")

実行結果：

自前の 1 回目: OverloadedError
自前の 2 回目: OverloadedError
自前の 3 回目: OverloadedError
サーバーが受け取った回数: 9
かかった時間: 7.0 秒

自前では 3 回のつもりでも、サーバーには 3 × 3 = 9 回届いています。混雑しているときに送る数が 3 倍になり、OpenAI のドキュメントが述べるとおり、通らなかったリクエストも分あたりの上限に数えられるため、回復がかえって遅くなります。

リトライは 1 か所にまとめます。 選び方は次のとおりです。

429と529の違い、上限超過と残高切れの見分け方

Anthropic のドキュメントによると、時間あたりの上限は 3 種類（分あたりのリクエスト数、入力トークン数、出力トークン数）で、容量は一定間隔で一気に戻るのではなく連続的に回復します。利用額の上限に達した 429 は、エラーの種類は同じ rate_limit_error のまま、retry-after ヘッダーが付かず、利用が再開されるまで通りません。OpenAI では、同じ 429 が「Rate limit reached」「Credit balance exhausted」「Organization spend limit」のように別の名前で返り、メッセージで区別できます。

そのため、429 を受け取ったら retry-after の有無とメッセージを見るのが確実です。

# classify.py  同じ 429 でも「待てば通る」か「待っても通らない」かを見分ける
import anthropic

def explain_429(error):
    retry_after = error.response.headers.get("retry-after")
    if retry_after:
        return f"時間あたりの上限。{retry_after} 秒待ってから送り直す"
    message = error.body["error"]["message"]
    return f"retry-after が無い。利用額の上限の可能性があるのでメッセージを読む: {message}"

for scenario in ("rate429", "cap429"):
    client = anthropic.Anthropic(api_key="dummy", max_retries=0,
                                 base_url=f"http://127.0.0.1:10320/{scenario}")
    try:
        client.messages.create(model="mock", max_tokens=16,
                               messages=[{"role": "user", "content": "hi"}])
    except anthropic.RateLimitError as e:
        print(e.status_code, e.body["error"]["type"], "/", explain_429(e))

実行結果：

429 rate_limit_error / 時間あたりの上限。1 秒待ってから送り直す
429 rate_limit_error / retry-after が無い。利用額の上限の可能性があるのでメッセージを読む: You have reached your API usage limits

なお、利用が急に増えたときにも 429 が返ることがあります（Anthropic のドキュメントでは acceleration limits、OpenAI では Slow down）。どちらも、送る量を少しずつ増やすことが対処として挙げられています。

残りの枠をヘッダーで見る

上限に近づいているかどうかは、応答ヘッダーで分かります。with_raw_response を使うと、SDK からヘッダーを読めます。

retry-afterを優先してジッター付きで待つコード

全体の締め切りを持たせたい、待った時間を記録したい、という場合は自前で書きます。ポイントは 4 つです。

• SDK 側は max_retries=0 にする（二重にしない）。
• retry-after があれば、その秒数を優先する。 複数の処理が同じ瞬間に再開しないよう、少しだけランダムに上乗せする。
• retry-after が無ければ、ジッター付きの指数バックオフで待つ。
• 送り直すステータスを絞り、回数と全体の締め切りの両方で打ち切る。

# retry_after.py  retry-after を優先し、無いときはジッター付きの指数バックオフで待つ
import random, time
import anthropic

RETRY_STATUS = {408, 409, 429, 500, 502, 503, 504, 529}

def wait_seconds(error, attempt, base=0.5, cap=30.0):
    """次に待つ秒数を決める。attempt は 0 始まり。"""
    header = None
    if isinstance(error, anthropic.APIStatusError):
        header = error.response.headers.get("retry-after")
    if header:
        try:
            # サーバーの指示どおり待ち、同時に再開しないよう少しだけ上乗せする
            return float(header) + random.uniform(0, 0.5)
        except ValueError:
            pass
    # 指示が無いとき: 0〜(base×2^attempt) の間でランダムに待つ（フルジッター）
    return random.uniform(0, min(cap, base * 2 ** attempt))

def call_with_retry(client, max_attempts=4, deadline=60.0, **params):
    start = time.monotonic()
    for attempt in range(max_attempts):
        try:
            return client.messages.create(**params)
        except anthropic.APIStatusError as e:
            if e.status_code not in RETRY_STATUS:
                raise                                   # 400・401 などは直してから送り直す
            error = e
        except anthropic.APIConnectionError as e:       # タイムアウトもここに含まれる
            error = e
        wait = wait_seconds(error, attempt)
        last = attempt == max_attempts - 1
        over = time.monotonic() - start + wait > deadline
        print(f"{attempt + 1} 回目: {type(error).__name__} "
              + ("→ 打ち切り" if last or over else f"→ {wait:.2f} 秒待つ"))
        if last or over:
            raise error
        time.sleep(wait)

if __name__ == "__main__":
    params = dict(model="mock", max_tokens=16, messages=[{"role": "user", "content": "hi"}])
    for scenario in ("rate2", "over2", "bad400"):
        # 自前で待つので SDK 側の自動リトライは 0 にする（二重にしない）
        client = anthropic.Anthropic(api_key="dummy", max_retries=0, timeout=10.0,
                                     base_url=f"http://127.0.0.1:10320/{scenario}")
        print(f"--- {scenario}")
        try:
            message = call_with_retry(client, **params)
            print("成功:", message.content[0].text)
        except anthropic.APIStatusError as e:
            print("そのまま返す:", e.status_code, type(e).__name__)

実行結果（待つ秒数は乱数を含むため、実行のたびに変わります）：

--- rate2
1 回目: RateLimitError → 2.23 秒待つ
2 回目: RateLimitError → 2.28 秒待つ
成功: OK
--- over2
1 回目: OverloadedError → 0.49 秒待つ
2 回目: OverloadedError → 0.69 秒待つ
成功: OK
--- bad400
そのまま返す: 400 BadRequestError

429 のときは retry-after の 2 秒に上乗せして待ち、529 のときは短い待ち時間から始め、400 は送り直さずにそのまま返しています。OpenAI の SDK でも例外クラスの名前とヘッダーの読み方は同じなので、anthropic を openai に、messages.create を chat.completions.create に置き換えれば同じ形で書けます。

並列実行：セマフォで同時実行数を抑える

たくさんのリクエストを非同期で一度に送ると、時間あたりの上限にすぐ届きます。セマフォ（同時に通れる数を決める仕組み）で、送る数を先に絞ります。 模擬サーバーは「同時に 4 件以上届いたら 429 を返す」設定です。

# semaphore.py  同時に送る数をセマフォで抑える（12 件を送る。サーバーは同時 4 件以上で 429 を返す）
import asyncio, json, time, urllib.request
import anthropic

BASE = "http://127.0.0.1:10320"

async def run(limit):
    urllib.request.urlopen(f"{BASE}/_reset").read()
    client = anthropic.AsyncAnthropic(api_key="dummy", base_url=f"{BASE}/limit3", max_retries=2)
    sem = asyncio.Semaphore(limit)

    async def one(i):
        async with sem:                       # ここで同時実行数が limit までに絞られる
            await client.messages.create(model="mock", max_tokens=16,
                                         messages=[{"role": "user", "content": f"質問 {i}"}])

    start = time.monotonic()
    results = await asyncio.gather(*(one(i) for i in range(12)), return_exceptions=True)
    await client.close()
    stats = json.load(urllib.request.urlopen(f"{BASE}/_stats"))
    errors = sum(isinstance(r, Exception) for r in results)
    print(f"同時 {limit:2} 件まで: 成功 {12 - errors:2} 件 / エラー {errors} 件 / "
          f"サーバーの受信 {stats['count']['limit3']:2} 回 / 同時処理の最大 {stats['peak']:2} / "
          f"{time.monotonic() - start:.1f} 秒")

asyncio.run(run(12))   # 絞らない（12 件を一度に送る）
asyncio.run(run(3))    # 3 件までに絞る

実行結果（絞らない場合の件数は、実行のたびに変わります）：

同時 12 件まで: 成功 11 件 / エラー 1 件 / サーバーの受信 27 回 / 同時処理の最大  6 / 3.5 秒
同時  3 件まで: 成功 12 件 / エラー 0 件 / サーバーの受信 12 回 / 同時処理の最大  3 / 2.2 秒

12 件を一度に送ると、429 と自動リトライが繰り返されてサーバーには 27 回届き、それでも 1 件はエラーで終わりました。3 件までに絞ると、12 回ちょうどで全件が成功し、かかった時間も短くなっています。リトライで取り戻すより、先に絞るほうが速くて確実です。

同時実行数の決め方の目安です。

ストリーミング途中で切れたときの扱い

ストリーミングでは、ステータス 200 を返したあとでエラーになることがあります。Anthropic のドキュメントには、混雑時にイベントの流れの中で次のようなエラーイベントが届くと書かれています。

event: error
data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}

このとき SDK の自動リトライは働きません。模擬サーバーで 3 通りの止まり方を再現したところ、次の結果でした。

つまり、ストリーミングでは 「例外が出なかった」だけで完了と判断せず、stop_reason が入っているかを確かめる必要があります。

Anthropic のドキュメントは、途中まで届いた内容を保存し、続きを頼むリクエストを送る方法を挙げています。Claude 4.6 以降のモデルでは、届いた内容と「続きを書いてほしい」という指示を user メッセージに入れます（Claude 4.5 以前では、届いた内容を assistant メッセージの書き出しとして渡します）。また、ツール呼び出しと拡張思考のブロックは途中から再開できず、再開できるのは直近のテキストブロックからです。

# stream_resume.py  ストリーミングの途中で切れたとき、届いた分を残して続きを頼む
import anthropic, httpx2

def stream_once(client, messages):
    """届いた文字を貯めながら読む。戻り値は (本文, 完了したか, 止まった理由)。"""
    text = ""
    try:
        with client.messages.stream(model="mock", max_tokens=256, messages=messages) as stream:
            for chunk in stream.text_stream:
                text += chunk
            stop_reason = stream.current_message_snapshot.stop_reason
        # 例外が出なくても、stop_reason が入っていなければ最後まで届いていない
        return text, stop_reason is not None, f"stop_reason={stop_reason}"
    except (anthropic.APIError, httpx2.HTTPError) as e:
        return text, False, type(e).__name__

def ask(make_client, question):
    messages = [{"role": "user", "content": question}]
    text, done, error = stream_once(make_client(0), messages)
    tries = 1
    while not done and tries < 3:
        print(f"途中で止まった: {error} / 届いた分: {text!r}")
        # 届いた分を渡して、続きだけを書いてもらう
        follow = messages + [{"role": "user", "content":
            f"Your previous response was interrupted and ended with {text}. "
            "Continue from where you left off."}]
        more, done, error = stream_once(make_client(tries), follow)
        text += more
        tries += 1
    return text, done

if __name__ == "__main__":
    for first in ("sse_error", "sse_cut", "sse_eof"):
        # 1 回目は途中で止まる応答、2 回目からは最後まで届く応答を返す模擬サーバーにつなぐ
        def make_client(n, first=first):
            scenario = first if n == 0 else "sse_ok"
            return anthropic.Anthropic(api_key="dummy", base_url=f"http://127.0.0.1:10320/{scenario}")
        print(f"--- {first}")
        text, done = ask(make_client, "設定の手順を教えて")
        print("完了:", done, "/ 送った回数:", 2 if done else 3)

実行結果：

--- sse_error
途中で止まった: APIStatusError / 届いた分: '手順は3つです。1. 設定を開く。'
完了: True / 送った回数: 2
--- sse_cut
途中で止まった: RemoteProtocolError / 届いた分: '手順は3つです。1. 設定を開く。'
完了: True / 送った回数: 2
--- sse_eof
途中で止まった: stop_reason=None / 届いた分: '手順は3つです。1. 設定を開く。'
完了: True / 送った回数: 2

この実行は、止まったことを検知して 2 回目を送る流れの確認です。模擬サーバーは決まった文を返すだけなので、続きの文章の出来ばえは実際のモデルで確かめてください。

扱い方の選び方です。

模擬サーバーの全文

この記事の実行結果は、すべて次のサーバーに SDK をつないで得たものです。URL の先頭の文字列（rate2、over529 など）で返し方を切り替えます。自分のリトライ処理を試す台としても使えます。

# mock_server.py  LLM API の代わりに 429・529・遅延・途中で切れるストリームを返す模擬サーバー
# 使い方: python3 mock_server.py 10320
# 動作は URL の先頭で切り替える（例: http://127.0.0.1:10320/rate2/v1/messages）
import json, sys, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

LOCK = threading.Lock()
COUNT = {}          # シナリオごとの受信回数
ACTIVE = 0          # いま処理中のリクエスト数
PEAK = 0            # 同時処理数の最大
LOG = []            # (経過秒, シナリオ, 何回目, x-stainless-retry-count)
T0 = time.monotonic()

def ok_body(path):
    if path.endswith("/messages"):
        return {"id": "msg_mock", "type": "message", "role": "assistant", "model": "mock",
                "content": [{"type": "text", "text": "OK"}], "stop_reason": "end_turn",
                "stop_sequence": None, "usage": {"input_tokens": 5, "output_tokens": 1}}
    return {"id": "chatcmpl-mock", "object": "chat.completion", "created": 0, "model": "mock",
            "choices": [{"index": 0, "finish_reason": "stop",
                         "message": {"role": "assistant", "content": "OK"}}],
            "usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6}}

def err(kind, message):
    return {"type": "error", "error": {"type": kind, "message": message}, "request_id": "req_mock"}

class H(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    def log_message(self, *a): pass

    def send_json(self, status, body, headers=None):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(data)))
        self.send_header("request-id", "req_mock")
        for k, v in (headers or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        global PEAK
        if self.path == "/_stats":
            with LOCK:
                self.send_json(200, {"count": COUNT, "peak": PEAK, "log": LOG})
        elif self.path == "/_reset":
            with LOCK:
                COUNT.clear(); LOG.clear(); PEAK = 0
            self.send_json(200, {"reset": True})

    def do_POST(self):
        global ACTIVE, PEAK
        n = int(self.headers.get("content-length", 0))
        req = json.loads(self.rfile.read(n) or b"{}")
        scenario = self.path.split("/")[1]
        with LOCK:
            COUNT[scenario] = COUNT.get(scenario, 0) + 1
            nth = COUNT[scenario]
            ACTIVE += 1
            PEAK = max(PEAK, ACTIVE)
            active_now = ACTIVE
            LOG.append((round(time.monotonic() - T0, 2), scenario, nth,
                        self.headers.get("x-stainless-retry-count")))
        try:
            self.route(scenario, nth, active_now, req)
        except (BrokenPipeError, ConnectionResetError):
            pass
        finally:
            with LOCK:
                ACTIVE -= 1

    def route(self, scenario, nth, active_now, req):
        if scenario == "rate2" and nth <= 2:        # 2回だけ 429（retry-after: 2）
            return self.send_json(429, err("rate_limit_error", "rate limited"), {"retry-after": "2"})
        if scenario == "rate429":                   # いつも 429（retry-after: 1）
            return self.send_json(429, err("rate_limit_error", "rate limited"), {"retry-after": "1"})
        if scenario == "cap429":                    # 上限額に達した形の 429（retry-after なし）
            return self.send_json(429, err("rate_limit_error", "You have reached your API usage limits"))
        if scenario == "over529":                   # いつも 529（retry-after なし）
            return self.send_json(529, err("overloaded_error", "Overloaded"))
        if scenario == "over2" and nth <= 2:        # 2回だけ 529
            return self.send_json(529, err("overloaded_error", "Overloaded"))
        if scenario == "bad400":
            return self.send_json(400, err("invalid_request_error", "max_tokens: Field required"))
        if scenario == "auth401":
            return self.send_json(401, err("authentication_error", "invalid x-api-key"))
        if scenario == "big413":
            return self.send_json(413, err("request_too_large", "Request exceeds the maximum size"))
        if scenario == "slow":                      # 5秒かかる応答
            time.sleep(5)
        if scenario == "drip":                      # 0.6秒おきに少しずつ、合計3秒かけて返す
            data = json.dumps(ok_body(self.path)).encode()
            self.send_response(200)
            self.send_header("content-type", "application/json")
            self.send_header("content-length", str(len(data)))
            self.end_headers()
            step = len(data) // 5 + 1
            for i in range(0, len(data), step):
                self.wfile.write(data[i:i + step]); self.wfile.flush()
                time.sleep(0.6)
            return
        if scenario == "limit3":                    # 同時に4件以上来たら 429
            if active_now > 3:
                return self.send_json(429, err("rate_limit_error", "too many concurrent"), {"retry-after": "1"})
            time.sleep(0.5)
        if scenario in ("sse_error", "sse_cut", "sse_eof") or (scenario == "sse_ok" and req.get("stream")):
            return self.stream(scenario)
        self.send_json(200, ok_body(self.path))

    def stream(self, scenario):
        self.send_response(200)
        self.send_header("content-type", "text/event-stream")
        self.send_header("request-id", "req_mock")
        self.send_header("transfer-encoding", "chunked")
        self.end_headers()
        def ev(name, data):
            body = f"event: {name}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode()
            self.wfile.write(f"{len(body):x}\r\n".encode() + body + b"\r\n")
            self.wfile.flush()
        ev("message_start", {"type": "message_start", "message": {
            "id": "msg_mock", "type": "message", "role": "assistant", "model": "mock", "content": [],
            "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 5, "output_tokens": 0}}})
        ev("content_block_start", {"type": "content_block_start", "index": 0,
                                   "content_block": {"type": "text", "text": ""}})
        parts = ["手順は3つです。", "1. 設定を開く。", "2. 値を入力する。", "3. 保存する。"]
        stop_at = len(parts) if scenario == "sse_ok" else 2
        for p in parts[:stop_at]:
            ev("content_block_delta", {"type": "content_block_delta", "index": 0,
                                       "delta": {"type": "text_delta", "text": p}})
            time.sleep(0.1)
        if scenario == "sse_error":                 # 200 を返したあとでエラーイベント
            ev("error", {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}})
        elif scenario == "sse_cut":                 # 終わりの印を送らずに接続が切れる
            self.close_connection = True
            return
        elif scenario == "sse_eof":                 # 終了イベントを送らずに本文だけ正常に閉じる
            pass
        else:
            ev("content_block_stop", {"type": "content_block_stop", "index": 0})
            ev("message_delta", {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None},
                                 "usage": {"output_tokens": 20}})
            ev("message_stop", {"type": "message_stop"})
        self.wfile.write(b"0\r\n\r\n")
        self.close_connection = True

if __name__ == "__main__":
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 10320
    ThreadingHTTPServer(("127.0.0.1", port), H).serve_forever()

起動と確認の手順：

python3 mock_server.py 10320 &
python3 probe.py

指数バックオフ・レート制限の実装との違い

似た話題と、この記事の範囲を整理します。

どこから手を付けるかの判断基準です。

確かめて分かったこと

• 両社の SDK は、初期値（2 回・10 分・0.5 秒から最大 8 秒）も待ち方も同じだった。 retry-after を返すと、どちらもその秒数どおり（2.0 秒）に待った。
• 自前の 3 回リトライを重ねると、サーバーには 9 回届いた。 回数は足し算ではなく掛け算になる。
• 529 を OpenAI の SDK に返すと InternalServerError になり、Anthropic の SDK では OverloadedError になった。 500 以上をまとめて捕まえるなら、どちらの SDK でも InternalServerError で受けられる。
• timeout=1.0 でも、少しずつ届く応答は 2.4 秒かけて最後まで受け取れた。 timeout は全体の締め切りではない。
• ストリーミング途中のエラーイベントは、status_code が 200 の APIStatusError として届き、自動では送り直されなかった。
• 終了イベントが無いまま本文が終わる応答では、例外が出なかった。 完了の判定には stop_reason を使う。

よくある質問

Q. 429 エラーとは何ですか？ A. 自分の組織が、時間あたりの上限か利用額の上限に達したことを示すエラーです。retry-after ヘッダーがあればその秒数だけ待てば通り、無い場合は利用額の上限や残高を確認します。

Q. 429 と 529 の違いは何ですか？ A. 429 は自分の使い方が原因、529 は提供側の混雑が原因です。429 は送る速さと同時実行数を見直し、529 は間隔を空けて送り直します。OpenAI では混雑が 503 として返ります。

Q. リトライ処理は自分で書く必要がありますか？ A. 多くの場合は不要です。公式 SDK が接続エラー・408・409・429・500 以上を初めから 2 回まで送り直すので、max_retries と timeout の設定で足ります。自前で書くときは max_retries=0 にして、リトライを 1 か所にまとめます。

Q. リトライしても 429 が直らないときはどうすればよいですか？ A. retry-after ヘッダーが付いているかを確認します。付いていなければ、待っても通らない種類（利用額の上限、残高切れ）の可能性が高いので、エラーメッセージを読み、管理画面で上限と残高を見直します。付いている場合は、同時実行数を下げます。

Q. タイムアウトは何秒にすればよいですか？ A. 「利用者が待てる時間 ÷（max_retries + 1）」より短くするのが目安です。たとえば 90 秒まで待てて max_retries が 2 なら、timeout は 30 秒弱にします。長い出力はストリーミングに切り替えます。

まとめ

• 公式 SDK は初めから自動でリトライする。 まず max_retries と timeout を設定する。
• 最悪の待ち時間は timeout ×（max_retries + 1）+ 待ち時間。 初期値のままだと約 30 分になる。
• 自前のリトライは重ねない。 書くなら max_retries=0 にして 1 か所にまとめる。
• 429 は retry-after に従う。 retry-after が無ければ、利用額の上限や残高を確認する。
• 529（OpenAI では 503）は提供側の混雑。 間隔を空けて送り直す。
• 並列実行はセマフォで先に絞る。 リトライで取り戻すより速く、確実に通る。
• ストリーミングは 200 のあとでも止まる。 stop_reason で完了を確かめ、届いた分を残して続きを頼む。

参考・出典

• Anthropic「Errors」 https://platform.claude.com/docs/en/api/errors （2026-10-06 確認）
• Anthropic「Rate limits」 https://platform.claude.com/docs/en/api/rate-limits （2026-10-06 確認）
• Anthropic「Streaming Messages」 https://platform.claude.com/docs/en/build-with-claude/streaming （2026-10-06 確認）
• Anthropic「Python SDK」 https://platform.claude.com/docs/en/api/sdks/python （2026-10-06 確認）
• OpenAI「Error codes」 https://developers.openai.com/api/docs/guides/error-codes （2026-10-06 確認）
• OpenAI「Rate limits」 https://developers.openai.com/api/docs/guides/rate-limits （2026-10-06 確認）
• OpenAI Python ライブラリ README https://github.com/openai/openai-python （2026-10-06 確認）

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。