agent-tool-resilience

A Python library for making AI agent tool calls resilient, with smart retries, fallbacks, circuit breakers, and result validation.

Why?

AI agents fail silently when tools break. This library provides production-grade resilience patterns:

• Smart Retries: Exponential backoff with jitter, respects rate limits
• Fallbacks: Graceful degradation when primary tools fail
• Circuit Breakers: Prevent cascade failures by stopping calls to broken services
• Result Validation: Ensure tool outputs meet expected schemas/conditions
• Observability: Full visibility into what happened during tool execution

Installation

pip install agent-tool-resilience

Quick Start

from agent_tool_resilience import ResilientTool, RetryPolicy, CircuitBreaker

# Wrap any tool with resilience
@ResilientTool(
    retry=RetryPolicy(max_attempts=3, backoff="exponential"),
    circuit_breaker=CircuitBreaker(failure_threshold=5, reset_timeout=60),
    fallback=lambda *args, **kwargs: {"error": "service unavailable", "cached": True}
)
def call_weather_api(location: str) -> dict:
    return requests.get(f"https://api.weather.com/{location}").json()

# Use it normally - resilience is automatic
result = call_weather_api("NYC")

Features

Retry Policies

from agent_tool_resilience import RetryPolicy

# Exponential backoff with jitter
policy = RetryPolicy(
    max_attempts=5,
    backoff="exponential",
    base_delay=1.0,
    max_delay=60.0,
    jitter=True,
    retry_on=[TimeoutError, ConnectionError, RateLimitError]
)

Circuit Breakers

Prevent cascade failures by temporarily stopping calls to failing services:

from agent_tool_resilience import CircuitBreaker

breaker = CircuitBreaker(
    failure_threshold=5,      # Open after 5 failures
    success_threshold=2,      # Close after 2 successes
    reset_timeout=60,         # Try again after 60 seconds
    half_open_max_calls=3     # Limited calls in half-open state
)

Fallback Strategies

from agent_tool_resilience import FallbackChain

fallbacks = FallbackChain([
    lambda loc: call_backup_weather_api(loc),
    lambda loc: get_cached_weather(loc),
    lambda loc: {"status": "unavailable", "location": loc}
])

Result Validation

from agent_tool_resilience import ResultValidator

validator = ResultValidator(
    schema={"type": "object", "required": ["temperature", "humidity"]},
    conditions=[
        lambda r: r.get("temperature") is not None,
        lambda r: -100 < r.get("temperature", 0) < 150
    ]
)

Observability

from agent_tool_resilience import ToolExecutionTracer

tracer = ToolExecutionTracer()

@ResilientTool(tracer=tracer)
def my_tool():
    ...

# After execution
print(tracer.get_execution_log())
# [
#   {"tool": "my_tool", "attempt": 1, "status": "failed", "error": "timeout", "duration_ms": 5023},
#   {"tool": "my_tool", "attempt": 2, "status": "success", "duration_ms": 234}
# ]

Rate Limit Awareness

from agent_tool_resilience import RateLimitHandler

handler = RateLimitHandler(
    requests_per_minute=60,
    respect_retry_after=True,
    auto_throttle=True
)

Integration with Agent Frameworks

Works with any Python agent framework:

# LangChain
from langchain.tools import Tool
from agent_tool_resilience import ResilientTool

@ResilientTool(retry=RetryPolicy(max_attempts=3))
def search(query: str) -> str:
    ...

langchain_tool = Tool(name="search", func=search, description="Search the web")

# OpenAI Function Calling
@ResilientTool(circuit_breaker=CircuitBreaker(failure_threshold=3))
def get_stock_price(symbol: str) -> dict:
    ...

License

MIT