nat.middleware.circuit_breaker

Circuit breaker middleware package.

Submodules

• nat.middleware.circuit_breaker.circuit_breaker_middleware
• nat.middleware.circuit_breaker.circuit_breaker_middleware_config
• nat.middleware.circuit_breaker.register

Exceptions

CircuitBreakerOpenError

Raised when a call is short-circuited because the circuit breaker is OPEN.

Classes

CircuitBreakerMiddleware	Middleware implementing the Circuit Breaker pattern with per-target isolation.
CircuitBreakerState	Possible states for the circuit breaker.
CircuitBreakerMiddlewareConfig	Configuration for circuit breaker middleware.

Package Contents

class CircuitBreakerMiddleware(

config: nat.middleware.circuit_breaker.circuit_breaker_middleware_config.CircuitBreakerMiddlewareConfig,

builder: nat.builder.builder.Builder,

)

Bases: nat.middleware.dynamic.dynamic_function_middleware.DynamicFunctionMiddleware

Middleware implementing the Circuit Breaker pattern with per-target isolation.

Monitors execution failures and short-circuits calls when downstream functions fail repeatedly, transitioning through CLOSED, OPEN, and HALF_OPEN states per target.

Initialize CircuitBreakerMiddleware.

Args:

config: Circuit breaker configuration parameters. builder: Workflow builder instance.

_cb_config: nat.middleware.circuit_breaker.circuit_breaker_middleware_config.CircuitBreakerMiddlewareConfig

_states: dict[str, _ToolState]

_lock: asyncio.Lock

_get_tool_state(target: str) → _ToolState

Get or initialize state record for a given target name under lock.

Args:

target: Identifier for the target function or tool.

Returns:

_ToolState: State record for the target.

async get_state(target: str) → CircuitBreakerState

Return the current circuit breaker state for a specific target.

Args:

target: Identifier for the target function or tool.

Returns:

CircuitBreakerState: Current state of the target.

async get_failure_count(target: str) → int

Return the consecutive failure count for a specific target.

Args:

target: Identifier for the target function or tool.

Returns:

int: Number of consecutive failures.

async get_success_count(target: str) → int

Return consecutive successes in HALF_OPEN state towards recovery for a target.

Args:

target: Identifier for the target function or tool.

Returns:

int: Number of consecutive probe successes.

async get_last_state_change(target: str) → float

Return the monotonic timestamp of the last state change for a specific target.

Args:

target: Identifier for the target function or tool.

Returns:

float: Timestamp in seconds.

async is_half_open_probing(target: str) → bool

Return whether a probe execution is actively in-flight for a target in HALF_OPEN state.

Args:

target: Identifier for the target function or tool.

Returns:

bool: True if a probe call is in-flight in HALF_OPEN state.

_get_short_circuit_message(target: str) → str

Format the short-circuit message when OPEN or busy probing.

Args:

target: Identifier for the target function or tool.

Returns:

str: Explanatory error message.

async _before_invocation(target: str) → bool

Evaluate circuit breaker state and determine whether invocation can proceed.

Args:

target: Identifier for the target function or tool.

Returns:

bool: True if execution is admitted as a HALF_OPEN probe, False if normal CLOSED.

Raises:

CircuitBreakerOpenError: If the circuit breaker is OPEN or busy probing in HALF_OPEN.

async _after_invocation_success(

target: str,

*,

is_probe: bool = False,

) → None

Update circuit breaker state upon a successful execution.

Args:

target: Identifier for the target function or tool. is_probe: Whether this execution was an admitted probe call in HALF_OPEN state.

async _after_invocation_failure(

target: str,

*,

is_probe: bool = False,

) → None

Update circuit breaker state upon a failed execution.

Args:

target: Identifier for the target function or tool. is_probe: Whether this execution was an admitted probe call in HALF_OPEN state.

async _after_invocation_cancellation(target: str) → None

Reset probing flag on cancellation without counting as failure.

Args:

target: Identifier for the target function or tool.

async function_middleware_invoke(

*args: Any,

call_next: nat.middleware.middleware.CallNext,

context: nat.middleware.middleware.FunctionMiddlewareContext,

\*\*kwargs: Any,

) → Any

Wrap downstream invocation with circuit breaker monitoring and short-circuiting.

Args:

args: Positional arguments for the intercepted function. call_next: Callable to invoke next middleware or target function. context: Static function metadata describing the tool being invoked. kwargs: Keyword arguments for the intercepted function.

Returns:

Any: The tool result.

Raises:

CircuitBreakerOpenError: If the circuit breaker is OPEN or busy probing. Exception: Any exception raised by the downstream callable.

async function_middleware_stream(

*args: Any,

call_next: nat.middleware.middleware.CallNextStream,

context: nat.middleware.middleware.FunctionMiddlewareContext,

\*\*kwargs: Any,

) → collections.abc.AsyncIterator[Any]

Wrap downstream streaming call with circuit breaker monitoring and short-circuiting.

Args:

args: Positional arguments for the intercepted function. call_next: Callable to invoke next middleware or target stream. context: Static function metadata describing the tool being invoked. kwargs: Keyword arguments for the intercepted function.

Yields:

Any: Stream chunks from downstream execution.

Raises:

CircuitBreakerOpenError: If the circuit breaker is OPEN or busy probing. Exception: Any exception raised by the downstream stream.

exception CircuitBreakerOpenError

Bases: RuntimeError

Raised when a call is short-circuited because the circuit breaker is OPEN.

Initialize self. See help(type(self)) for accurate signature.

class CircuitBreakerState

Bases: enum.StrEnum

Possible states for the circuit breaker.

Initialize self. See help(type(self)) for accurate signature.

CLOSED = 'CLOSED'

OPEN = 'OPEN'

HALF_OPEN = 'HALF_OPEN'

class CircuitBreakerMiddlewareConfig

Bases: nat.middleware.dynamic.dynamic_middleware_config.DynamicMiddlewareConfig

Configuration for circuit breaker middleware.

Attributes:

failure_threshold: Number of consecutive failures required to trip the circuit breaker. cooldown_period: Time in seconds to wait in OPEN state before probing. half_open_success_threshold: Consecutive successful probes required to recover to CLOSED. probe_timeout: Optional timeout in seconds for probe calls in HALF_OPEN state. circuit_breaker_message: Optional custom message returned when short-circuited.

failure_threshold: int = None

cooldown_period: float = None

half_open_success_threshold: int = None

probe_timeout: float | None = None

circuit_breaker_message: str | None = None