pybreaker
Circuit breaker pattern implementation for Python — prevents cascading failures by stopping calls to failing services. pybreaker features: CircuitBreaker decorator/wrapper, three states (CLOSED=normal, OPEN=failing, HALF_OPEN=testing), fail_max for failure threshold, reset_timeout for recovery wait, exclude= for non-failure exceptions, state persistence via Redis/SQLite listeners, per-listener state callbacks (before_state_change, after_state_change), threading.Lock for thread safety, CircuitBreakerError on open circuit, MongoDBStorage for distributed state, and async support via AsyncCircuitBreaker.
Score Breakdown
⚙ Agent Friendliness
🔒 Security
Resilience pattern library. No direct security concerns. Circuit breaker state visibility (open/closed) could leak information about backend service health — protect circuit status endpoints. Redis storage uses Redis auth — ensure Redis connection is secured.
⚡ Reliability
Best When
Protecting agent external service calls from cascading failures — circuit breaker pattern stops futile calls to failing services and allows automatic recovery after a cooldown period.
Avoid When
Retry logic (use tenacity), rate limiting (use slowapi), or simple timeout wrapping (use asyncio.wait_for).
Use Cases
- • Agent service call protection — from pybreaker import CircuitBreaker; breaker = CircuitBreaker(fail_max=5, reset_timeout=60); @breaker; def call_external_api(url): return requests.get(url) — circuit breaks after 5 failures; agent external API calls stop immediately when service is down; CircuitBreakerError raised on open circuit instead of waiting for timeout
- • Agent async circuit breaker — from pybreaker import AsyncCircuitBreaker; breaker = AsyncCircuitBreaker(fail_max=3, reset_timeout=30); @breaker; async def call_llm(prompt): return await llm.complete(prompt) — async circuit breaker; agent LLM calls protected from cascading failures; open circuit fails fast without blocking event loop
- • Agent state monitoring — from pybreaker import CircuitBreaker, CircuitBreakerListener; class LogListener(CircuitBreakerListener): def state_change(self, cb, old_state, new_state): logger.warning(f'Circuit {cb.name} changed from {old_state} to {new_state}'); breaker = CircuitBreaker(name='api', listeners=[LogListener()]) — state change callbacks; agent monitors circuit state changes for alerting
- • Agent exception filtering — breaker = CircuitBreaker(fail_max=5, exclude=[HTTPError]); @breaker; def api_call(): response = requests.get(url); response.raise_for_status() — exclude client errors; agent 4xx errors (bad request, auth failure) don't count as circuit failures; only server errors and timeouts trip the breaker
- • Agent Redis state sharing — from pybreaker import CircuitBreaker, RedisStorage; storage = RedisStorage(redis.Redis()); breaker = CircuitBreaker(fail_max=5, state_storage=storage); @breaker; def service_call(): pass — distributed state; multiple agent instances share circuit state via Redis; one agent's failures trip circuit for all instances
Not For
- • Retry logic — circuit breaker stops calls; for automatic retry use tenacity
- • Rate limiting — circuit breaker reacts to failures; for proactive rate limiting use ratelimit or slowapi
- • Async-first framework without pybreaker — for native async resilience use stamina or circuitbreaker library
Interface
Authentication
No auth — resilience pattern library. Redis storage for distributed state uses Redis auth.
Pricing
pybreaker is BSD licensed. Free for all use.
Agent Metadata
Known Gotchas
- ⚠ CircuitBreakerError is not the underlying exception — when circuit is open, pybreaker raises CircuitBreakerError not the original network error; agent code catching specific exceptions (requests.exceptions.Timeout) must also catch CircuitBreakerError; handle both: except (requests.Timeout, CircuitBreakerError): handle_failure()
- ⚠ fail_max counts consecutive failures — 5 failures followed by 1 success resets counter; circuit opens when fail_max failures occur WITHOUT intervening success; agent services with intermittent failures that succeed sometimes may never trip circuit; tune fail_max to your failure pattern
- ⚠ reset_timeout starts from last failure — circuit stays OPEN for reset_timeout seconds after the LAST failure; if failures keep occurring during test period, timer resets; agent code should not hammer OPEN circuit with retries — it extends the recovery time
- ⚠ HALF_OPEN allows exactly one call — after reset_timeout, circuit enters HALF_OPEN and allows one test call; if test succeeds: circuit closes; if test fails: circuit reopens for another reset_timeout; agent parallel requests during HALF_OPEN: one succeeds (test), others raise CircuitBreakerError
- ⚠ Thread safety with threading.Lock — pybreaker uses threading.Lock for thread safety; in asyncio, synchronous lock causes event loop blocking; use AsyncCircuitBreaker for async code; mixing sync CircuitBreaker with async code causes subtle blocking issues
- ⚠ State storage required for multi-process sharing — in-memory state (default) is per-process; each agent process has independent circuit state; agent deployed with multiple workers: use RedisStorage for shared circuit state; without shared storage, each worker independently trips/resets circuit
Alternatives
Full Evaluation Report
Detailed scoring breakdown, competitive positioning, security analysis, and improvement recommendations for pybreaker.
Scores are editorial opinions as of 2026-03-06.