agent-tool-resilience 0.1.0
Production-grade resilience for AI agent tool calls: retries, fallbacks, circuit breakers, and validation
License: MIT
Requires Python: >=3.10
Keywords: ai, agents, llm, resilience, retry, circuit-breaker, fallback, tools, langchain, openai

# 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

```bash
pip install agent-tool-resilience
```

## Quick Start

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
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

```python
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:

```python
# 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