Your AI Agent Crashes on 429 Errors and You're Burning Money

Your AI Agent Crashes on 429 Errors and You're Burning Money

Handling rate limits the right way saves you from unexpected bills and keeps your app reliable.

When you call an AI service (OpenAI, Anthropic, Cohere, etc.) you're usually limited by a rate limit – a maximum number of requests per minute or per day. If you exceed it, the API returns HTTP 429 Too Many Requests.

If your code treats a 429 like any other error and just crashes or retries blindly, two bad things happen:

#!/usr/bin/env python3 """ Example: Robust AI API caller with 429 handling. Replace the dummy `call_ai_api` with your actual SDK call. """ import time import random import logging from typing import Any, Callable # ---------------------------------------------------------------------- # Configuration – tweak these for your workload # ---------------------------------------------------------------------- MAX_RETRIES = 5 # How many times we'll retry a 429 BASE_BACKOFF = 1.0 # Seconds for the first back‑off step JITTER = 0.1 # Random jitter to avoid sync retries LOG_LEVEL = logging.

INFO logging. basicConfig ( level = LOG_LEVEL, format = " %(asctime)s %(levelname)s %(message)s ", ) # ---------------------------------------------------------------------- # Dummy AI API function – replace with real call (e.g., openai.ChatCompletion.create) # ---------------------------------------------------------------------- def call_ai_api ( prompt: str ) -> dict: """ Simulates an AI endpoint that randomly returns 429. In real code, this would be the actual HTTP request to the provider. """ # Simulate network latency time. sleep ( random. uniform ( 0.2, 0.5 )) # 20% chance of hitting a rate limit for demo purposes if random. random () < 0.2: # Pretend the server asks us to wait 2 seconds fake_response = type ( " Response ", (), {})() fake_response. status_code = 429 fake_response. headers = { " Retry-After ": " 2 " } fake_response. text = " Rate limit exceeded " return fake_response # Successful response fake_response = type ( " Response ", (), {})() fake_response. status_code = 200 fake_response. json = lambda: { " answer ": f " You said: { prompt } " } return fake_response # ---------------------------------------------------------------------- # Core retry logic # ---------------------------------------------------------------------- def robust_ai_call ( func: Callable [..., Any ], * args, max_retries: int = MAX_RETRIES, base_backoff: float = BASE_BACKOFF, jitter: float = JITTER, ** kwargs, ) -> Any: """ Calls `func(*args, **kwargs)` and retries on HTTP 429.

Returns the successful response or raises the last exception. """ attempt = 0 while True: attempt += 1 response = func ( * args, ** kwargs ) # Assume response objects have a `status_code` attribute (like requests) if getattr ( response, " status_code ", None ) == 429: if attempt > max_retries: logging. error ( f " Exceeded max retries ( { max_retries } ) for 429. Giving up.

" ) raise RuntimeError ( " Rate limit exceeded after retries " ) # Use Retry-After header if present, else exponential back‑off retry_after = response. headers. get ( " Retry-After " ) if retry_after: try: wait = float ( retry_after ) except ValueError: wait = base_backoff * ( 2 ** ( attempt - 1 )) else: wait = base_backoff * ( 2 ** ( attempt - 1 )) # Add jitter to avoid synchronized retries across many instances wait += random. uniform ( - jitter, jitter ) wait = max ( wait, 0 ) # guard against negative logging. warning ( f " Received 429 on attempt { attempt }. " f " Waiting { wait:. 2 f } s before retry...

" ) time. sleep ( wait ) continue # Any non‑429 status is treated as final (success or other error) if getattr ( response, " status_code ", None )!= 200: logging. error ( f " Non‑200 response ( { response. status_code } ): { getattr ( response, ' text ', '' ) } " ) return response # let caller decide what to do # Success! logging. info ( f " Successful call on attempt { attempt } " ) return response # ---------------------------------------------------------------------- # Example usage # ---------------------------------------------------------------------- if __name__ == " __main__ ": prompt = " Explain quantum entanglement in one sentence. " try: result = robust_ai_call ( call_ai_api, prompt ) if result. status_code == 200: print ( " AI response: ", result. json ()[ " answer " ]) else: print ( " Failed after retries. " ) except Exception as exc: print ( " Error: ", exc ) Enter fullscreen mode Exit fullscreen mode How the script works Step What happens 1️⃣ Call robust_ai_call invokes your real API function ( call_ai_api ).

2️⃣ Check 429 If the response status is 429, we look for a Retry-After header. 3️⃣ Wait Use the header value; otherwise, wait base_backoff * 2^(attempt‑1) seconds, plus a small random jitter. 4️⃣ Retry Loop up to MAX_RETRIES times.

If we exceed it, we log an error and raise. 5️⃣ Success On any non‑429 response (including 200), we return the result to the caller. 6️⃣ Logging Every attempt, wait, and outcome is logged so you can monitor traffic.

You can drop this function into any project—just replace call_ai_api with the actual request you make to your AI provider (OpenAI, Azure, HuggingFace Inference API, etc.). The retry logic stays the same.

In short, handling 429s isn't just about avoiding errors; it's a cost‑control mechanism that keeps your AI‑powered service reliable and affordable.

Have you ever seen a sudden spike in your AI bill because of unhandled rate limits? What strategies (queues, token buckets, etc.) have you tried to smooth traffic? Share your thoughts and any tips in the comments below!