Skip to main content

CODITECT Circuit Breaker Service

Implements the Circuit Breaker pattern to prevent cascading failures when calling agents or external services.

States:

  • CLOSED: Normal operation, requests pass through
  • OPEN: Too many failures, requests fail immediately (fast-fail)
  • HALF_OPEN: Testing recovery, limited requests allowed

Key Features:

  • Configurable failure thresholds and timeouts
  • Per-agent circuit breakers with automatic creation
  • Integration with AgentDiscoveryService for fallback routing
  • Async support for non-blocking operations
  • Metrics tracking for monitoring
  • Optional pybreaker compatibility

Part of Track H.2: Inter-Agent Communication Infrastructure

Usage: from circuit_breaker import CircuitBreaker, AgentCircuitBreaker

# Simple usage
breaker = CircuitBreaker("my-service")
result = await breaker.call(my_async_function, arg1, arg2)

# With decorator
@circuit_breaker("api-service")
async def call_api():
    return await fetch_data()

# Agent circuit breaker with fallback
agent_breaker = AgentCircuitBreaker(discovery_service)
result = await agent_breaker.call_agent("agent-1", task_func, task_data)

Author: CODITECT Framework Version: 1.0.0 Created: January 8, 2026

File: circuit_breaker.py

Classes​

CircuitState​

Circuit breaker states

FailureType​

Types of failures to track

CircuitBreakerError​

Base exception for circuit breaker errors

CircuitOpenError​

Raised when circuit is open and request is rejected

CircuitBreakerConfigError​

Raised for configuration errors

CircuitBreakerConfig​

Configuration for a circuit breaker

FailureRecord​

Record of a single failure

CircuitMetrics​

Metrics for a circuit breaker

CircuitBreaker​

Circuit breaker implementation with state machine.

CircuitBreakerRegistry​

Registry for managing multiple circuit breakers.

Functions​

circuit_breaker(name, config, registry)​

Decorator to wrap a function with circuit breaker.

get_global_registry()​

Get the global circuit breaker registry

create_circuit_breaker(name, fail_max, recovery_timeout)​

Create a circuit breaker with common settings.

from_env(cls, prefix)​

Create config from environment variables

to_dict()​

No description

state()​

Get current circuit state

is_closed()​

No description

is_open()​

No description

is_half_open()​

No description

call_sync(func)​

Call a synchronous function through the circuit breaker.

force_open()​

Force the circuit to open (for testing/maintenance)

force_close()​

Force the circuit to close (for recovery)

reset()​

Reset all state and metrics

get_metrics()​

Get current metrics

get(name, config)​

Get or create a circuit breaker by name

Usage​

python circuit_breaker.py