Error handling and fallbacks

Function calling introduces a challenge that classic code doesn’t face: when your function fails, there are two recipients for that error. Your application, as usual—and the model, which is waiting for a result to continue. Forgetting this creates the most baffling symptom in production: a conversation that stops dead with no explanation, because an exception was swallowed on the server side and the model is still waiting.

The rule that follows structures this entire lesson: always return a function_call_output, even on failure. A model informed that an API is unavailable can propose an alternative, ask for clarification, or explain the situation to the user. A model that receives nothing can do nothing.

Returning errors to the model

The format of the error matters just as much as sending it. A bare “Error” teaches the model nothing; a JSON object that distinguishes the error type, whether it’s transient or permanent, and the next steps allows the model to decide. This is counterintuitive: you are writing error messages meant to be read by a reasoning machine, not logged for humans.

import json
from openai import OpenAI

client = OpenAI()

def execute_function(name: str, arguments: str, call_id: str) -> dict:
    try:
        args = json.loads(arguments)
        result = dispatcher[name](**args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except KeyError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "unknown_function",
                "message": f"The function '{name}' does not exist"
            })
        }
    except json.JSONDecodeError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "invalid_arguments",
                "message": "The JSON arguments are malformed"
            })
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "internal_error",
                "message": str(e)
            })
        }

Retry with exponential backoff

Retry only makes sense for issues that might resolve on a second attempt: network cuts, 429 status codes, or temporary outages. Retrying a request rejected due to invalid authentication or incorrect parameters only delays failure while multiplying calls. That’s why the code below explicitly filters errors—the filtering does the work, while backoff is just the courtesy you owe the remote service.

import time
import random

def execute_with_retry(func, args, max_retries=3, base_delay=1.0):
    """Retry with exponential backoff and jitter."""
    for attempt in range(max_retries + 1):
        try:
            return func(**args)
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            time.sleep(delay)
        except ValueError:
            # Data error: do not retry
            raise

def execute_function_robust(name, arguments, call_id):
    args = json.loads(arguments)
    try:
        result = execute_with_retry(dispatcher[name], args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": type(e).__name__,
                "message": str(e),
                "retries_performed": 3
            })
        }

Per-function timeouts

A single timeout for all functions is always misconfigured: too short, and it kills heavy processing; too long, and it makes the user wait on a simple cache lookup. Base each value on observed durations under normal conditions, with a margin—not on the worst imaginable case, which would effectively mean having no timeout at all.

import signal
from functools import wraps

TIMEOUTS = {
    "search_client": 5,
    "generate_report": 30,
    "send_email": 10,
}

def with_timeout(func, timeout_seconds):
    """Executes a function with a timeout."""
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(func)
        try:
            return future.result(timeout=timeout_seconds)
        except concurrent.futures.TimeoutError:
            raise TimeoutError(
                f"Function interrupted after {timeout_seconds}s"
            )

def execute_with_timeout(name, arguments, call_id):
    args = json.loads(arguments)
    timeout = TIMEOUTS.get(name, 10)  # 10s by default

    try:
        result = with_timeout(
            lambda: dispatcher[name](**args),
            timeout
        )
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(result)
        }
    except TimeoutError as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "timeout",
                "message": str(e),
                "suggestion": "Retry with more specific criteria"
            })
        }

Fallback between functions

Fallback is a trade-off, not a reflex: serving six-hour-old cached data is better than nothing for a dashboard, and far worse than nothing for a bank balance. When you degrade, say so—in the result returned to the model, so it can warn the user instead of presenting stale data as fresh.

async def search_product(term: str) -> dict:
    """Search with fallback: primary API -> cache -> local database."""
    # Attempt 1: Catalog API
    try:
        return await api_catalog.search(term)
    except Exception:
        pass

    # Attempt 2: Redis cache
    try:
        cached = await redis_client.get(f"product:{term}")
        if cached:
            result = json.loads(cached)
            result["_source"] = "cache"
            result["_warning"] = "Data may be outdated"
            return result
    except Exception:
        pass

    # Attempt 3: Local SQLite database
    try:
        row = db.execute(
            "SELECT * FROM products WHERE name LIKE ?",
            (f"%{term}%",)
        ).fetchone()
        if row:
            return {
                "name": row["name"],
                "price": row["price"],
                "_source": "local_db",
                "_warning": "Data from last synchronization"
            }
    except Exception:
        pass

    return {
        "error": "unavailable",
        "message": "No data source accessible for this search"
    }

Circuit breaker

The circuit breaker handles what retry cannot: a service that is down for the long haul. After N failures, it trips and returns an error immediately, without even trying, then occasionally allows a test request through to check for recovery. The gain is twofold—your users stop waiting for a timeout on every request, and the struggling service isn’t kept down by your retries.

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold=5, open_duration=60):
        self.threshold = failure_threshold
        self.duration = open_duration
        self.failures = 0
        self.last_failure = None
        self.state = "closed"  # closed, open, half-open

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if datetime.now() - self.last_failure > timedelta(seconds=self.duration):
                self.state = "half-open"
                return True
            return False
        return True  # half-open: allow one test attempt

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = datetime.now()
        if self.failures >= self.threshold:
            self.state = "open"

# One circuit breaker per external service
breakers = {
    "weather_api": CircuitBreaker(failure_threshold=3, open_duration=30),
    "catalog_api": CircuitBreaker(failure_threshold=5, open_duration=60),
}

Complete robust loop

The full pattern brings together the five mechanisms above. One precaution makes it truly safe, and it’s easy to miss: cap the number of iterations. A model that loops calling a function whose result it dislikes is rare, but without a ceiling it will burn through your budget until exhaustion—and that’s the kind of incident you discover on the bill.

def robust_loop(prompt, tools, max_turns=10):
    messages = prompt
    for turn in range(max_turns):
        response = client.responses.create(
            model="gpt-5.6-terra",
            input=messages,
            tools=tools
        )

        calls = [i for i in response.output if i.type == "function_call"]
        if not calls:
            return response.output_text

        results = []
        for call in calls:
            result = execute_function_robust(
                call.name, call.arguments, call.call_id
            )
            results.append(result)

        messages = response.output + results

    return "Workflow interrupted: turn limit reached"

Key takeaways

Always return a function_call_output, even on error
Structure errors as JSON so the model can use them
Retry only transient errors with exponential backoff
Implement fallbacks between data sources
Use a circuit breaker to protect failing services