Implementing Agent Tool Calling with Fallback Chains: Handling Tool Errors, Timeouts, and Partial Results in Production

Introduction

You've built an AI agent that can call tools — maybe it queries a live pricing API, runs SQL against your data warehouse, or fetches documents from a vector store. In your development environment, everything works beautifully. Then you deploy to production, and within 48 hours you're staring at a wall of errors: the pricing API timed out, the SQL query returned a partial result set, a third-party service returned a 503, and your agent — with no strategy for any of these situations — either crashed outright or hallucinated its way to a confident-sounding wrong answer.

This is the unglamorous reality of production agent systems. Tool calling is the mechanism that gives agents their power, but tools are external systems, and external systems fail. What separates a toy demo from a system you'd actually trust with a business-critical workflow is how gracefully it handles failure. Fallback chains — ordered sequences of alternative strategies that activate when a primary tool fails — are the core pattern you need. But implementing them well requires thinking carefully about error classification, timeout budgets, partial result handling, and how to keep the LLM informed about what happened without derailing the conversation.

By the end of this lesson, you'll have a production-ready fallback framework you can drop into any agent architecture.

What you'll learn:

• How to classify tool errors (transient vs. permanent, partial vs. total) and why classification drives your fallback strategy
• How to implement tiered fallback chains that cascade through alternative tools, degraded modes, and cached results
• How to set and enforce timeout budgets across a chain of tool calls without blocking your agent
• How to handle partial results — when a tool returns some data but not all of it — and surface that uncertainty to the LLM
• How to instrument and monitor fallback behavior in production so you can improve your chains over time

Prerequisites

This lesson assumes you're comfortable with AI agent fundamentals — specifically function/tool calling, the ReAct loop, and agent state management. You should also have some familiarity with building multi-step agents with planning and memory and ideally have already connected at least one external tool to an agent. We'll use Python with OpenAI's API throughout, but the patterns apply to any LLM framework.

---

Understanding Tool Failure Modes Before You Write a Line of Code

The single biggest mistake practitioners make when implementing fallbacks is treating all tool failures identically. They catch Exception, log it, and move to the next option. This is better than nothing, but it's sloppy in ways that bite you in production.

Tool failures fall into distinct categories, and your fallback strategy should depend on which category you're in:

Transient failures are temporary. The API is overloaded right now but will recover. A database connection dropped. A rate limit window just closed. For transient failures, the right first response is often a retry with exponential backoff — not an immediate switch to a fallback.

Permanent failures won't resolve on retry. The API key is invalid. The endpoint doesn't exist. The user doesn't have permissions for this resource. Retrying wastes time and budget; you should fall through to the next strategy immediately.

Partial failures are the trickiest. The tool returned something, but it's incomplete. Your SQL query returned 47 of 50 records before hitting a row limit. Your vector search returned results but with low confidence scores. The API returned data for 8 of 10 requested items. Partial results require a different response than total failures — you may be able to use what you got, supplement with a fallback, or surface the incompleteness to the LLM explicitly.

Timeout failures are a special case of transient failure, but they deserve separate treatment because of the latency implications. If you wait the full 30-second timeout before activating a fallback, your user experience is already destroyed. You need timeout budgets that are much shorter than the maximum theoretical wait time.

Here's the classification logic we'll build on throughout this lesson:

import httpx
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Any, Optional

class FailureMode(Enum):
    TRANSIENT = "transient"      # Retry is worthwhile
    PERMANENT = "permanent"      # Don't retry, fall through immediately
    PARTIAL = "partial"          # Got some data, may be usable
    TIMEOUT = "timeout"          # Time budget exceeded
    RATE_LIMITED = "rate_limited" # Wait before retry or fall through

@dataclass
class ToolResult:
    success: bool
    data: Any
    failure_mode: Optional[FailureMode] = None
    error_message: Optional[str] = None
    completeness_score: float = 1.0  # 1.0 = complete, 0.0 = nothing useful
    latency_ms: float = 0.0
    tool_name: str = ""

def classify_http_error(status_code: int, response_body: str = "") -> FailureMode:
    """Classify an HTTP error into a failure mode."""
    if status_code == 429:
        return FailureMode.RATE_LIMITED
    elif status_code in (500, 502, 503, 504):
        return FailureMode.TRANSIENT
    elif status_code in (400, 401, 403, 404, 422):
        return FailureMode.PERMANENT
    elif status_code == 206:  # HTTP Partial Content
        return FailureMode.PARTIAL
    else:
        return FailureMode.TRANSIENT  # Default to transient for unknown errors

def classify_exception(exc: Exception) -> FailureMode:
    """Classify a Python exception into a failure mode."""
    if isinstance(exc, asyncio.TimeoutError):
        return FailureMode.TIMEOUT
    elif isinstance(exc, httpx.ConnectError):
        return FailureMode.TRANSIENT
    elif isinstance(exc, httpx.TimeoutException):
        return FailureMode.TIMEOUT
    elif isinstance(exc, PermissionError):
        return FailureMode.PERMANENT
    elif isinstance(exc, (ValueError, TypeError)):
        return FailureMode.PERMANENT  # Bad input, won't improve on retry
    else:
        return FailureMode.TRANSIENT

---

Building the Core Fallback Chain Architecture

A fallback chain is an ordered list of tool strategies, where each strategy has its own timeout budget, retry policy, and activation conditions. When the first strategy fails, you evaluate whether to retry it, then move to the next strategy if appropriate, continuing until either one succeeds or you've exhausted all options.

Let's build this as a composable system:

import time
import asyncio
import logging
from typing import Callable, List, Optional
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

@dataclass
class RetryPolicy:
    max_attempts: int = 3
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 30.0
    backoff_multiplier: float = 2.0
    retry_on: List[FailureMode] = field(
        default_factory=lambda: [FailureMode.TRANSIENT, FailureMode.RATE_LIMITED]
    )

@dataclass
class FallbackStrategy:
    name: str
    tool_fn: Callable
    timeout_seconds: float
    retry_policy: RetryPolicy = field(default_factory=RetryPolicy)
    min_completeness_to_accept: float = 0.5  # Accept result if >= 50% complete
    # If True, this strategy can supplement a partial result from a previous one
    can_supplement: bool = False

class FallbackChain:
    def __init__(self, strategies: List[FallbackStrategy], chain_name: str = "unnamed"):
        self.strategies = strategies
        self.chain_name = chain_name
        self._execution_log: List[dict] = []

    async def execute(self, **tool_kwargs) -> ToolResult:
        """
        Execute the fallback chain, trying each strategy in order.
        Returns the best result achieved, even if imperfect.
        """
        best_partial_result: Optional[ToolResult] = None
        
        for strategy in self.strategies:
            result = await self._execute_strategy_with_retries(
                strategy, **tool_kwargs
            )
            
            self._execution_log.append({
                "strategy": strategy.name,
                "success": result.success,
                "completeness": result.completeness_score,
                "failure_mode": result.failure_mode.value if result.failure_mode else None,
                "latency_ms": result.latency_ms,
            })

            if result.success and result.completeness_score >= strategy.min_completeness_to_accept:
                logger.info(
                    f"[{self.chain_name}] Strategy '{strategy.name}' succeeded "
                    f"(completeness={result.completeness_score:.2f})"
                )
                return result

            if result.completeness_score > 0:
                # Store as best partial result seen so far
                if (best_partial_result is None or 
                        result.completeness_score > best_partial_result.completeness_score):
                    best_partial_result = result

            # Permanent failures: skip remaining retries for THIS strategy
            # (already handled in _execute_strategy_with_retries), fall through
            if result.failure_mode == FailureMode.PERMANENT:
                logger.warning(
                    f"[{self.chain_name}] Strategy '{strategy.name}' permanently failed. "
                    f"Moving to next strategy."
                )

        # All strategies exhausted
        if best_partial_result is not None:
            logger.warning(
                f"[{self.chain_name}] All strategies exhausted. Returning best partial result "
                f"(completeness={best_partial_result.completeness_score:.2f})"
            )
            return best_partial_result

        return ToolResult(
            success=False,
            data=None,
            failure_mode=FailureMode.PERMANENT,
            error_message=f"All fallback strategies in chain '{self.chain_name}' failed.",
            completeness_score=0.0,
            tool_name=self.chain_name,
        )

    async def _execute_strategy_with_retries(
        self, strategy: FallbackStrategy, **tool_kwargs
    ) -> ToolResult:
        """Execute a single strategy with its retry policy."""
        policy = strategy.retry_policy
        attempt = 0
        last_result = None

        while attempt < policy.max_attempts:
            attempt += 1
            start_time = time.monotonic()

            try:
                result = await asyncio.wait_for(
                    strategy.tool_fn(**tool_kwargs),
                    timeout=strategy.timeout_seconds,
                )
                result.latency_ms = (time.monotonic() - start_time) * 1000
                result.tool_name = strategy.name

                if result.success:
                    return result

                # Check if this failure mode is retryable
                if result.failure_mode not in policy.retry_on:
                    return result  # Not retryable, return immediately

                last_result = result

            except asyncio.TimeoutError:
                latency_ms = (time.monotonic() - start_time) * 1000
                last_result = ToolResult(
                    success=False,
                    data=None,
                    failure_mode=FailureMode.TIMEOUT,
                    error_message=f"Strategy '{strategy.name}' timed out after {strategy.timeout_seconds}s",
                    completeness_score=0.0,
                    latency_ms=latency_ms,
                    tool_name=strategy.name,
                )
                if FailureMode.TIMEOUT not in policy.retry_on:
                    return last_result

            except Exception as exc:
                latency_ms = (time.monotonic() - start_time) * 1000
                failure_mode = classify_exception(exc)
                last_result = ToolResult(
                    success=False,
                    data=None,
                    failure_mode=failure_mode,
                    error_message=str(exc),
                    completeness_score=0.0,
                    latency_ms=latency_ms,
                    tool_name=strategy.name,
                )
                if failure_mode not in policy.retry_on:
                    return last_result

            # Calculate backoff delay
            if attempt < policy.max_attempts:
                delay = min(
                    policy.base_delay_seconds * (policy.backoff_multiplier ** (attempt - 1)),
                    policy.max_delay_seconds,
                )
                logger.info(
                    f"[{strategy.name}] Attempt {attempt} failed, retrying in {delay:.1f}s"
                )
                await asyncio.sleep(delay)

        return last_result

---

A Real-World Example: Product Pricing Agent

Let's make this concrete. You're building an agent for a retail analytics platform. One of its tools fetches current product pricing. Here's the landscape:

• Primary source: Your internal pricing API (fast, authoritative, sometimes overloaded during peak hours)
• Secondary source: A third-party pricing aggregator API (slightly slower, usually available, charges per call)
• Tertiary source: Your data warehouse (always available, but prices may be up to 4 hours stale)
• Last resort: Return cached prices from the last successful fetch (may be very stale, explicitly flagged)

This is a textbook fallback chain. Let's implement the actual tool functions:

import httpx
import json
from datetime import datetime, timedelta
from typing import Optional
import redis

# Assume these are configured elsewhere
PRICING_API_BASE = "https://pricing.internal.company.com"
AGGREGATOR_API_BASE = "https://api.pricingaggregator.io"
REDIS_CLIENT = redis.Redis(host="localhost", port=6379, db=0)

async def fetch_from_internal_api(product_ids: list[str]) -> ToolResult:
    """Primary tool: Internal pricing API"""
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                f"{PRICING_API_BASE}/v2/prices/batch",
                json={"product_ids": product_ids},
                headers={"Authorization": f"Bearer {INTERNAL_API_KEY}"},
            )
            
            if response.status_code == 200:
                data = response.json()
                prices = data.get("prices", [])
                
                # Check for partial results — API may not return all requested IDs
                completeness = len(prices) / len(product_ids) if product_ids else 1.0
                
                return ToolResult(
                    success=True,
                    data={"prices": prices, "source": "internal_api", "as_of": datetime.utcnow().isoformat()},
                    completeness_score=completeness,
                )
            else:
                failure_mode = classify_http_error(response.status_code)
                return ToolResult(
                    success=False,
                    data=None,
                    failure_mode=failure_mode,
                    error_message=f"Internal API returned {response.status_code}: {response.text[:200]}",
                )
        except Exception as exc:
            return ToolResult(
                success=False,
                data=None,
                failure_mode=classify_exception(exc),
                error_message=str(exc),
            )

async def fetch_from_aggregator(product_ids: list[str]) -> ToolResult:
    """Secondary tool: Third-party pricing aggregator"""
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                f"{AGGREGATOR_API_BASE}/prices",
                params={"ids": ",".join(product_ids), "currency": "USD"},
                headers={"X-API-Key": AGGREGATOR_API_KEY},
            )
            
            if response.status_code == 200:
                data = response.json()
                prices = data.get("results", [])
                completeness = len(prices) / len(product_ids) if product_ids else 1.0
                
                return ToolResult(
                    success=True,
                    data={"prices": prices, "source": "aggregator", "as_of": datetime.utcnow().isoformat()},
                    completeness_score=completeness,
                )
            elif response.status_code == 429:
                return ToolResult(
                    success=False,
                    data=None,
                    failure_mode=FailureMode.RATE_LIMITED,
                    error_message="Aggregator rate limit exceeded",
                )
            else:
                return ToolResult(
                    success=False,
                    data=None,
                    failure_mode=classify_http_error(response.status_code),
                    error_message=f"Aggregator API error: {response.status_code}",
                )
        except Exception as exc:
            return ToolResult(
                success=False,
                data=None,
                failure_mode=classify_exception(exc),
                error_message=str(exc),
            )

async def fetch_from_warehouse(product_ids: list[str]) -> ToolResult:
    """Tertiary tool: Data warehouse (synchronous query wrapped in async)"""
    import asyncio
    
    def _query_warehouse(ids):
        # This would be your actual DB query
        import psycopg2
        conn = psycopg2.connect(WAREHOUSE_DSN)
        cursor = conn.cursor()
        cursor.execute(
            """
            SELECT product_id, current_price, updated_at
            FROM product_pricing_snapshot
            WHERE product_id = ANY(%s)
              AND updated_at > NOW() - INTERVAL '4 hours'
            """,
            (ids,)
        )
        rows = cursor.fetchall()
        conn.close()
        return [
            {"product_id": r[0], "price": float(r[1]), "updated_at": r[2].isoformat()}
            for r in rows
        ]
    
    try:
        # Run blocking DB call in a thread pool
        loop = asyncio.get_event_loop()
        prices = await loop.run_in_executor(None, _query_warehouse, product_ids)
        completeness = len(prices) / len(product_ids) if product_ids else 1.0
        
        return ToolResult(
            success=True,
            data={"prices": prices, "source": "warehouse_snapshot", "as_of": datetime.utcnow().isoformat(), "may_be_stale": True},
            completeness_score=completeness,
        )
    except Exception as exc:
        return ToolResult(
            success=False,
            data=None,
            failure_mode=classify_exception(exc),
            error_message=str(exc),
        )

async def fetch_from_cache(product_ids: list[str]) -> ToolResult:
    """Last resort: Redis cache"""
    prices = []
    missing = []
    
    for pid in product_ids:
        cached = REDIS_CLIENT.get(f"price:{pid}")
        if cached:
            prices.append(json.loads(cached))
        else:
            missing.append(pid)
    
    if not prices:
        return ToolResult(
            success=False,
            data=None,
            failure_mode=FailureMode.PERMANENT,
            error_message="No cached prices available",
        )
    
    completeness = len(prices) / len(product_ids)
    return ToolResult(
        success=True,
        data={"prices": prices, "source": "redis_cache", "missing_ids": missing, "is_cached": True},
        completeness_score=completeness,
    )

Now wire it all together into a chain:

def build_pricing_fallback_chain() -> FallbackChain:
    return FallbackChain(
        chain_name="product_pricing",
        strategies=[
            FallbackStrategy(
                name="internal_pricing_api",
                tool_fn=fetch_from_internal_api,
                timeout_seconds=3.0,  # Aggressive timeout — we expect this to be fast
                retry_policy=RetryPolicy(
                    max_attempts=2,
                    base_delay_seconds=0.5,
                    retry_on=[FailureMode.TRANSIENT],
                ),
                min_completeness_to_accept=0.9,  # Need 90%+ of prices to accept this result
            ),
            FallbackStrategy(
                name="pricing_aggregator",
                tool_fn=fetch_from_aggregator,
                timeout_seconds=8.0,
                retry_policy=RetryPolicy(
                    max_attempts=1,  # Don't retry aggregator — charges per call
                    retry_on=[],
                ),
                min_completeness_to_accept=0.7,
            ),
            FallbackStrategy(
                name="warehouse_snapshot",
                tool_fn=fetch_from_warehouse,
                timeout_seconds=15.0,  # Warehouse queries can be slow
                retry_policy=RetryPolicy(
                    max_attempts=2,
                    retry_on=[FailureMode.TRANSIENT],
                ),
                min_completeness_to_accept=0.5,
            ),
            FallbackStrategy(
                name="redis_cache",
                tool_fn=fetch_from_cache,
                timeout_seconds=1.0,
                retry_policy=RetryPolicy(max_attempts=1, retry_on=[]),
                min_completeness_to_accept=0.0,  # Accept whatever cache has
            ),
        ],
    )

---

Surfacing Uncertainty to the LLM

Here's where most implementations get it wrong: they return a partial or degraded result to the LLM and let the LLM respond as if it got perfect data. The agent then produces a confident answer based on incomplete information. That's a failure of honesty with downstream consequences.

The right approach is to enrich the tool result with context about its reliability, and inject that context into the system prompt or tool response in a way the LLM can use. Let's build that layer:

def format_tool_result_for_llm(result: ToolResult, original_request: dict) -> dict:
    """
    Format a ToolResult into a structured tool response that communicates
    data quality and completeness to the LLM.
    """
    if result.success and result.completeness_score >= 0.95:
        # Clean result, no caveats needed
        return {
            "status": "success",
            "data": result.data,
        }
    
    elif result.success and result.completeness_score > 0:
        # Partial success — communicate what's missing
        missing_count = round((1 - result.completeness_score) * len(original_request.get("product_ids", [])))
        staleness_warning = ""
        
        if result.data and result.data.get("may_be_stale"):
            staleness_warning = " Note: this data comes from a warehouse snapshot and may be up to 4 hours old."
        if result.data and result.data.get("is_cached"):
            staleness_warning = " Note: this data comes from cache and may be significantly out of date."
        
        return {
            "status": "partial_success",
            "data": result.data,
            "data_quality_warning": (
                f"Retrieved prices for {round(result.completeness_score * 100)}% of requested products. "
                f"Approximately {missing_count} product(s) could not be priced.{staleness_warning} "
                f"Inform the user of this limitation in your response."
            ),
        }
    
    else:
        # Total failure
        return {
            "status": "error",
            "data": None,
            "error_details": (
                f"The pricing tool failed completely ({result.error_message}). "
                f"Do not attempt to provide pricing information. "
                f"Tell the user the pricing system is temporarily unavailable and suggest they check back shortly."
            ),
        }

This formatted response gets returned as the tool's result to the LLM. The key insight is that you're not just returning data — you're returning instructions about how to use that data. A well-prompted LLM will follow those instructions and produce appropriately hedged responses.

---

Integrating the Fallback Chain with the OpenAI Tool Calling Interface

Let's wire this into an actual agent that uses OpenAI's tool calling API:

from openai import AsyncOpenAI
import json

client = AsyncOpenAI()

# Define the tool schema for OpenAI
PRICING_TOOL_SCHEMA = {
    "type": "function",
    "function": {
        "name": "get_product_prices",
        "description": "Retrieve current pricing for one or more products by their product IDs.",
        "parameters": {
            "type": "object",
            "properties": {
                "product_ids": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of product IDs to retrieve prices for",
                },
            },
            "required": ["product_ids"],
        },
    },
}

async def execute_tool_call(tool_name: str, tool_args: dict) -> str:
    """Execute a tool call, routing through the appropriate fallback chain."""
    if tool_name == "get_product_prices":
        chain = build_pricing_fallback_chain()
        result = await chain.execute(product_ids=tool_args["product_ids"])
        formatted = format_tool_result_for_llm(result, tool_args)
        return json.dumps(formatted)
    else:
        return json.dumps({"error": f"Unknown tool: {tool_name}"})

async def run_pricing_agent(user_message: str) -> str:
    """Run the pricing agent with full fallback chain support."""
    messages = [
        {
            "role": "system",
            "content": (
                "You are a retail analytics assistant. When pricing data is incomplete or stale, "
                "always communicate this clearly to the user. Never present partial data as complete. "
                "If the pricing system is unavailable, say so directly and suggest alternatives."
            ),
        },
        {"role": "user", "content": user_message},
    ]
    
    while True:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[PRICING_TOOL_SCHEMA],
            tool_choice="auto",
        )
        
        message = response.choices[0].message
        
        if message.tool_calls:
            # Process tool calls
            messages.append(message)
            
            for tool_call in message.tool_calls:
                tool_args = json.loads(tool_call.function.arguments)
                tool_result = await execute_tool_call(
                    tool_call.function.name,
                    tool_args,
                )
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                })
        else:
            # No more tool calls — return the final response
            return message.content

---

Managing the Total Latency Budget

One of the subtler problems with fallback chains is that naively chaining timeouts can produce terrible user experience. If your primary tool times out after 3 seconds, your secondary after 8 seconds, and your tertiary after 15 seconds, and all three fail, you've kept the user waiting 26 seconds before admitting defeat. That's unacceptable.

The solution is a wall-clock budget — a single timer for the entire chain that overrides individual strategy timeouts:

import asyncio
import time

class BudgetedFallbackChain(FallbackChain):
    def __init__(self, strategies: List[FallbackStrategy], chain_name: str = "unnamed", 
                 total_budget_seconds: float = 10.0):
        super().__init__(strategies, chain_name)
        self.total_budget_seconds = total_budget_seconds

    async def execute(self, **tool_kwargs) -> ToolResult:
        """Execute with a hard wall-clock deadline across the entire chain."""
        deadline = time.monotonic() + self.total_budget_seconds
        best_partial: Optional[ToolResult] = None

        for strategy in self.strategies:
            remaining = deadline - time.monotonic()
            if remaining <= 0.2:  # Less than 200ms left — not worth trying
                logger.warning(
                    f"[{self.chain_name}] Budget exhausted before trying '{strategy.name}'"
                )
                break

            # Cap this strategy's timeout at remaining budget
            effective_timeout = min(strategy.timeout_seconds, remaining - 0.1)
            adjusted_strategy = FallbackStrategy(
                name=strategy.name,
                tool_fn=strategy.tool_fn,
                timeout_seconds=effective_timeout,
                retry_policy=strategy.retry_policy,
                min_completeness_to_accept=strategy.min_completeness_to_accept,
            )

            result = await self._execute_strategy_with_retries(
                adjusted_strategy, **tool_kwargs
            )

            self._execution_log.append({
                "strategy": strategy.name,
                "success": result.success,
                "completeness": result.completeness_score,
                "failure_mode": result.failure_mode.value if result.failure_mode else None,
                "latency_ms": result.latency_ms,
                "budget_remaining_ms": (deadline - time.monotonic()) * 1000,
            })

            if result.success and result.completeness_score >= strategy.min_completeness_to_accept:
                return result

            if result.completeness_score > 0:
                if best_partial is None or result.completeness_score > best_partial.completeness_score:
                    best_partial = result

        return best_partial or ToolResult(
            success=False,
            data=None,
            failure_mode=FailureMode.TIMEOUT,
            error_message=f"Chain '{self.chain_name}' exhausted total budget of {self.total_budget_seconds}s",
            completeness_score=0.0,
        )

---

Monitoring and Improving Your Fallback Chains

A fallback chain you can't observe is a fallback chain you can't improve. In production, you need to track:

1. Activation rate by strategy: If your secondary tool is activating 40% of the time, your primary tool has a reliability problem that deserves a fix.
2. Average completeness score by source: If your warehouse fallback consistently returns 60% complete results, investigate why.
3. Chain execution time distribution: The P99 latency matters as much as the mean.
4. Tool error distribution: Are you seeing more transient or permanent failures? Is rate limiting increasing?

Here's a simple Prometheus-compatible instrumentation layer:

from prometheus_client import Counter, Histogram, Gauge

# Metrics
tool_executions_total = Counter(
    "agent_tool_executions_total",
    "Total tool execution attempts",
    ["chain_name", "strategy_name", "outcome"]
)
tool_completeness = Histogram(
    "agent_tool_completeness_score",
    "Distribution of tool result completeness scores",
    ["chain_name", "strategy_name"],
    buckets=[0.0, 0.25, 0.5, 0.75, 0.9, 0.95, 1.0]
)
chain_latency_seconds = Histogram(
    "agent_chain_latency_seconds",
    "Total chain execution time",
    ["chain_name", "final_outcome"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
fallback_activations = Counter(
    "agent_fallback_activations_total",
    "Number of times a non-primary strategy was used",
    ["chain_name", "strategy_name"]
)

class InstrumentedFallbackChain(BudgetedFallbackChain):
    async def execute(self, **tool_kwargs) -> ToolResult:
        start = time.monotonic()
        result = await super().execute(**tool_kwargs)
        elapsed = time.monotonic() - start

        # Record per-strategy metrics from execution log
        for i, log_entry in enumerate(self._execution_log):
            outcome = "success" if log_entry["success"] else log_entry.get("failure_mode", "unknown")
            tool_executions_total.labels(
                chain_name=self.chain_name,
                strategy_name=log_entry["strategy"],
                outcome=outcome,
            ).inc()
            tool_completeness.labels(
                chain_name=self.chain_name,
                strategy_name=log_entry["strategy"],
            ).observe(log_entry["completeness"])
            
            if i > 0 and log_entry["success"]:
                # This is a fallback activation
                fallback_activations.labels(
                    chain_name=self.chain_name,
                    strategy_name=log_entry["strategy"],
                ).inc()

        # Record chain-level metrics
        final_outcome = "success" if result.success else "failure"
        if result.success and result.completeness_score < 0.95:
            final_outcome = "partial_success"
        
        chain_latency_seconds.labels(
            chain_name=self.chain_name,
            final_outcome=final_outcome,
        ).observe(elapsed)

        return result

The patterns here integrate well with production RAG monitoring practices — the same infrastructure that watches your retrieval pipeline should watch your tool call chains.

---

Hands-On Exercise

Build a fallback chain for a news article retrieval tool used by a research agent. The agent needs to retrieve full article text given a URL or article ID. Here's your tool landscape:

1. Primary: A content extraction API (fast, but has aggressive rate limits)
2. Secondary: Your internal article archive (may not have recent articles)
3. Tertiary: The Wayback Machine / Common Crawl API (slow, but has broad coverage)
4. Last resort: Return just the article title and URL with a note that full text is unavailable

Your tasks:

1. Implement the four tool functions following the ToolResult pattern above. For the purposes of the exercise, you can mock the actual HTTP calls but implement the error handling and classification logic for real.

2. Wire them into a BudgetedFallbackChain with a 12-second total budget. Think carefully about timeout allocation — the Wayback Machine is slow, so if it activates, how much budget should it get?

3. Implement format_tool_result_for_llm for this use case. What should the LLM say when it only has the title and URL?

4. Add one metric that isn't in the example above: track the reason that each fallback strategy was skipped (timeout, rate limit, etc.). This is valuable for diagnosing chain failures.

Stretch goal: Implement a simple cache-warming strategy. After a successful retrieval from the Wayback Machine (slow), write the result to the primary cache so that subsequent requests for the same article are fast.

---

Common Mistakes & Troubleshooting

Mistake 1: Timeouts that are too generous for the UX context

Setting a 30-second timeout on your primary tool because that's what the SLA says sounds reasonable. But if the primary tool regularly takes 15 seconds during peak hours, your fallback chain will often incur that 15-second wait before even starting. Profile your tools' actual latency distributions and set timeouts based on the P95, not the theoretical maximum.

Mistake 2: Retrying permanent failures

If your API key is invalid, retrying three times with backoff just means you fail three times more slowly. Always check for 401/403 status codes and classify them as PERMANENT so you fall through immediately.

Mistake 3: Not clearing the execution log between calls

If you're reusing a FallbackChain instance (which you should, for efficiency), make sure to clear self._execution_log at the start of each execute() call. Stale log entries from previous executions will corrupt your metrics.

Mistake 4: Ignoring partial results from early strategies

If your primary tool returns 80% of the requested products and then your secondary returns the other 20%, you should merge those results rather than returning them separately. Implement a result-merging layer for your specific data types.

Mistake 5: Treating the LLM as a black box for error handling

Some practitioners let the LLM decide what to do with an error message. This works sometimes but is fragile. Be explicit in your tool response format about what the LLM should do. "Tell the user pricing is unavailable" is better guidance than just {"error": "503"}.

Mistake 6: Not accounting for cascading failures

If all your data sources depend on the same upstream service (say, a common authentication provider), all your fallback strategies can fail simultaneously and for the same reason. Design your fallback chain so that strategies at different tiers have different dependencies wherever possible.

---

Summary & Next Steps

You now have a complete framework for handling tool failures in production agent systems. Let's recap what we built:

• A failure classification system that distinguishes transient, permanent, partial, and timeout failures — because your response strategy depends on which type you're facing.
• A composable FallbackChain with per-strategy timeouts, retry policies, and completeness thresholds that correctly cascade through alternatives.
• A wall-clock budget system that prevents your chain from burning 30+ seconds on cascading timeouts.
• A result formatting layer that communicates data quality and incompleteness to the LLM in actionable terms.
• An instrumentation layer that gives you the observability you need to improve your chains over time.

This is hard-won production knowledge. Most agent frameworks give you tool calling; none of them give you a robust failure handling strategy. You have to build that yourself, and now you have the foundation.

Where to go next:

• If your agent uses multiple specialized tools that should activate based on query type, read Query Routing in RAG — the routing patterns there apply directly to tool selection.
• For agents that need to recover from retrieval failures specifically, Agentic RAG: Building Self-Correcting Retrieval Pipelines extends these fallback concepts into the retrieval layer.
• If you're building a system where multiple agents collaborate and each has its own tool chains, Multi-Agent Orchestration covers how to manage failures at the orchestration level.
• For the broader question of how your agent decides what information to trust when tools return conflicting or incomplete results, Tool Memory vs. Retrieval Memory in AI Agents explores memory architectures that can help.
• Finally, if you want to add input validation and output filtering around your tool calls — not just error handling — Guardrails for RAG Pipelines covers the production safety layer in depth.