Circuit Breakers
Prevent cascading failures with automatic circuit breaker state management.
How It Works
Circuit breakers prevent cascading failures when external tools go down. Instead of repeatedly calling a broken service, the circuit "opens" and fails fast until the service recovers.
┌──────────┐ failures >= threshold ┌──────────┐
│ │ ──────────────────────────────→│ │
│ CLOSED │ │ OPEN │
│ │←────────────────────────────── │ │
└──────────┘ success_threshold met └──────────┘
↑ │
│ timeout expires
│ │
│ ┌───────────┐ │
└──────────────│ HALF_OPEN │←───────────────┘
success └───────────┘
States
- CLOSED — Normal operation. Calls pass through. Failures are counted.
- OPEN — Calls are rejected immediately with
CircuitOpenError. No load on the failing service. - HALF_OPEN — A limited number of test calls are allowed. If they succeed, circuit closes. If they fail, circuit opens again.
Configuration
python
from agentguard import guard
from agentguard.config import CircuitBreakerConfig
@guard(
circuit_breaker=CircuitBreakerConfig(
failure_threshold=5, # Open after 5 failures
success_threshold=2, # Close after 2 successes in half-open
timeout=60.0, # Try half-open after 60 seconds
half_open_max_calls=3, # Allow 3 test calls in half-open
)
)
def external_api(query: str) -> dict:
return requests.get(f"https://api.example.com?q={{query}}").json()Configuration Fields
| Field | Type | Default | Description |
|---|---|---|---|
failure_threshold | int | 5 | Failures before opening the circuit |
success_threshold | int | 2 | Successes in half-open before closing |
timeout | float | 60.0 | Seconds before trying half-open |
half_open_max_calls | int | 3 | Max concurrent calls in half-open state |
excluded_exceptions | tuple | () | Exceptions that don't count as failures |
Monitoring State
python
from agentguard import guard
from agentguard.config import CircuitBreakerConfig
@guard(circuit_breaker=CircuitBreakerConfig(failure_threshold=3))
def my_api(x: str) -> dict:
return requests.get(f"https://api.example.com/{{x}}").json()
# Check circuit state
print(my_api.circuit_breaker.state) # "closed"
print(my_api.circuit_breaker.failure_count) # 0
# Handle circuit open
from agentguard.errors import CircuitOpenError
try:
result = my_api("test")
except CircuitOpenError as e:
print(f"Circuit is open, retry after {{e.retry_after:.0f}}s")
# Use fallback or cached result💡 Shared circuit breakers
If multiple tools depend on the same service, share a circuit breaker instance between them using GuardConfig. When one tool triggers the breaker, all tools using that service will fail fast.