Error Handling & Fallbacks

Things will go wrong. Payment APIs time out. Inventory services return errors. External APIs rate limit you. The difference between a good AI agent and a great one is how gracefully it handles failures.

Error Categories

Error Response Structure

Standardize your error responses so AI models can understand and handle them:

from dataclasses import dataclass, asdict
from typing import Optional, List
from enum import Enum

class ErrorCode(Enum):
    # Input errors
    INVALID_INPUT = "INVALID_INPUT"
    MISSING_REQUIRED_FIELD = "MISSING_REQUIRED_FIELD"
    INVALID_FORMAT = "INVALID_FORMAT"

    # Authentication/Authorization
    UNAUTHORIZED = "UNAUTHORIZED"
    FORBIDDEN = "FORBIDDEN"

    # Business logic
    INSUFFICIENT_FUNDS = "INSUFFICIENT_FUNDS"
    INSUFFICIENT_INVENTORY = "INSUFFICIENT_INVENTORY"
    ORDER_NOT_FOUND = "ORDER_NOT_FOUND"
    CUSTOMER_NOT_FOUND = "CUSTOMER_NOT_FOUND"

    # External service
    PAYMENT_DECLINED = "PAYMENT_DECLINED"
    SHIPPING_UNAVAILABLE = "SHIPPING_UNAVAILABLE"
    SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE"

    # Rate limiting
    RATE_LIMITED = "RATE_LIMITED"

    # Unknown
    INTERNAL_ERROR = "INTERNAL_ERROR"

@dataclass
class ToolError:
    code: ErrorCode
    message: str
    details: Optional[dict] = None
    retryable: bool = False
    retry_after_seconds: Optional[int] = None
    suggestions: Optional[List[str]] = None

def error_response(error: ToolError) -> dict:
    """Format error for tool response"""
    response = {
        "success": False,
        "error": {
            "code": error.code.value,
            "message": error.message,
            "retryable": error.retryable
        }
    }

    if error.details:
        response["error"]["details"] = error.details
    if error.retry_after_seconds:
        response["error"]["retry_after_seconds"] = error.retry_after_seconds
    if error.suggestions:
        response["error"]["suggestions"] = error.suggestions

    return response

Implementing Retry Logic

import asyncio
from functools import wraps
import random

def with_retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    exponential_base: float = 2.0,
    retryable_exceptions: tuple = (TimeoutError, ConnectionError)
):
    """Decorator for automatic retry with exponential backoff"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)

                except retryable_exceptions as e:
                    last_exception = e

                    if attempt < max_attempts - 1:
                        # Calculate delay with jitter
                        delay = min(
                            base_delay * (exponential_base ** attempt),
                            max_delay
                        )
                        jitter = random.uniform(0, delay * 0.1)
                        actual_delay = delay + jitter

                        logger.warning(
                            f"Attempt {attempt + 1} failed: {e}. "
                            f"Retrying in {actual_delay:.2f}s"
                        )
                        await asyncio.sleep(actual_delay)

            # All retries exhausted
            raise last_exception

        return wrapper
    return decorator

# Usage
@with_retry(max_attempts=3, retryable_exceptions=(TimeoutError, RateLimitError))
async def call_payment_api(amount: int, customer_id: str):
    return await stripe.PaymentIntent.create(
        amount=amount,
        customer=customer_id
    )

Smart Retry for HTTP Requests

import httpx
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    retry_if_result
)

def is_retryable_status(response: httpx.Response) -> bool:
    """Check if HTTP status indicates a retryable error"""
    return response.status_code in (429, 500, 502, 503, 504)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=(
        retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)) |
        retry_if_result(is_retryable_status)
    )
)
async def fetch_shipping_rates(origin: str, destination: str, weight: float):
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            "https://api.shipping.com/rates",
            json={
                "origin": origin,
                "destination": destination,
                "weight_oz": weight
            }
        )

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            raise RateLimitError(f"Rate limited. Retry after {retry_after}s")

        response.raise_for_status()
        return response.json()

Circuit Breaker Pattern

Prevent cascading failures by stopping calls to failing services:

from datetime import datetime, timedelta
from enum import Enum
from dataclasses import dataclass
import asyncio

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if service recovered

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: int = 30
    half_open_max_calls: int = 3

    def __post_init__(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_recovery():
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError("Service unavailable (circuit open)")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.half_open_max_calls:
                raise CircuitOpenError("Circuit in half-open state, max calls reached")
            self.half_open_calls += 1

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _should_attempt_recovery(self) -> bool:
        if not self.last_failure_time:
            return False
        elapsed = (datetime.now() - self.last_failure_time).total_seconds()
        return elapsed >= self.recovery_timeout

    def _on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
shipping_circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30)

async def get_shipping_rates_with_breaker(destination: str, weight: float):
    try:
        return await shipping_circuit.call(
            fetch_shipping_rates,
            destination=destination,
            weight=weight
        )
    except CircuitOpenError:
        # Return cached/default rates
        return get_fallback_rates(destination, weight)

Fallback Strategies

When primary tools fail, provide alternatives:

async def calculate_shipping_with_fallbacks(
    destination: str,
    weight: float,
    carriers: list = None
):
    """Try multiple shipping providers with fallbacks"""

    # Strategy 1: Try real-time rates
    try:
        rates = await get_live_shipping_rates(destination, weight, carriers)
        return {"source": "live", "rates": rates}
    except (TimeoutError, ServiceUnavailableError) as e:
        logger.warning(f"Live rates failed: {e}")

    # Strategy 2: Use cached rates (less than 1 hour old)
    cached_rates = await get_cached_rates(destination, weight)
    if cached_rates and cached_rates["cached_at"] > datetime.now() - timedelta(hours=1):
        return {
            "source": "cached",
            "rates": cached_rates["rates"],
            "cached_at": cached_rates["cached_at"].isoformat(),
            "warning": "Using cached rates - prices may have changed"
        }

    # Strategy 3: Use estimated rates based on zone/weight
    estimated = calculate_estimated_rates(destination, weight)
    return {
        "source": "estimated",
        "rates": estimated,
        "warning": "Using estimated rates - actual shipping cost calculated at checkout"
    }

Inventory Fallback

async def check_inventory_with_fallback(product_id: str, warehouse_id: str = None):
    """Check inventory with graceful degradation"""

    try:
        # Try real-time inventory
        inventory = await inventory_service.get_stock(product_id, warehouse_id)
        return {
            "source": "real_time",
            "product_id": product_id,
            "available": inventory["available"],
            "status": "in_stock" if inventory["available"] > 0 else "out_of_stock"
        }

    except InventoryServiceError as e:
        logger.error(f"Inventory service failed: {e}")

        # Fallback to last known state
        cached = await cache.get(f"inventory:{product_id}")
        if cached:
            age_seconds = (datetime.now() - cached["timestamp"]).total_seconds()
            return {
                "source": "cached",
                "product_id": product_id,
                "available": cached["available"],
                "status": "likely_available" if cached["available"] > 5 else "check_availability",
                "cache_age_seconds": int(age_seconds),
                "warning": "Real-time inventory unavailable - showing last known state"
            }

        # Ultimate fallback - assume available but verify at checkout
        return {
            "source": "unknown",
            "product_id": product_id,
            "status": "verify_at_checkout",
            "warning": "Cannot confirm availability - will verify when you checkout"
        }

Error Handling in Tool Calls

from mcp import types
import json
import traceback

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Central error handling for all tool calls"""

    try:
        # Input validation
        validation_error = validate_tool_input(name, arguments)
        if validation_error:
            return format_tool_error(validation_error)

        # Route to handler
        result = await route_tool_call(name, arguments)
        return [types.TextContent(type="text", text=json.dumps(result))]

    except ValidationError as e:
        return format_tool_error(ToolError(
            code=ErrorCode.INVALID_INPUT,
            message=str(e),
            retryable=False
        ))

    except RateLimitError as e:
        return format_tool_error(ToolError(
            code=ErrorCode.RATE_LIMITED,
            message="Too many requests. Please wait before trying again.",
            retryable=True,
            retry_after_seconds=e.retry_after
        ))

    except InsufficientFundsError as e:
        return format_tool_error(ToolError(
            code=ErrorCode.INSUFFICIENT_FUNDS,
            message="Payment method declined. Please try a different payment method.",
            retryable=False,
            suggestions=["Use a different card", "Check card balance", "Contact your bank"]
        ))

    except ExternalServiceError as e:
        logger.error(f"External service error: {e}", exc_info=True)
        return format_tool_error(ToolError(
            code=ErrorCode.SERVICE_UNAVAILABLE,
            message="An external service is temporarily unavailable.",
            retryable=True,
            retry_after_seconds=30
        ))

    except Exception as e:
        # Log unexpected errors with full traceback
        logger.error(f"Unexpected error in {name}: {e}\n{traceback.format_exc()}")
        return format_tool_error(ToolError(
            code=ErrorCode.INTERNAL_ERROR,
            message="An unexpected error occurred. Please try again.",
            retryable=True
        ))

def format_tool_error(error: ToolError) -> list[types.TextContent]:
    """Format error response for tool result"""
    return [types.TextContent(
        type="text",
        text=json.dumps(error_response(error))
    )]

Real-World Error Scenarios

Scenario 1: Payment Timeout

Customer: "Please charge my card for order #12345"

AI Agent: [Calls process_payment]
  process_payment(amount=9999, customer_id="cus_abc", idempotency_key="ord_12345_pay")

Tool Response:
{
  "success": false,
  "error": {
    "code": "SERVICE_UNAVAILABLE",
    "message": "Payment processing timed out",
    "retryable": true,
    "retry_after_seconds": 5
  }
}

AI Agent: [Waits 5 seconds, retries with SAME idempotency key]
  process_payment(amount=9999, customer_id="cus_abc", idempotency_key="ord_12345_pay")

Tool Response:
{
  "success": true,
  "payment_id": "pi_xyz789",
  "status": "succeeded"
}

AI Agent: "Your payment of $99.99 has been processed successfully!
Order #12345 is confirmed and will ship within 2 business days."

Scenario 2: Inventory Service Down

Customer: "Is the blue jacket in stock?"

AI Agent: [Calls check_inventory]
  check_inventory(product_id="JACKET-BLUE-M")

Tool Response:
{
  "success": true,
  "source": "cached",
  "product_id": "JACKET-BLUE-M",
  "available": 23,
  "status": "likely_available",
  "cache_age_seconds": 180,
  "warning": "Real-time inventory unavailable - showing last known state"
}

AI Agent: "The blue jacket appears to be in stock - we showed 23 units
about 3 minutes ago. I'll confirm the exact availability when you
proceed to checkout. Would you like to add it to your cart?"

Key Takeaways

Next up: MCP Assessment Quiz - Test your knowledge of everything you've learned about MCP and tool use for e-commerce.