Rate Limiting and Retry Strategies with OpenAI Agents SDK: Complete Guide
Building production-grade AI agents with the OpenAI Agents SDK means dealing with the realities of distributed systems: network hiccups, API rate limits, and transient failures. Without a robust rate limiting and retry strategy, your agents will fail unpredictably under load, frustrate users, and potentially incur unnecessary costs. This guide walks through everything you need to know to make your agents resilient.
What Is Rate Limiting and Why It Matters
Rate limiting is the practice of controlling the frequency at which your application makes requests to an external API. OpenAI enforces several types of limits on its APIs, including requests per minute (RPM), tokens per minute (TPM), and daily token caps. When you exceed these limits, the API responds with an HTTP 429 status code instead of a completion.
For agent-based applications, rate limiting matters even more than for simple chat wrappers. Agents often run in loops, making multiple tool calls and LLM completions per user turn. A single user request might trigger five or ten API calls. Multiply that by concurrent users, and you can blow through your rate limits in seconds. Without proper handling, one burst of traffic can cascade into failures across your entire system.
Retry strategies complement rate limiting by gracefully handling the inevitable 429 responses and other transient errors like 500s and 503s. A well-designed retry strategy uses exponential backoff, jitter, and intelligent error classification to recover from failures without overwhelming the API further.
Understanding OpenAI's Rate Limit Headers
OpenAI returns helpful headers with every response that tell you exactly how close you are to your limits. Reading and respecting these headers is the foundation of any good rate limiting strategy.
x-ratelimit-limit-requests— Maximum requests per minute for your tierx-ratelimit-limit-tokens— Maximum tokens per minute for your tierx-ratelimit-remaining-requests— Requests remaining in the current windowx-ratelimit-remaining-tokens— Tokens remaining in the current windowx-ratelimit-reset-requests— Time until the request limit resetsx-ratelimit-reset-tokens— Time until the token limit resets
By tracking these headers, you can proactively throttle your requests before hitting a 429, rather than reacting after the fact.
Setting Up the OpenAI Agents SDK
Before implementing rate limiting, let's set up a basic agent. Make sure you have the SDK installed:
pip install openai-agentsHere's a minimal agent setup that we'll build upon:
import os
from agents import Agent, Runner
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
agent = Agent(
name="ResearchAssistant",
instructions="You are a helpful research assistant. Answer questions concisely.",
model="gpt-4o",
)
async def run_agent(user_input: str) -> str:
result = await Runner.run(agent, user_input)
return result.final_output
This works fine for a single user making occasional requests. But in production, you need more control.
Implementing a Token Bucket Rate Limiter
The token bucket algorithm is one of the most effective rate limiting strategies. It allows bursts of requests up to a maximum capacity while maintaining an average rate over time. Here's an async implementation you can use alongside the Agents SDK:
import asyncio
import time
from dataclasses import dataclass
@dataclass
class TokenBucket:
capacity: float
refill_rate: float # tokens per second
tokens: float = 0
last_refill: float = 0
lock: asyncio.Lock = None
def __post_init__(self):
self.tokens = self.capacity
self.last_refill = time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self, tokens_needed: float = 1.0) -> bool:
async with self.lock:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.refill_rate
)
self.last_refill = now
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return True
return False
async def wait_for_tokens(self, tokens_needed: float = 1.0):
while not await self.acquire(tokens_needed):
# Calculate how long to wait for enough tokens
async with self.lock:
deficit = tokens_needed - self.tokens
wait_time = deficit / self.refill_rate
await asyncio.sleep(min(wait_time, 0.5))
# Configure based on your OpenAI tier
# Example: 500 RPM = ~8.33 requests per second
request_limiter = TokenBucket(capacity=50, refill_rate=8.0)
With this in place, you can wrap your agent calls to enforce rate limiting:
async def run_agent_with_rate_limit(user_input: str) -> str:
await request_limiter.wait_for_tokens(1)
result = await Runner.run(agent, user_input)
return result.final_output
Building a Robust Retry Strategy
Even with rate limiting, you'll occasionally hit 429s and transient errors. A retry strategy with exponential backoff and jitter is essential. The OpenAI Python SDK has built-in retry logic, but when working with the Agents SDK, you often want more control.
import asyncio
import random
import logging
from typing import TypeVar, Callable, Awaitable
logger = logging.getLogger(__name__)
T = TypeVar("T")
MAX_RETRIES = 5
INITIAL_DELAY = 1.0 # seconds
MAX_DELAY = 60.0 # seconds
async def retry_with_backoff(
func: Callable[..., Awaitable[T]],
*args,
max_retries: int = MAX_RETRIES,
initial_delay: float = INITIAL_DELAY,
max_delay: float = MAX_DELAY,
**kwargs
) -> T:
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
error_type = type(e).__name__
# Check if this is a retryable error
retryable = is_retryable_error(e)
if not retryable or attempt == max_retries:
logger.error(
f"Request failed permanently after {attempt} attempts: {e}"
)
raise
# Calculate delay with exponential backoff and jitter
delay = min(
initial_delay * (2 ** attempt) + random.uniform(0, 1),
max_delay
)
logger.warning(
f"Attempt {attempt + 1} failed with {error_type}: {e}. "
f"Retrying in {delay:.2f}s"
)
await asyncio.sleep(delay)
raise last_exception
def is_retryable_error(error: Exception) -> bool:
"""Determine if an error is worth retrying."""
error_str = str(error).lower()
# Retry on rate limits
if "rate limit" in error_str or "429" in error_str:
return True
# Retry on server errors
if "500" in error_str or "502" in error_str or "503" in error_str:
return True
# Retry on timeout and connection errors
retryable_keywords = [
"timeout",
"connection error",
"connection reset",
"overloaded",
"server_error",
"api_connection_error",
"apiconnectionerror",
]
return any(keyword in error_str for keyword in retryable_keywords)
Now you can wrap any agent execution with this retry logic:
async def run_agent_resilient(user_input: str) -> str:
async def _run():
await request_limiter.wait_for_tokens(1)
result = await Runner.run(agent, user_input)
return result.final_output
return await retry_with_backoff(_run)
Respecting the Retry-After Header
When OpenAI returns a 429, it often includes a Retry-After header telling you exactly how long to wait. A sophisticated retry strategy reads this value instead of guessing with backoff. Here's how to extract and use it:
from openai import RateLimitError
async def run_agent_with_retry_after(user_input: str) -> str:
for attempt in range(MAX_RETRIES + 1):
try:
await request_limiter.wait_for_tokens(1)
result = await Runner.run(agent, user_input)
return result.final_output
except RateLimitError as e:
if attempt == MAX_RETRIES:
raise
# Try to extract retry-after from the response
retry_after = 2.0 # default fallback
if hasattr(e, 'response') and e.response is not None:
retry_after_header = e.response.headers.get('retry-after')
if retry_after_header:
try:
retry_after = float(retry_after_header)
except ValueError:
pass
# Add a small jitter to avoid thundering herd
wait_time = retry_after + random.uniform(0, 0.5)
logger.warning(f"Rate limited. Waiting {wait_time:.2f}s before retry.")
await asyncio.sleep(wait_time)
except Exception as e:
if not is_retryable_error(e) or attempt == MAX_RETRIES:
raise
delay = min(INITIAL_DELAY * (2 ** attempt), MAX_DELAY)
await asyncio.sleep(delay + random.uniform(0, 1))
Handling Concurrent Agent Runs
When running multiple agents concurrently — for example, processing a batch of user messages — you need a semaphore to limit parallelism alongside your rate limiter. This prevents overwhelming both your local resources and the API:
import asyncio
from agents import Agent, Runner
# Limit concurrent in-flight requests
MAX_CONCURRENT = 10
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def process_message(agent: Agent, message: str) -> str:
async with semaphore:
return await run_agent_resilient(message)
async def process_batch(messages: list[str]) -> list[str]:
tasks = [process_message(agent, msg) for msg in messages]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Handle any exceptions in results
final_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"Message {i} failed: {result}")
final_results.append("Error: Could not process this request.")
else:
final_results.append(result)
return final_results
Creating a Custom Model Provider with Built-in Rate Limiting
The Agents SDK allows you to customize how models are called. You can create a wrapper that automatically applies rate limiting and retries to every LLM call, including tool-call iterations. This is the most elegant approach because it centralizes the logic:
from agents import set_default_openai_client
from openai import AsyncOpenAI
import asyncio
import random
import logging
logger = logging.getLogger(__name__)
class RateLimitedOpenAIClient:
"""Wraps the AsyncOpenAI client with rate limiting and retries."""
def __init__(
self,
api_key: str,
rpm_limit: int = 500,
tpm_limit: int = 150000,
max_retries: int = 5
):
self._client = AsyncOpenAI(api_key=api_key)
self.request_limiter = TokenBucket(
capacity=rpm_limit / 6, # burst capacity
refill_rate=rpm_limit / 60.0
)
self.max_retries = max_retries
self._token_usage_estimate = 0
async def _call_with_retry(self, method, *args, **kwargs):
last_exception = None
for attempt in range(self.max_retries + 1):
try:
await self.request_limiter.wait_for_tokens(1)
return await method(*args, **kwargs)
except Exception as e:
last_exception = e
if not is_retryable_error(e) or attempt == self.max_retries:
raise
delay = min(2 ** attempt + random.uniform(0, 1), 60)
logger.warning(
f"API call failed (attempt {attempt + 1}): {e}. "
f"Retrying in {delay:.1f}s"
)
await asyncio.sleep(delay)
raise last_exception
async def chat_completions_create(self, *args, **kwargs):
return await self._call_with_retry(
self._client.chat.completions.create, *args, **kwargs
)
async def responses_create(self, *args, **kwargs):
return await self._call_with_retry(
self._client.responses.create, *args, **kwargs
)
# Delegate other attributes to the underlying client
def __getattr__(self, name):
return getattr(self._client, name)
# Set up the custom client
custom_client = RateLimitedOpenAIClient(
api_key=os.environ["OPENAI_API_KEY"],
rpm_limit=500,
tpm_limit=150000,
)
# Use it with the Agents SDK
set_default_openai_client(custom_client)
Token-Based Rate Limiting
Request-based rate limiting isn't always enough. If your agent processes long documents, a single request might consume a huge number of tokens. Token-based rate limiting tracks estimated token usage and throttles accordingly:
import tiktoken
class TokenAwareRateLimiter:
def __init__(self, tpm_limit: int = 150000):
self.tpm_limit = tpm_limit
self.tokens_used = 0
self.window_start = time.monotonic()
self.lock = asyncio.Lock()
self.encoder = tiktoken.encoding_for_model("gpt-4o")
async def estimate_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
async def acquire(self, estimated_tokens: int) -> None:
async with self.lock:
now = time.monotonic()
elapsed = now - self.window_start
# Reset window every 60 seconds
if elapsed >= 60:
self.tokens_used = 0
self.window_start = now
elapsed = 0
# If we'd exceed the limit, wait until the window resets
if self.tokens_used + estimated_tokens > self.tpm_limit:
wait_time = 60 - elapsed
logger.info(
f"Token limit approaching. Waiting {wait_time:.1f}s "
f"for window reset."
)
await asyncio.sleep(wait_time)
self.tokens_used = 0
self.window_start = time.monotonic()
self.tokens_used += estimated_tokens
async def record_actual_usage(self, usage) -> None:
"""Update with actual token usage from the API response."""
async with self.lock:
# Adjust our estimate based on actual usage
actual = usage.total_tokens
# We already added an estimate, so replace it
# This is a simplification; in production you'd track this more carefully
pass
token_limiter = TokenAwareRateLimiter(tpm_limit=150000)
async def run_agent_token_aware(user_input: str) -> str:
# Estimate input tokens
input_tokens = await token_limiter.estimate_tokens(user_input)
# Add buffer for system prompt and output
estimated_total = input_tokens + 1000
await token_limiter.acquire(estimated_total)
result = await Runner.run(agent, user_input)
# Record actual usage if available
if hasattr(result, 'usage') and result.usage:
await token_limiter.record_actual_usage(result.usage)
return result.final_output
Circuit Breaker Pattern
When the API is consistently failing, continuing to retry wastes resources and can make things worse. A circuit breaker stops all requests for a cooldown period after detecting repeated failures:
import time
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject all requests
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = 0
self.half_open_calls = 0
self.lock = asyncio.Lock()
async def can_execute(self) -> bool:
async with self.lock:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
logger.info("Circuit breaker entering half-open state")
return True
return False
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls < self.half_open_max_calls:
self.half_open_calls += 1
return True
return False
return False
async def record_success(self):
async with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.failure_count = 0
logger.info("Circuit breaker closed - service recovered")
elif self.state == CircuitState.CLOSED:
self.failure_count = 0
async def record_failure(self):
async with self.lock:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
logger.warning("Circuit breaker opened - service still failing")
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(
f"Circuit breaker opened after {self.failure_count} failures"
)
circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
async def run_agent_with_circuit_breaker(user_input: str) -> str:
if not await circuit_breaker.can_execute():
raise Exception(
"Circuit breaker is open. Service appears to be down. "
"Please try again later."
)
try:
result = await run_agent_resilient(user_input)
await circuit_breaker.record_success()
return result
except Exception as e:
await circuit_breaker.record_failure()
raise
Putting It All Together: A Production-Ready Agent Runner
Now let's combine everything into a single, production-ready agent execution function that uses rate limiting, retries, circuit breaking, and concurrency control:
import asyncio
import logging
import os
import random
import time
from agents import Agent, Runner, set_default_openai_client
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# --- Configuration ---
RPM_LIMIT = 500
TPM_LIMIT = 150000
MAX_CONCURRENT = 10
MAX_RETRIES = 5
CIRCUIT_FAILURE_THRESHOLD = 5
CIRCUIT_RECOVERY_TIMEOUT = 60
# --- Initialize components ---
request_limiter = TokenBucket(
capacity=RPM_LIMIT / 6,
refill_rate=RPM_LIMIT / 60.0
)
token_limiter = TokenAwareRateLimiter(tpm_limit=TPM_LIMIT)
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
circuit_breaker = CircuitBreaker(
failure_threshold=CIRCUIT_FAILURE_THRESHOLD,
recovery_timeout=CIRCUIT_RECOVERY_TIMEOUT
)
# --- Create agent ---
agent = Agent(
name="ProductionAgent",
instructions=(
"You are a helpful assistant. Provide accurate, concise answers. "
"If you're unsure, say so rather than guessing."
),
model="gpt-4o",
)
# --- Production runner ---
async def run_production_agent(
user_input: str,
fallback_message: str = "I'm having trouble processing your request right now. Please try again in a moment."
) -> str:
"""
Execute an agent with full rate limiting, retry, and circuit breaker protection.
Returns a fallback message instead of raising on persistent failures.
"""
async with semaphore:
# Check circuit breaker
if not await circuit_breaker.can_execute():
logger.warning("Circuit breaker open, returning fallback")
return fallback_message
# Estimate and acquire token budget
input_tokens = await token_limiter.estimate_tokens(user_input)
estimated_total = input_tokens + 1000
await token_limiter.acquire(estimated_total)
last_exception = None
for attempt in range(MAX_RETRIES + 1):
try:
# Rate limit the request
await request_limiter.wait_for_tokens(1)
# Run the agent
result = await Runner.run(agent, user_input)
# Record success
await circuit_breaker.record_success()
# Track actual usage
if hasattr(result, 'usage') and result.usage:
logger.info(
f"Token usage - prompt: {result.usage.prompt_tokens}, "
f"completion: {result.usage.completion_tokens}"
)
return result.final_output
except Exception as e:
last_exception = e
error_type = type(e).__name__
if not is_retryable_error(e) or attempt == MAX_RETRIES:
await circuit_breaker.record_failure()
logger.error(
f"Agent execution failed permanently: {error_type}: {e}"
)
return fallback_message
# Calculate backoff with jitter
delay = min(2 ** attempt + random.uniform(0, 1), 60)
logger.warning(
f"Attempt {attempt + 1}/{MAX_RETRIES} failed "
f"({error_type}): {e}. Retrying in {delay:.1f}s"
)
await asyncio.sleep(delay)
await circuit_breaker.record_failure()
return fallback_message
# --- Batch processing ---
async def process_user_messages(messages: list[str]) -> list[str]:
"""Process multiple user messages concurrently with full protection."""
tasks = [run_production_agent(msg) for msg in messages]
return await asyncio.gather(*tasks)
# --- Example usage ---
async def main():
messages = [
"What is the capital of France?",
"Explain quantum computing in simple terms.",
"Write a haiku about autumn.",
"What are the benefits of exercise?",
"How does photosynthesis work?",
]
results = await process_user_messages(messages)
for msg, result in zip(messages, results):
print(f"Q: {msg}")
print(f"A: {result}\n")
if __name__ == "__main__":
asyncio.run(main())
Monitoring and Observability
Rate limiting and retry strategies are only as good as your visibility into their behavior. You should log key metrics to understand how your system performs under load. Here's a simple metrics collector:
from dataclasses import dataclass, field
from datetime import datetime
import asyncio
@dataclass
class AgentMetrics:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
retries: int = 0
rate_limit_hits: int = 0
circuit_breaker_trips: int = 0
total_tokens_used: int = 0
avg_latency_ms: float = 0
_latency_sum: float = 0
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def record_request(
self,
success: bool,
latency_ms: float,
retries: int = 0,
tokens: int = 0,
rate_limited: bool = False
):
async with self._lock:
self.total_requests += 1
self._latency_sum += latency_ms
self.avg_latency_ms = self._latency_sum / self.total_requests
self.retries += retries
self.total_tokens_used += tokens
if success:
self.successful_requests += 1
else:
self.failed_requests += 1
if rate_limited:
self.rate_limit_hits += 1
def summary(self) -> dict:
return {
"total_requests": self.total_requests,
"success_rate": (
self.successful_requests / self.total_requests * 100
if self.total_requests > 0 else 0
),
"failed_requests": self.failed_requests,
"total_retries": self.retries,
"rate_limit_hits": self.rate_limit_hits,
"circuit_breaker_trips": self.circuit_breaker_trips,
"total_tokens_used": self.total_tokens_used,
"avg_latency_ms": round(self.avg_latency_ms, 2),
}
metrics = AgentMetrics()
# Integrate into your runner
async def run_with_metrics(user_input: str) -> str:
start_time = time.monotonic()
retries_made = 0
rate_limited = False
# ... your existing retry loop ...
# Track retries and rate limit hits as they occur
latency_ms = (time.monotonic() - start_time) * 1000
await metrics.record_request(
success=True,
latency_ms=latency_ms,
retries=retries_made,
rate_limited=rate_limited
)
# Log metrics periodically
logger.info(f"Metrics: {metrics.summary()}")
Best Practices
Here are the key principles to follow when implementing rate limiting and retry strategies with the OpenAI Agents SDK:
- Always use exponential backoff with jitter. Pure exponential backoff causes thundering herd problems when multiple clients retry simultaneously. Adding random jitter spreads retries over time.
- Classify errors before retrying. Don't retry authentication errors (401), bad requests (400), or content policy violations. Only retry transient failures like 429, 500, 502, 503, timeouts, and connection errors.
- Respect the Retry-After header. When OpenAI tells you how long to wait, listen. Guessing with backoff is a fallback, not the primary strategy.
- Set a maximum retry count. Infinite retries can hang your application. Five retries with exponential backoff is a reasonable default for most use cases.
- Use semaphores for concurrency control. Rate limiters control the rate of requests, but semaphores control how many are in flight simultaneously. You need both.
- Implement circuit breakers for cascading failure protection. If the API is consistently failing, stop hammering it. Give it time to recover.
- Monitor token usage, not just request counts. A single request with a large context window can consume more tokens than dozens of short requests. Track both RPM and TPM.
- Provide graceful fallbacks. When all retries fail, return a helpful message to the user instead of crashing. Consider queuing the request for later processing.
- Log everything. Track retry counts, rate limit hits, latency, and failure reasons. This data is invaluable for tuning your limits and identifying issues.
- Test under load. Rate limiting behavior that works in development might fail in production. Use load testing to verify your strategy handles real traffic patterns.
- Consider request prioritization. Not all requests are equal. A user-facing chat response should take priority over a background batch job. Implement priority queues when you have mixed workloads.
- Cache when possible. If users ask similar questions, caching responses can dramatically reduce your API usage. Even a short TTL cache can help with burst traffic.
Conclusion
Rate limiting and retry strategies are not optional add-ons for production AI agents — they are foundational requirements. The OpenAI Agents SDK makes it easy to build powerful multi-step agents, but that power comes with the responsibility of managing API consumption carefully. By combining token bucket rate limiting for proactive throttling, exponential backoff with jitter for reactive retries, circuit breakers for cascading failure protection, and comprehensive monitoring for visibility, you can build agents that remain reliable and responsive even under heavy load. Start with the production-ready runner from this guide, adapt the configuration to your OpenAI tier and usage patterns, and iterate based on the metrics you collect. Your users — and your API bill — will thank you.