# Agent Circuit Breaker

> A lightweight Python circuit breaker for agent and LLM calls — monitors failures and temporarily disables expensive operations when a threshold is exceeded.
Published: 2026-03-15
Tags: circuit-breaker, agents, llm, python, resilience

Source: https://hugo-portfolio-teal.vercel.app/projects/agentic-circuit-breaker/

<!-- synced from GitHub README (scripts/sync-readmes.mjs) — edit the repo README, not this file -->

# Agent Circuit Breaker

Circuit breaker pattern for agent/LLM calls: monitors failures and temporarily disables expensive operations when a threshold is exceeded.

- **Lightweight** — zero dependencies for the core library.
- **Sync and async** — decorator, context manager, and `call()` / `call_async()`.
- **Configurable** — consecutive or sliding-window failure counting, custom predicate, fallback, excluded exceptions.

## Install

```bash
pip install -e .
```

Or add to your project and use the package `agent_circuit_breaker`.

## Quick start

### Decorator

```python
from agent_circuit_breaker import circuit_breaker

@circuit_breaker(failure_threshold=5, recovery_timeout=60)
def call_llm(prompt: str) -> str:
    # your LLM/agent call
    return response

@circuit_breaker(failure_threshold=3, recovery_timeout=30)
async def async_call_llm(prompt: str) -> str:
    return await some_async_client(prompt)
```

### Context manager

```python
from agent_circuit_breaker import CircuitBreaker

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

with breaker:
    result = agent.run(task)

# async
async with breaker:
    result = await agent.run_async(task)
```

### Class-based

```python
from agent_circuit_breaker import CircuitBreaker, CircuitBreakerOpenError

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

try:
    result = breaker.call(agent_function, arg1, arg2)
except CircuitBreakerOpenError:
    result = "Service unavailable"
```

## Configuration

| Parameter | Default | Description |
|-----------|---------|-------------|
| `failure_threshold` | `5` | Number of failures that open the circuit. |
| `recovery_timeout` | `60` | Seconds the circuit stays open before a trial (half-open). |
| `failure_window` | `None` | If set, use a sliding time window (seconds) instead of consecutive failures. |
| `failure_predicate` | `None` | Callable `(exc) -> bool`: only count exception as failure when it returns `True`. |
| `fallback` | `None` | Callable to run when the circuit is open instead of raising. |
| `excluded_exceptions` | `None` | Tuple of exception types that never count as failures (still re-raised). |

Exception handling order: first check `excluded_exceptions`; if the exception is in that tuple, do not count it. Otherwise use `failure_predicate` if set, else treat as failure.

## States

- **CLOSED** — Calls allowed; failures are counted.
- **OPEN** — Calls blocked; `CircuitBreakerOpenError` (or fallback) until `recovery_timeout` has passed.
- **HALF_OPEN** — One trial call allowed; success closes the circuit, failure reopens it.

## Monitoring

Set callbacks on the breaker:

- `on_state_change(old_state, new_state)`
- `on_failure()` / `on_success()`
- `on_open()` / `on_close()` / `on_half_open()`

## Requirements

- Python 3.9+

## License

MIT