Agent Circuit Breaker

TL;DR
#

What this is: A zero-dependency Python library that wraps agent/LLM calls with circuit breaker semantics — decorator, context manager, or explicit call() / call_async().

What this isn’t: An LLM client or agent framework. It only guards calls you already make.

Install: pip install -e . (from the repo) or add agent_circuit_breaker to your project.

Features
#

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
Observable — callbacks for state changes, failures, successes, and open/close/half-open transitions

Quick Start
#

Decorator
#

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
#

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
#

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+

TL;DR#

Features#

Quick Start#

Decorator#

Context manager#

Class-based#

Configuration#

States#

Monitoring#

Requirements#