The Problem
An agent calls your get_flight_prices tool. The airline API is down. Without error handling, the agent crashes and the user sees a 500 error. With good error handling, the agent catches the failure, explains it to the model, and the model responds: "I couldn't retrieve live prices right now — here's what I know from cached data." The agent loop continues. The user experience is preserved.
The difference isn't in the model — it's entirely in how your tool executor handles failures.
Error taxonomy for tools
Not all errors are created equal. The correct response depends on the error type. Build your tool executor around this taxonomy:
| Error Type | Example | Correct Response |
|---|
| Transient | 429 rate limit, 503 temporary outage, network timeout | Retry with exponential backoff |
| Client error | 400 bad input, 401 unauthorized, 404 not found | Return structured error, let model handle — do NOT retry |
| Partial success | Got 3 of 5 records before timeout | Return what you have with a partial_response note |
| Unrecoverable | Implementation bug, data corruption, 500 internal error | Surface to human, do not retry |
The most important distinction: transient errors should be retried; client errors should not. Retrying a 404 wastes time. Retrying a 429 without backoff makes your rate limit problem worse.
Retry patterns
Exponential backoff with jitter is the standard pattern for retrying transient failures:
- Start with a 1s delay after the first failure
- Double the delay on each subsequent failure: 1s → 2s → 4s
- Add random jitter (±20%) to prevent thundering herd — if 100 clients all get rate-limited simultaneously, jitter spreads their retries over time instead of all retrying at the same moment
- Cap at a maximum delay (e.g., 30s)
- Maximum 3 retry attempts for most tools
Attempt 1: immediate
Failure → wait 1.0s (±jitter)
Attempt 2: after 1.1s
Failure → wait 2.0s (±jitter)
Attempt 3: after 1.9s
Failure → wait 4.0s (±jitter)
Attempt 4: after 4.2s
Failure → return structured error
When NOT to retry:
- 400 Bad Request — the input is wrong, retrying won't fix it
- 401 Unauthorized — credentials are invalid, retrying won't fix it
- 404 Not Found — the resource doesn't exist, retrying won't make it appear
- 422 Unprocessable Entity — schema validation failed
Circuit breaker pattern: if a dependency keeps failing, stop sending requests to it. Track the failure rate over a sliding window. When failures exceed a threshold (e.g., 5 failures in 60 seconds), "open" the circuit — immediately return a cached or degraded response without calling the failing API. Reset the circuit after a cool-down period. This prevents slow failures from creating cascading latency across your entire agent.
Your tool call to a weather API fails with a 429 (rate limit) error. What's the correct response?
429s are transient — the API is telling you to slow down. Exponential backoff + jitter spreads retries over time. After 3 retries, return a structured error (not a crash) so the model can explain the situation gracefully. Immediate retries make rate limits worse, not better. Switching to a different API is a valid fallback strategy but only after exhausting retries.
Returning errors to the model
The model can reason about errors IF you return structured information. Don't return "Error: 503" — the model has nothing to work with.
Unhelpful error response:
{"error": "503"}
Helpful structured error response:
{
"status": "error",
"error_type": "transient",
"error_code": "airline_api_unavailable",
"reason": "Airline API is temporarily down (HTTP 503)",
"suggestion": "Try again in 2 minutes or check alternative routes",
"fallback_data": {
"cached_prices": [{"airline": "Delta", "price": 342, "cached_at": "2024-01-15T10:00:00Z"}],
"cache_age_minutes": 45
}
}
With the structured response, the model can tell the user: "I couldn't get live prices right now — the airline API is temporarily down. Based on data from 45 minutes ago, Delta has flights around $342. I'd recommend checking again in a few minutes for current prices."
The model cannot generate that response from "Error: 503".
A tool returns partial data: you got 3 of 5 requested records before the connection timed out. What should you return?
Partial data is almost always better than no data. Return what you have, clearly marked as partial, with context on what's missing. The model can decide: use the 3 records and note the limitation, or tell the user only partial information was retrieved. Never silently truncate or silently fail — the model needs to know the response is incomplete to give an accurate answer.
Build
Build it
The code below implements the complete error handling toolkit: a ToolResult dataclass, a retry_with_backoff decorator, a CircuitBreaker, and a demo agent that handles all error types without crashing.
"""
lesson_02_02_error_handling.py
Resilient tool execution with retry, circuit breaker, and structured errors.
Run:
pip install openai
export OPENAI_API_KEY=...
python lesson_02_02_error_handling.py
"""
import asyncio
import json
import os
import random
import time
from dataclasses import dataclass, field
from enum import Enum
from functools import wraps
from typing import Any, Callable, Optional
from openai import OpenAI
class ErrorType(str, Enum):
SUCCESS = "success"
TRANSIENT = "transient" # Retry: 429, 503, timeout
CLIENT_ERROR = "client_error" # Do not retry: 400, 401, 404
NOT_FOUND = "not_found" # Specific 404
PERMISSION_DENIED = "permission_denied" # 401/403
PARTIAL_SUCCESS = "partial_success" # Got some data, not all
UNRECOVERABLE = "unrecoverable" # Bug, corruption, 500
@dataclass
class ToolResult:
"""Structured result returned from any tool call."""
status: ErrorType
data: Optional[Any] = None
error_code: Optional[str] = None
reason: Optional[str] = None
suggestion: Optional[str] = None
fallback_data: Optional[Any] = None
retrieved: Optional[int] = None
requested: Optional[int] = None
def to_dict(self) -> dict:
result = {"status": self.status.value}
if self.data is not None:
result["data"] = self.data
if self.error_code:
result["error_code"] = self.error_code
if self.reason:
result["reason"] = self.reason
if self.suggestion:
result["suggestion"] = self.suggestion
if self.fallback_data is not None:
result["fallback_data"] = self.fallback_data
if self.retrieved is not None:
result["retrieved"] = self.retrieved
result["requested"] = self.requested
return result
def to_json(self) -> str:
return json.dumps(self.to_dict(), indent=2)
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
jitter: float = 0.2,
retryable_errors: tuple = (ErrorType.TRANSIENT,),
) -> Callable:
"""Decorator: retry a tool function with exponential backoff + jitter."""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> ToolResult:
last_result = None
for attempt in range(max_retries + 1):
result = func(*args, **kwargs)
if result.status == ErrorType.SUCCESS or result.status not in retryable_errors:
return result
last_result = result
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
delay *= 1 + random.uniform(-jitter, jitter)
print(f" [retry] Attempt {attempt + 1} failed ({result.error_code}). Waiting {delay:.1f}s...")
time.sleep(delay)
return last_result
return wrapper
return decorator
class CircuitBreaker:
"""Stop sending requests to a failing dependency after repeated failures."""
def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self._failures = 0
self._opened_at: Optional[float] = None
self._state = "closed" # closed | open | half-open
@property
def state(self) -> str:
if self._state == "open":
if time.time() - self._opened_at > self.reset_timeout:
self._state = "half-open"
return self._state
def record_success(self) -> None:
self._failures = 0
self._state = "closed"
self._opened_at = None
def record_failure(self) -> None:
self._failures += 1
if self._failures >= self.failure_threshold:
self._state = "open"
self._opened_at = time.time()
print(f" [circuit] Circuit opened after {self._failures} failures.")
def call(self, func: Callable, *args, **kwargs) -> ToolResult:
if self.state == "open":
return ToolResult(
status=ErrorType.TRANSIENT,
error_code="circuit_open",
reason=f"Dependency circuit is open — too many recent failures.",
suggestion=f"Circuit will reset in ~{int(self.reset_timeout - (time.time() - self._opened_at))}s.",
)
result = func(*args, **kwargs)
if result.status in (ErrorType.TRANSIENT, ErrorType.UNRECOVERABLE):
self.record_failure()
else:
self.record_success()
return result
# ---------------------------------------------------------------------------
# Simulated flaky tools for demonstration
# ---------------------------------------------------------------------------
CALL_COUNT = {}
def _flaky_api_call(tool_name: str, failure_mode: str) -> ToolResult:
"""Simulate an API that fails in controlled ways."""
CALL_COUNT[tool_name] = CALL_COUNT.get(tool_name, 0) + 1
count = CALL_COUNT[tool_name]
if failure_mode == "transient_then_succeed":
if count <= 2:
return ToolResult(
status=ErrorType.TRANSIENT,
error_code="rate_limit_429",
reason="Rate limit exceeded",
suggestion="Retry after backoff",
)
CALL_COUNT[tool_name] = 0
return ToolResult(status=ErrorType.SUCCESS, data={"price": 342, "airline": "Delta"})
elif failure_mode == "not_found":
return ToolResult(
status=ErrorType.NOT_FOUND,
error_code="order_not_found",
reason="Order ORD-00000000 does not exist",
suggestion="Verify the order ID and try again",
)
elif failure_mode == "partial":
return ToolResult(
status=ErrorType.PARTIAL_SUCCESS,
data=[{"id": 1, "name": "Widget A"}, {"id": 2, "name": "Widget B"}],
retrieved=2,
requested=5,
reason="Connection timeout after 2 records",
)
return ToolResult(status=ErrorType.SUCCESS, data={"result": "ok"})
@retry_with_backoff(max_retries=3, base_delay=0.5) # shorter delay for demo
def get_flight_prices(origin: str, destination: str) -> ToolResult:
return _flaky_api_call("flight_prices", "transient_then_succeed")
def get_order(order_id: str) -> ToolResult:
return _flaky_api_call("get_order", "not_found")
def get_product_batch(product_ids: list) -> ToolResult:
return _flaky_api_call("product_batch", "partial")
# ---------------------------------------------------------------------------
# Demo agent that handles all error types
# ---------------------------------------------------------------------------
DEMO_TOOL_SCHEMAS = [
{
"type": "function",
"function": {
"name": "get_flight_prices",
"description": "Get current flight prices between two airports.",
"parameters": {
"type": "object",
"properties": {
"origin": {"type": "string", "description": "Origin airport code (e.g., JFK)"},
"destination": {"type": "string", "description": "Destination airport code (e.g., LAX)"},
},
"required": ["origin", "destination"],
},
},
},
{
"type": "function",
"function": {
"name": "get_order",
"description": "Get details for a specific order by order ID.",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string", "description": "Order ID to look up"}},
"required": ["order_id"],
},
},
},
]
def run_tool(name: str, args: dict) -> str:
"""Execute a tool and return its result as a JSON string."""
print(f"\n [tool] Calling {name}({args})")
if name == "get_flight_prices":
result = get_flight_prices(**args)
elif name == "get_order":
result = get_order(**args)
else:
result = ToolResult(status=ErrorType.CLIENT_ERROR, error_code="unknown_tool", reason=f"Tool '{name}' not found")
print(f" [tool] Result status: {result.status.value}")
return result.to_json()
def run_demo_agent(user_message: str, client: OpenAI) -> None:
print(f"\n{'='*60}")
print(f"User: {user_message}")
messages = [
{"role": "system", "content": "You are a helpful travel assistant. When a tool returns an error, explain what happened to the user and offer alternatives."},
{"role": "user", "content": user_message},
]
for _ in range(5): # max turns
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=DEMO_TOOL_SCHEMAS,
tool_choice="auto",
)
msg = response.choices[0].message
messages.append(msg.model_dump(exclude_unset=True))
if not msg.tool_calls:
print(f"\nAssistant: {msg.content}")
break
for tc in msg.tool_calls:
tool_result = run_tool(tc.function.name, json.loads(tc.function.arguments))
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": tool_result,
})
if __name__ == "__main__":
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
print("RESILIENT TOOL EXECUTOR DEMO")
print("Demonstrating: retry with backoff, structured errors, model recovery\n")
# Scenario 1: Transient failure → retry → success
print("Scenario 1: Transient failure (rate limit) — retries with backoff")
run_demo_agent("What are the cheapest flights from JFK to LAX?", client)
# Scenario 2: Not-found error — model handles gracefully
print("\nScenario 2: Not-found error — model explains to user")
run_demo_agent("Can you look up order ORD-00000000 for me?", client)
# Scenario 3: Partial success
print("\nScenario 3: Partial success — model uses available data")
partial = get_product_batch(["p1", "p2", "p3", "p4", "p5"])
print(f"\nPartial result:\n{partial.to_json()}")
Try the retry-with-backoff logic in isolation below. It simulates a flaky tool that fails twice before succeeding, and shows how the delay grows between attempts.
import random
import time
def flaky_tool(call_count: list) -> dict:
"""Simulate a tool that fails with a 429 on the first 2 calls."""
call_count[0] += 1
if call_count[0] <= 2:
return {"status": "transient_error", "error_code": "rate_limit_429"}
return {"status": "success", "data": {"price": 342}}
def retry_with_backoff(tool_fn, max_retries=3, base_delay=1.0, jitter=0.2):
call_count = [0]
for attempt in range(max_retries + 1):
result = tool_fn(call_count)
if result["status"] == "success":
print(f"Attempt {attempt + 1}: success -> {result['data']}")
return result
delay = base_delay * (2 ** attempt)
jittered = delay * (1 + random.uniform(-jitter, jitter))
print(f"Attempt {attempt + 1}: {result['error_code']} -> would wait {jittered:.2f}s")
return {"status": "failed"}
result = retry_with_backoff(flaky_tool)
print("Final result:", result)