CircuitBreaker

Circuit breaker pattern for preventing cascading failures.

Quick Example

from mamba_agents.errors import CircuitBreaker

breaker = CircuitBreaker(
    name="model-api",
    failure_threshold=5,
    timeout=30.0,
)

# Use as context manager
async with breaker:
    result = await call_api()

# Check state
if breaker.is_open:
    print("Circuit is open")

# Get stats
stats = breaker.get_stats()
print(f"Failures: {stats.failure_count}")

# Manual reset
breaker.reset()

States

1. Closed - Normal, requests pass through
2. Open - Too many failures, requests rejected
3. Half-Open - Testing if service recovered

Configuration

Option	Type	Default	Description
name	str	Required	Unique identifier
failure_threshold	int	5	Failures to open
timeout	float	30.0	Seconds to stay open
success_threshold	int	2	Successes to close

API Reference

CircuitBreaker

CircuitBreaker(
    name: str, config: CircuitBreakerConfig | None = None
)

Circuit breaker for protecting against cascading failures.

The circuit breaker monitors calls to a service and trips (opens) when failures exceed a threshold, preventing further calls until a timeout elapses.

States: - CLOSED: Normal operation, calls pass through - OPEN: Failures exceeded threshold, calls are rejected - HALF_OPEN: Testing recovery, limited calls allowed

Example

breaker = CircuitBreaker("model-api")

async def call_api(): async with breaker: return await api.call()

Or manually:

if breaker.allow_request(): try: result = await api.call() breaker.record_success() except Exception as e: breaker.record_failure(e)

Initialize the circuit breaker.

PARAMETER	DESCRIPTION
name	Name for this circuit breaker. TYPE: str
config	Circuit breaker configuration. TYPE: CircuitBreakerConfig | None DEFAULT: None

Source code in src/mamba_agents/errors/circuit_breaker.py

def __init__(
    self,
    name: str,
    config: CircuitBreakerConfig | None = None,
) -> None:
    """Initialize the circuit breaker.

    Args:
        name: Name for this circuit breaker.
        config: Circuit breaker configuration.
    """
    self.name = name
    self.config = config or CircuitBreakerConfig()
    self._state = CircuitState.CLOSED
    self._failures: deque[FailureRecord] = deque()
    self._last_failure_time: float | None = None
    self._half_open_successes = 0
    self._stats = CircuitStats()

state property

state: CircuitState

Get current circuit state, transitioning if needed.

stats property

stats: CircuitStats

Get circuit breaker statistics.

allow_request

allow_request() -> bool

Check if a request should be allowed.

RETURNS	DESCRIPTION
bool	True if the request should proceed.

Source code in src/mamba_agents/errors/circuit_breaker.py

def allow_request(self) -> bool:
    """Check if a request should be allowed.

    Returns:
        True if the request should proceed.
    """
    current_state = self.state  # This may trigger transition

    # Allow requests in CLOSED or HALF_OPEN states, deny in OPEN
    return current_state != CircuitState.OPEN

record_success

record_success() -> None

Record a successful call.

Source code in src/mamba_agents/errors/circuit_breaker.py

def record_success(self) -> None:
    """Record a successful call."""
    self._stats.total_calls += 1
    self._stats.successful_calls += 1

    if self._state == CircuitState.HALF_OPEN:
        self._half_open_successes += 1
        if self._half_open_successes >= self.config.success_threshold:
            self._transition_to(CircuitState.CLOSED)
            self._failures.clear()

record_failure

record_failure(exception: Exception) -> None

Record a failed call.

PARAMETER	DESCRIPTION
exception	The exception that occurred. TYPE: Exception

Source code in src/mamba_agents/errors/circuit_breaker.py

def record_failure(self, exception: Exception) -> None:
    """Record a failed call.

    Args:
        exception: The exception that occurred.
    """
    self._stats.total_calls += 1
    self._stats.failed_calls += 1

    now = time.time()
    self._failures.append(FailureRecord(timestamp=now, exception=exception))
    self._last_failure_time = now

    if self._state == CircuitState.HALF_OPEN:
        # Immediate trip back to open on failure in half-open
        self._transition_to(CircuitState.OPEN)
    elif (
        self._state == CircuitState.CLOSED
        and self._count_recent_failures() >= self.config.failure_threshold
    ):
        self._transition_to(CircuitState.OPEN)

get_time_until_retry

get_time_until_retry() -> float

Get seconds until retry is allowed.

RETURNS	DESCRIPTION
float	Seconds until retry, 0 if allowed now.

Source code in src/mamba_agents/errors/circuit_breaker.py

def get_time_until_retry(self) -> float:
    """Get seconds until retry is allowed.

    Returns:
        Seconds until retry, 0 if allowed now.
    """
    if self._state != CircuitState.OPEN:
        return 0.0

    if self._last_failure_time is None:
        return 0.0

    elapsed = time.time() - self._last_failure_time
    remaining = self.config.timeout - elapsed
    return max(0.0, remaining)

reset

reset() -> None

Force reset the circuit breaker to closed state.

Source code in src/mamba_agents/errors/circuit_breaker.py

def reset(self) -> None:
    """Force reset the circuit breaker to closed state."""
    self._state = CircuitState.CLOSED
    self._failures.clear()
    self._last_failure_time = None
    self._half_open_successes = 0