Introduction to Error Recovery in the OpenAI Agents SDK
Building production-grade AI agents means accepting one fundamental truth: things will fail. API calls will time out, models will return malformed output, tools will throw exceptions, and rate limits will bite you at the worst possible moment. The OpenAI Agents SDK provides a flexible framework for orchestrating LLM-powered agents, but it does not magically shield your application from these failures. That is where error recovery patterns come in.
Error recovery patterns are structured strategies for detecting, classifying, and responding to failures during agent execution. Instead of letting a single failed tool call or a transient network error crash your entire workflow, these patterns allow your agents to degrade gracefully, retry intelligently, and even reason about their own mistakes. In this guide, we will explore the most important error recovery patterns available when working with the OpenAI Agents SDK, complete with practical, copy-paste-ready code examples.
Why Error Recovery Matters
When you move from a prototype to a production deployment, the reliability expectations shift dramatically. A demo that works 90% of the time is impressive in a hackathon but unacceptable in a customer-facing product. Here are the core reasons error recovery is non-negotiable:
- Transient failures are common: Network blips, rate limits, and temporary model unavailability happen regularly at scale.
- Tool calls are unpredictable: External APIs, databases, and custom functions can fail for reasons outside your control.
- Model output is non-deterministic: Even with structured outputs, models occasionally produce invalid JSON or unexpected formats.
- User trust is fragile: A single unhandled crash can destroy confidence in an AI system.
- Cost matters: Naive retry strategies can multiply your token costs and latency if not designed carefully.
A well-designed error recovery layer transforms your agent from a fragile script into a resilient system that can survive the messy reality of production traffic.
Understanding the OpenAI Agents SDK Execution Model
Before diving into specific patterns, it is important to understand how the OpenAI Agents SDK executes agents and where errors can occur. The SDK revolves around a few core primitives: Agent, Runner, and tool functions. When you call Runner.run(), the SDK enters a loop where the model generates responses, decides whether to call tools, executes those tools, and feeds results back to the model until a final response is produced.
Errors can surface at several points in this loop:
- Model API calls: Rate limits, timeouts, authentication errors, or server errors from OpenAI.
- Tool execution: Exceptions raised inside your custom tool functions.
- Output parsing: Failures when the model output does not match the expected schema.
- Guardrail violations: Input or output guardrails rejecting content.
- Max turns exceeded: The agent loops too many times without producing a final answer.
Each of these failure modes calls for a different recovery strategy. Let us explore them one by one.
Pattern 1: Retry with Exponential Backoff
The most fundamental error recovery pattern is retrying transient failures with exponential backoff. This is especially relevant for model API calls that fail due to rate limits (HTTP 429) or temporary server errors (HTTP 5xx). The OpenAI Agents SDK does not automatically retry these for you in all cases, so wrapping your runner calls in a retry loop is a good practice.
Basic Retry Wrapper
import asyncio
import random
from openai import RateLimitError, APIConnectionError, APITimeoutError
from agents import Agent, Runner
async def run_with_retry(
agent: Agent,
prompt: str,
max_retries: int = 4,
base_delay: float = 1.0
):
"""
Run an agent with exponential backoff retry for transient errors.
"""
for attempt in range(max_retries + 1):
try:
result = await Runner.run(agent, prompt)
return result
except (RateLimitError, APIConnectionError, APITimeoutError) as exc:
if attempt == max_retries:
raise
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"Attempt {attempt + 1} failed: {exc}. Retrying in {delay:.1f}s")
await asyncio.sleep(delay)
# Usage
agent = Agent(name="Helper", instructions="You are a helpful assistant.")
async def main():
try:
result = await run_with_retry(agent, "Explain quantum computing briefly.")
print(result.final_output)
except Exception as exc:
print(f"All retries exhausted: {exc}")
asyncio.run(main())
The key elements of this pattern are the jitter (random uniform addition) to avoid thundering herd problems, the exponential delay growth to respect rate limit windows, and the selective exception handling so that non-transient errors propagate immediately.
Pattern 2: Tool-Level Error Handling and Self-Correction
One of the most powerful features of the OpenAI Agents SDK is that tool errors can be fed back to the model as context. Instead of crashing when a tool raises an exception, you can catch the error, return it as a string result, and let the agent decide how to recover. This enables genuine self-correction: the model sees what went wrong and can adjust its approach.
Returning Errors as Tool Results
from agents import Agent, Runner, function_tool
import json
@function_tool
def query_database(sql: str) -> str:
"""Execute a SQL query against the product database."""
try:
# Simulated database call
if "DROP" in sql.upper():
raise ValueError("Destructive operations are not allowed.")
if not sql.strip().upper().startswith("SELECT"):
raise ValueError("Only SELECT queries are permitted.")
# Simulated successful query
return json.dumps({"rows": [{"id": 1, "name": "Widget"}], "count": 1})
except Exception as exc:
# Return the error message as the tool result
# The agent will see this and can self-correct
return f"ERROR: {str(exc)}. Please revise your query and try again."
agent = Agent(
name="DBAgent",
instructions=(
"You are a database assistant. Use the query_database tool to "
"retrieve data. If a tool returns an ERROR, read the message "
"carefully and fix your approach before trying again."
),
tools=[query_database],
)
async def main():
result = await Runner.run(
agent,
"Get all products from the database. If that doesn't work, "
"try a different approach."
)
print(result.final_output)
asyncio.run(main())
In this pattern, the error message itself becomes instructional. The model reads "Only SELECT queries are permitted" and can adjust its next tool call accordingly. This is far more robust than throwing an exception that terminates the entire run.
Structured Tool Error Handling
For more complex scenarios, you can return structured error objects that give the model richer context about what went wrong and what options it has:
from pydantic import BaseModel
from typing import Literal
class ToolError(BaseModel):
status: Literal["error"] = "error"
error_type: str
message: str
suggestion: str
@function_tool
def fetch_weather(city: str, units: str = "celsius") -> str:
"""Fetch current weather for a city."""
valid_cities = ["New York", "London", "Tokyo", "Sydney"]
valid_units = ["celsius", "fahrenheit"]
if city not in valid_cities:
error = ToolError(
error_type="invalid_city",
message=f"City '{city}' is not supported.",
suggestion=f"Try one of: {', '.join(valid_cities)}"
)
return error.model_dump_json()
if units not in valid_units:
error = ToolError(
error_type="invalid_units",
message=f"Units '{units}' not recognized.",
suggestion=f"Use one of: {', '.join(valid_units)}"
)
return error.model_dump_json()
return json.dumps({"city": city, "temperature": 22, "units": units})
By including a suggestion field, you guide the model toward a successful retry without requiring it to guess what went wrong.
Pattern 3: Fallback Agents and Model Degradation
Sometimes the primary model is unavailable or consistently failing. In these cases, having a fallback agent that uses a different (perhaps cheaper or more available) model can keep your system running. The OpenAI Agents SDK makes this straightforward since each agent can be configured with a specific model.
Cascading Fallback
from agents import Agent, Runner, OpenAIChatCompletionsModel
from openai import AsyncOpenAI
# Primary agent using a powerful model
primary_agent = Agent(
name="PrimaryAgent",
instructions="You are a precise, thorough assistant.",
model="gpt-4o",
)
# Fallback agent using a lighter, more available model
fallback_agent = Agent(
name="FallbackAgent",
instructions="You are a helpful assistant. Provide concise answers.",
model="gpt-4o-mini",
)
async def run_with_fallback(prompt: str):
"""
Try the primary agent first, fall back to a lighter model on failure.
"""
try:
result = await Runner.run(primary_agent, prompt)
return result.final_output, "primary"
except Exception as exc:
print(f"Primary agent failed: {exc}. Falling back to lighter model.")
try:
result = await Runner.run(fallback_agent, prompt)
return result.final_output, "fallback"
except Exception as fallback_exc:
print(f"Fallback also failed: {fallback_exc}")
return "I'm sorry, I'm unable to process your request right now.", "failed"
async def main():
answer, source = await run_with_fallback("Summarize the French Revolution.")
print(f"[Source: {source}]")
print(answer)
asyncio.run(main())
This pattern is particularly useful when you have strict uptime requirements. The fallback does not need to be as capable as the primary; it just needs to provide a reasonable response so the user is not left hanging.
Pattern 4: Circuit Breaker for Repeated Failures
When an external dependency (like a tool that calls a third-party API) is consistently failing, retrying endlessly wastes resources and increases latency. A circuit breaker pattern monitors failure rates and temporarily stops attempting calls when a threshold is exceeded, allowing the system to recover.
Implementing a Circuit Breaker
import time
from enum import Enum
from agents import function_tool
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject calls
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
success_threshold: int = 2,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = 0
def record_success(self):
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
elif self.state == CircuitState.CLOSED:
self.failure_count = 0
def record_failure(self):
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.success_count = 0
elif self.state == CircuitState.CLOSED:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def can_execute(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.success_count = 0
return True
return False
return True # HALF_OPEN allows limited testing
# Global circuit breaker for an external API
payment_circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)
@function_tool
def process_payment(amount: float, currency: str) -> str:
"""Process a payment through the payment gateway."""
if not payment_circuit.can_execute():
return (
"ERROR: Payment service is temporarily unavailable. "
"Please inform the user to try again in a few minutes."
)
try:
# Simulated payment processing
if amount <= 0:
raise ValueError("Amount must be positive")
payment_circuit.record_success()
return json.dumps({"status": "success", "amount": amount, "currency": currency})
except Exception as exc:
payment_circuit.record_failure()
return f"ERROR: Payment failed - {str(exc)}"
With this pattern, after repeated failures the circuit opens and immediately returns a graceful error message without even attempting the call. After the recovery timeout, it enters a half-open state to test whether the service has recovered.
Pattern 5: Guardrails with Graceful Rejection
The OpenAI Agents SDK supports input and output guardrails that validate content before and after processing. Instead of treating guardrail failures as crashes, you can use them as recovery points where the agent gets feedback and tries a different approach.
Output Guardrail with Retry
from agents import (
Agent,
Runner,
GuardrailFunctionOutput,
OutputGuardrailTripwireTriggered,
output_guardrail,
)
from pydantic import BaseModel
class ResponseCheck(BaseModel):
contains_pii: bool
reasoning: str
@output_guardrail
async def no_pii_guardrail(agent, output):
"""Check if the agent output contains personally identifiable information."""
response_check = ResponseCheck(
contains_pii="ssn" in output.lower() or "social security" in output.lower(),
reasoning="Checking for PII patterns in the response."
)
return GuardrailFunctionOutput(
output_info=response_check,
tripwire_triggered=response_check.contains_pii,
)
agent = Agent(
name="SafeAgent",
instructions=(
"You are a helpful assistant. Never include personal information "
"like SSNs, credit card numbers, or addresses in your responses."
),
output_guardrails=[no_pii_guardrail],
)
async def run_with_guardrail_recovery(prompt: str, max_attempts: int = 3):
"""
Run the agent and recover from guardrail violations by
asking the agent to regenerate without the offending content.
"""
current_prompt = prompt
for attempt in range(max_attempts):
try:
result = await Runner.run(agent, current_prompt)
return result.final_output
except OutputGuardrailTripwireTriggered:
print(f"Guardrail triggered on attempt {attempt + 1}. Asking for revision.")
current_prompt = (
f"Your previous response contained sensitive personal information. "
f"Please answer the following question again, but this time do NOT "
f"include any SSNs, credit card numbers, or personal addresses: {prompt}"
)
return "I'm unable to provide a response that meets our safety requirements."
async def main():
answer = await run_with_guardrail_recovery(
"What is John Doe's SSN and address? Just make something up for an example."
)
print(answer)
asyncio.run(main())
This pattern turns guardrail violations from hard failures into soft recovery points. The agent gets a chance to fix its output based on explicit feedback about what went wrong.
Pattern 6: Max Turns and Infinite Loop Prevention
Agents can sometimes get stuck in loops, repeatedly calling tools without converging on a final answer. The SDK allows you to set a max_turns limit, but you should also handle the resulting MaxTurnsExceeded exception gracefully.
Handling Max Turns Exceeded
from agents import Agent, Runner, MaxTurnsExceeded
agent = Agent(
name="ResearchAgent",
instructions=(
"You are a research assistant. Use tools to find information, "
"then provide a clear summary. Do not call tools more than necessary."
),
tools=[query_database, fetch_weather],
)
async def run_bounded_agent(prompt: str, max_turns: int = 10):
"""
Run an agent with a turn limit and graceful handling of exceeded turns.
"""
try:
result = await Runner.run(agent, prompt, max_turns=max_turns)
return result.final_output
except MaxTurnsExceeded:
# The agent ran out of turns. Try a more direct prompt.
print("Max turns exceeded. Retrying with a more constrained prompt.")
direct_prompt = (
f"Answer this question directly without using any tools. "
f"If you don't know, say so. Question: {prompt}"
)
result = await Runner.run(agent, direct_prompt, max_turns=3)
return result.final_output
async def main():
answer = await run_bounded_agent("What's the weather and latest products?")
print(answer)
asyncio.run(main())
The recovery strategy here is to switch from an exploratory mode (where the agent uses tools freely) to a direct mode (where the agent must answer without tools). This ensures the user always gets some response, even if it is less detailed than ideal.
Pattern 7: Comprehensive Error Boundary
In production, you want a single error boundary that catches all unhandled exceptions, logs them, and returns a user-friendly message. This is your last line of defense.
Full Error Boundary Implementation
import logging
import traceback
from datetime import datetime
from agents import Agent, Runner
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger("agent_error_boundary")
# Track error rates for monitoring
error_log = []
async def safe_agent_run(
agent: Agent,
prompt: str,
user_id: str = "anonymous",
) -> dict:
"""
Comprehensive error boundary for agent execution.
Returns a structured response with status and content.
"""
start_time = datetime.now()
try:
result = await Runner.run(agent, prompt)
return {
"status": "success",
"content": result.final_output,
"user_id": user_id,
"duration_ms": (datetime.now() - start_time).total_seconds() * 1000,
}
except MaxTurnsExceeded:
logger.warning(f"Max turns exceeded for user {user_id}")
return {
"status": "partial",
"content": (
"I was working on your request but ran into complexity limits. "
"Could you try simplifying your question?"
),
"user_id": user_id,
"duration_ms": (datetime.now() - start_time).total_seconds() * 1000,
}
except OutputGuardrailTripwireTriggered:
logger.warning(f"Guardrail triggered for user {user_id}")
return {
"status": "blocked",
"content": (
"I can't provide that type of response. "
"Could you rephrase your request?"
),
"user_id": user_id,
"duration_ms": (datetime.now() - start_time).total_seconds() * 1000,
}
except (RateLimitError, APIConnectionError, APITimeoutError) as exc:
logger.error(f"Transient API error for user {user_id}: {exc}")
error_log.append({"user_id": user_id, "error": str(exc), "time": datetime.now()})
return {
"status": "unavailable",
"content": (
"I'm experiencing temporary connectivity issues. "
"Please try again in a moment."
),
"user_id": user_id,
"duration_ms": (datetime.now() - start_time).total_seconds() * 1000,
}
except Exception as exc:
logger.error(
f"Unexpected error for user {user_id}: {exc}\n{traceback.format_exc()}"
)
error_log.append({"user_id": user_id, "error": str(exc), "time": datetime.now()})
return {
"status": "error",
"content": (
"Something went wrong on our end. Our team has been notified. "
"Please try again later."
),
"user_id": user_id,
"duration_ms": (datetime.now() - start_time).total_seconds() * 1000,
}
# Usage in a web handler context
async def handle_user_message(user_id: str, message: str):
agent = Agent(name="ChatAgent", instructions="You are a helpful assistant.")
response = await safe_agent_run(agent, message, user_id)
# In production, you would send this to your frontend
print(f"[{response['status']}] {response['content']}")
return response
This error boundary ensures that no matter what goes wrong, the user receives a coherent message and your system logs the details for debugging. The structured return format also makes it easy to integrate with monitoring and alerting systems.
Pattern 8: Checkpointing for Long-Running Agents
For agents that perform long-running tasks with multiple steps, checkpointing allows you to save intermediate state so that if a failure occurs partway through, you can resume from the last successful checkpoint rather than starting over.
State Checkpointing
import json
import os
from agents import Agent, Runner, function_tool
CHECKPOINT_DIR = "checkpoints"
def save_checkpoint(session_id: str, state: dict):
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
path = os.path.join(CHECKPOINT_DIR, f"{session_id}.json")
with open(path, "w") as f:
json.dump(state, f)
def load_checkpoint(session_id: str) -> dict | None:
path = os.path.join(CHECKPOINT_DIR, f"{session_id}.json")
if os.path.exists(path):
with open(path, "r") as f:
return json.load(f)
return None
@function_tool
def save_progress(session_id: str, step: str, data: str) -> str:
"""Save progress for the current session so work can be resumed."""
existing = load_checkpoint(session_id) or {"completed_steps": []}
existing["completed_steps"].append({"step": step, "data": data})
save_checkpoint(session_id, existing)
return f"Progress saved at step: {step}"
@function_tool
def load_progress_tool(session_id: str) -> str:
"""Load previously saved progress for a session."""
state = load_checkpoint(session_id)
if state is None:
return "No previous progress found for this session."
return json.dumps(state)
agent = Agent(
name="LongRunningAgent",
instructions=(
"You are a task execution agent. For multi-step tasks, save your "
"progress after each step using save_progress. If you need to resume "
"a task, first load previous progress with load_progress_tool."
),
tools=[save_progress, load_progress_tool],
)
async def run_with_checkpoint_recovery(session_id: str, task: str):
"""
Run a long task with checkpoint-based recovery.
"""
# Check for existing progress
existing = load_checkpoint(session_id)
if existing:
prompt = (
f"Resume the following task. You previously completed these steps: "
f"{json.dumps(existing['completed_steps'])}. "
f"Continue from where you left off. Task: {task}"
)
else:
prompt = f"Complete the following task, saving progress after each step: {task}"
try:
result = await Runner.run(agent, prompt, max_turns=20)
return result.final_output
except Exception as exc:
print(f"Task failed, but progress was checkpointed: {exc}")
return (
"The task was interrupted, but your progress has been saved. "
"You can resume from where you left off."
)
This pattern is especially valuable for agents that perform research, data processing, or multi-step workflows where re-doing completed work would be expensive or time-consuming.
Best Practices for Error Recovery
Now that we have covered the major patterns, here are the best practices that tie them together into a cohesive resilience strategy:
1. Classify Errors Before Responding
Not all errors are equal. A rate limit error deserves a retry; a malformed user input does not. Always classify errors into categories like transient, permanent, validation, and system before deciding on a recovery strategy. This prevents wasteful retries on errors that will never succeed.
2. Always Provide User-Facing Fallback Messages
Never let a raw exception or stack trace reach the end user. Every error path should terminate with a clear, human-readable message that sets expectations and, where possible, suggests next steps.
3. Log Everything, Expose Nothing
Log full error details including stack traces, request context, and agent state for debugging. But never expose these details to users. Use the error boundary pattern to separate internal diagnostics from external communication.
4. Set Sensible Retry Limits
Retries are not free. Each retry consumes tokens, time, and potentially rate limit budget. A good default is 3 to 4 retries with exponential backoff. For tool calls, consider whether the tool is idempotent before retrying.
5. Make Tool Errors Instructional
When a tool fails, return error messages that tell the model exactly what went wrong and how to fix it. "Invalid input: expected ISO date format YYYY-MM-DD, got 'tomorrow'" is far more useful than "Error" and will lead to faster self-correction.
6. Monitor Error Rates in Production
# Simple error rate monitor
class ErrorMonitor:
def __init__(self, window_size: int = 100):
self.window_size = window_size
self.events = []
def record(self, success: bool):
self.events.append(success)
if len(self.events) > self.window_size:
self.events.pop(0)
@property
def error_rate(self) -> float:
if not self.events:
return 0.0
failures = sum(1 for s in self.events if not s)
return failures / len(self.events)
@property
def is_healthy(self) -> bool:
return self.error_rate < 0.1 # Less than 10% error rate
monitor = ErrorMonitor()
# Use in your error boundary
async def monitored_run(agent, prompt):
try:
result = await Runner.run(agent, prompt)
monitor.record(success=True)
return result
except Exception:
monitor.record(success=False)
if not monitor.is_healthy:
# Trigger alert
print(f"ALERT: Error rate is {monitor.error_rate:.1%}")
raise
7. Test Your Error Paths
It is easy to test the happy path and assume errors are handled. Deliberately inject failures into your tests: force timeouts, return invalid data from tools, trigger guardrails, and exceed max turns. Only by exercising error paths can you be confident they work.
8. Avoid Silent Failures
A tool that catches all exceptions and returns an empty string is dangerous. The agent will proceed as if nothing happened, potentially producing incorrect results. Always include enough information in error returns for the model to make an informed decision about recovery.
Conclusion
Error recovery is what separates a demo from a production system. The OpenAI Agents SDK gives you the building blocks, but it is up to you to assemble them into a resilient architecture. By combining retry with exponential backoff for transient failures, tool-level self-correction for function errors, fallback agents for model unavailability, circuit breakers for degraded dependencies, guardrail recovery for safety violations, max-turns handling for runaway loops, comprehensive error boundaries for user-facing safety, and checkpointing for long-running tasks, you create an agent system that can withstand the chaos of real-world operation. Start with the error boundary pattern as your foundation, then layer in the more specialized patterns as your use case demands. Remember that every error is an opportunity for your agent to demonstrate resilience, and a well-handled failure can build more user trust than a flawless success.