OpenAI API Error Handling and Retry: 401/429/5xx Patterns

Production OpenAI API calls hit four recurring error types - each with its own handling. 401/429/5xx/timeout are not interchangeable.

1. Error taxonomy at a glance

Rule of thumb: don't retry 401/400; do retry 429/5xx/timeout.

2. 401 - fail fast, investigate auth

401 means "your request didn't authenticate." Common causes:

• API key expired or revoked
• Project quota exhausted (hard limit)
• Organization permission changed

Strategy: fail fast, surface the error to the caller, alert internally, and run an automated key health-check (e.g. daily ping to /v1/models).

3. 429 - honor Retry-After + exponential backoff

OpenAI returns a Retry-After header on 429:

HTTP/1.1 429 Too Many Requests
retry-after: 0.5

import random, time

def retry_after_429(attempt, base=1.0, cap=60.0, jitter=0.2):
    delay = min(cap, base * (2 ** attempt))
    delay *= 1 + random.uniform(-jitter, jitter)
    return max(0.1, delay)

Honor Retry-After first; fall back to exponential backoff if missing.

Common pitfalls:

• Ignoring Retry-After and retrying immediately → sustains the 429
• Multiple workers hitting the same key simultaneously → synchronized retries (thundering herd)
• Jitter-less backoff → requests cluster at the same interval

4. 5xx - 3-5 retries with request-id + degradation

5xx means upstream trouble. Strategy:

def call_api_with_5xx_retry(client, request, max_retries=5):
    for attempt in range(max_retries):
        try:
            return client.responses.create(**request)
        except APIStatusError as e:
            if e.status_code < 500:
                raise  # 4xx don't retry
            request_id = e.request_id  #  log this  -  ticket support needs it
            log.warning("5xx from OpenAI", extra={"request_id": request_id, "attempt": attempt})
            if attempt == max_retries - 1:
                # exhausted → degrade
                return degraded_response()
            time.sleep(retry_after_429(attempt, cap=30))
        except APITimeoutError:
            time.sleep(retry_after_429(attempt, base=2, cap=10))
    raise RetryExhausted()

**Key details**:

• request-id from the response header must be logged - OpenAI ticket support uses it
• After exhausting retries, degrade: switch to fallback model (GPT-5.6 Luna) or return a degraded response
• Never block the user on 5xx - front-end should show "processing" or queue state

5. Streaming errors - replay from last_successful_token

SSE-stream errors don't arrive as a single chunk - they're embedded as error events in the stream. Three things matter:

• Don't replay the full conversation - token cost doubles
• Maintain last_successful_index
• On error, replay from that index

async for event in stream:
    if event.type == "response.output_text.delta":
        output_text += event.delta
        last_successful_index = len(output_text)
    elif event.type == "error":
        # replay from last_successful_index, ( preserved content above)
        await resume_from(last_successful_index)

**Add these protections too**:

• Set a stream-level timeout (60s suggested) - abandon if exceeded
• Client-side buffer already-received tokens - survive stream disconnect
• Use event-id for client-side dedup - avoid duplicate events on reconnect

6. Error budgets

Don't retry forever. Two thresholds:

After a trip, gradually restore (half-open): probe with a few requests, fully open on recovery.

def is_circuit_open(window_minutes=5):
    recent_errors = redis.get(f"errors:{window_minutes}m") or 0
    recent_total = redis.get(f"total:{window_minutes}m") or 1
    error_rate = recent_errors / recent_total
    return error_rate > 0.05  # 5%

## 7. Battle-tested code template

from openai import OpenAI, APIStatusError, APITimeoutError, RateLimitError
import backoff
import logging

log = logging.getLogger(__name__)

class ResilientOpenAIClient:
    def __init__(self, api_key, fallback_model="gpt-5.6-luna"):
        self.client = OpenAI(api_key=api_key)
        self.fallback_model = fallback_model
    
    @backoff.on_exception(
        backoff.expo,
        (APIStatusError, APITimeoutError),
        max_tries=4,
        giveup=lambda e: isinstance(e, APIStatusError) and 400 <= e.status_code < 500,
    )
    def chat(self, **kwargs):
        try:
            return self.client.responses.create(**kwargs)
        except RateLimitError as e:
            retry_after = float(e.headers.get("retry-after", "1"))
            log.warning("429 from OpenAI", extra={"retry_after": retry_after, "request_id": e.request_id})
            time.sleep(retry_after)
            raise  # backoff rethrows for retry
        except APIStatusError as e:
            if e.status_code >= 500:
                log.error("5xx from OpenAI", extra={"request_id": e.request_id, "status": e.status_code})
            raise
    
    def chat_with_fallback(self, **kwargs):
        try:
            return self.chat(**kwargs)
        except Exception as e:
            log.exception("Primary API failed, switching to fallback", extra={"error": str(e)})
            kwargs["model"] = self.fallback_model
            return self.client.responses.create(**kwargs)

## 8. Common errors and troubleshooting

• 401 persistent → expired key; check platform.openai.com and rotate
• 429 persistent → rate-limit exceeded; upgrade tier or add organization-wide rate limit
• 5xx spikes → upstream trouble; subscribe to status.openai.com RSS; trip circuit breaker + switch to fallback
• Streaming disconnect → client not buffering received tokens; persist last_successful_index
• Repeated circuit-breaker trips → fallback model alone isn't enough; stand up a secondary provider (Azure OpenAI as second line)

9. What's Next

• OpenAI API Beginner: Your First GPT-5.6 Call Explained
• GPT-5.6 Model Selection Guide: Sol, Terra, Luna
• GPTMap Article Maintenance Guide: Re-testing, Version Sync, and Sunset - when error-rate shifts should trigger re-tests

Update log

• 2026-08-08: Initial publish