AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Add retry and fallback logic to tool calls

Add Retry and Fallback Logic to Tool Calls

In production agent systems, tool calls fail. APIs time out, rate limits hit, services go down. A resilient agent does not crash on the first failure — it retries with backoff, respects timeouts, and falls back to alternative tools when the primary is unavailable.

What you are building

Create a ResilientToolExecutor class that wraps tool execution with:

Exponential backoff retry — retry transient failures (network errors, 429s, 5xx) up to a configurable number of attempts, with exponential delay between retries.
Per-call timeout — if a single tool call exceeds the timeout, cancel it and count it as a failure.
Fallback tools — when all retries on the primary tool are exhausted, try a list of fallback tools in order.
Structured result — return a result object that includes which tool actually succeeded (or that all failed), the number of attempts, total latency, and any error messages.

Why this matters

Agent reliability is not about making individual tools perfect — it is about making the orchestration layer resilient. A web search tool might fail, but a cached search or a different search provider can substitute. This pattern is used in every production agent deployment and is the difference between a demo and a product.

Constraints

Use asyncio for timeout handling.
Backoff formula: min(base_delay * 2^attempt, max_delay) with configurable base and max.
The executor should be reusable across different tools, not hardcoded to one.
Include a is_retryable(error) predicate that callers can customize.

Agents / intermediate / Step 3 of 8

Practice stage

Agent architecture patterns

Hint

Focus on explicit control surfaces — function schemas, state machines, and structured outputs. Agents are most useful when the task requires dynamic tool selection or multi-step reasoning. Start with the simplest pattern (single tool call) before reaching for ReAct loops.

Next drill

Parse structured outputs with error recovery

Success criteria

- Tools return structured, typed responses
- Agent completes the task within a bounded number of steps
- All tool calls include error handling and retries
- Memory/state management prevents unbounded context growth

Review checklist

- Tool schemas validate inputs and handle errors gracefully
- Agent loop has explicit termination conditions
- State is serializable and inspectable between steps
- Cost and token usage are tracked per invocation
- Fallback behavior exists for tool call failures

from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable

class ToolCallError(Exception):
    """Wraps errors from tool execution with metadata."""
    def __init__(self, tool_name: str, message: str, retryable: bool = True):
        super().__init__(message)
        self.tool_name = tool_name
        self.retryable = retryable

@dataclass
class ToolCallResult:
    """Outcome of a resilient tool call."""
    success: bool
    tool_used: str | None = None
    result: Any = None
    attempts: int = 0
    total_latency_ms: float = 0.0
    errors: list[str] = field(default_factory=list)

class ResilientToolExecutor:
    """Wraps tool calls with retry, timeout, and fallback logic."""

def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 0.5,
        max_delay: float = 8.0,
        timeout: float = 10.0,
        is_retryable: Callable[[Exception], bool] | None = None,
    ) -> None:
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.timeout = timeout
        self.is_retryable = is_retryable or self._default_retryable

@staticmethod
    def _default_retryable(error: Exception) -> bool:
        """Default predicate: retry ToolCallErrors marked retryable."""
        # TODO: implement
        raise NotImplementedError

async def _call_with_timeout(
        self, fn: Callable[..., Awaitable[Any]], **kwargs: Any
    ) -> Any:
        """Call an async tool function with a timeout."""
        # TODO: use asyncio.wait_for with self.timeout
        raise NotImplementedError

async def _attempt_tool(
        self, tool_name: str, fn: Callable[..., Awaitable[Any]], **kwargs: Any
    ) -> ToolCallResult:
        """Try calling a single tool with retries and backoff."""
        # TODO: loop up to max_retries
        #   - call _call_with_timeout
        #   - on success, return ToolCallResult
        #   - on retryable failure, sleep with exponential backoff
        #   - on non-retryable failure, break immediately
        raise NotImplementedError

async def execute(
        self,
        primary: tuple[str, Callable[..., Awaitable[Any]]],
        fallbacks: list[tuple[str, Callable[..., Awaitable[Any]]]] | None = None,
        **kwargs: Any,
    ) -> ToolCallResult:
        """Execute primary tool with fallbacks."""
        # TODO: try primary first via _attempt_tool
        # TODO: if primary fails, try each fallback in order
        # TODO: if all fail, return aggregate failure result
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.