src.core.error_strategies module#

Marcus Error Handling Strategies.

Advanced error handling patterns for autonomous agent environments: - Retry policies with exponential backoff - Circuit breaker pattern for external services - Fallback mechanisms for graceful degradation - Error aggregation for batch operations

Status#

AVAILABLE BUT NOT INTEGRATED into the MCP tool layer. These components are tested in isolation but have zero imports from src/marcus_mcp/. The simpler src/core/resilience.py decorators (@with_retry, @with_fallback) are what production code currently uses.

This module is the recommended target for future integration when adding resilience to Kanban API calls, AI provider calls, or other external service boundaries in the MCP layer.

class src.core.error_strategies.CircuitBreakerState[source]#

Bases: Enum

Circuit breaker states.

CLOSED = 'closed'#

OPEN = 'open'#

HALF_OPEN = 'half_open'#

class src.core.error_strategies.RetryPolicy[source]#

Bases: Enum

Retry policy types.

NONE = 'none'#

FIXED_DELAY = 'fixed_delay'#

EXPONENTIAL_BACKOFF = 'exponential_backoff'#

LINEAR_BACKOFF = 'linear_backoff'#

JITTERED_EXPONENTIAL = 'jittered_exponential'#

class src.core.error_strategies.RetryConfig[source]#

Bases: object

Configuration for retry behavior.

max_attempts: int = 3#

base_delay: float = 1.0#

max_delay: float = 60.0#

multiplier: float = 2.0#

jitter: bool = True#

retry_on: tuple[type[Exception], ...] = (<class 'src.core.error_framework.TransientError'>, <class 'src.core.error_framework.IntegrationError'>)#

stop_on: tuple[type[Exception], ...] = ()#

__init__(max_attempts=3, base_delay=1.0, max_delay=60.0, multiplier=2.0, jitter=True, retry_on=(<class 'src.core.error_framework.TransientError'>, <class 'src.core.error_framework.IntegrationError'>), stop_on=())#

Parameters:

max_attempts (int)
base_delay (float)
max_delay (float)
multiplier (float)
jitter (bool)
retry_on (tuple[type[Exception], ...])
stop_on (tuple[type[Exception], ...])

Return type:

None

class src.core.error_strategies.CircuitBreakerConfig[source]#

Bases: object

Configuration for circuit breaker behavior.

failure_threshold: int = 5#

success_threshold: int = 2#

timeout: float = 60.0#

monitor_window: float = 300.0#

max_failures_per_window: int = 10#

__init__(failure_threshold=5, success_threshold=2, timeout=60.0, monitor_window=300.0, max_failures_per_window=10)#

Parameters:

failure_threshold (int)
success_threshold (int)
timeout (float)
monitor_window (float)
max_failures_per_window (int)

Return type:

None

class src.core.error_strategies.CircuitBreakerStatus[source]#

Bases: object

Current status of a circuit breaker.

state: CircuitBreakerState = 'closed'#

failure_count: int = 0#

success_count: int = 0#

last_failure_time: datetime | None = None#

next_attempt_time: datetime | None = None#

failure_history: List[datetime]#

__init__(state=CircuitBreakerState.CLOSED, failure_count=0, success_count=0, last_failure_time=None, next_attempt_time=None, failure_history=<factory>)#

Parameters:

state (CircuitBreakerState)
failure_count (int)
success_count (int)
last_failure_time (datetime | None)
next_attempt_time (datetime | None)
failure_history (List[datetime])

Return type:

None

class src.core.error_strategies.CircuitBreaker[source]#

Bases: object

Circuit breaker implementation for external service calls.

Prevents cascading failures by temporarily blocking calls to failing services.

__init__(name, config=None)[source]#

Parameters:

name (str)
config (CircuitBreakerConfig | None)

property lock: Lock#: Get circuit breaker lock, creating it if needed in the current event loop.

async call(func, *args, **kwargs)[source]#

Execute function with circuit breaker protection.

Return type:

Any

Parameters:

func (Callable[[...], Any])
args (Any)
kwargs (Any)

class src.core.error_strategies.RetryHandler[source]#

Bases: object

Advanced retry handler with multiple backoff strategies.

Supports exponential backoff, jitter, and configurable retry policies.

__init__(config=None)[source]#

Parameters:: config (RetryConfig | None)

async execute(func, *args, context=None, **kwargs)[source]#

Execute function with retry logic.

Return type:

Any

Parameters:

func (Callable[[...], Any])
args (Any)
context (ErrorContext | None)
kwargs (Any)

class src.core.error_strategies.FallbackHandler[source]#

Bases: object

Handles fallback strategies for graceful degradation.

Provides multiple fallback options when primary operations fail.

__init__(name)[source]#

Parameters:: name (str)

fallback_functions: List[Tuple[int, Callable[[...], Any]]]#

cache: Dict[str, Any]#

add_fallback(func, priority=0)[source]#

Add a fallback function with priority (lower = higher priority).

Return type:

None

Parameters:

func (Callable[[...], Any])
priority (int)

async execute_with_fallback(primary_func, *args, cache_key=None, context=None, **kwargs)[source]#

Execute primary function with fallback options.

Return type:

Any

Parameters:

primary_func (Callable[[...], Any])
args (Any)
cache_key (str | None)
context (ErrorContext | None)
kwargs (Any)

class src.core.error_strategies.ErrorAggregator[source]#

Bases: object

Aggregates errors from batch operations.

Collects multiple errors and provides summary reporting.

__init__(operation_name)[source]#

Parameters:: operation_name (str)

errors: List[MarcusBaseError]#

successes: int#

total_operations: int#

add_success()[source]#

Record a successful operation.

Return type:: None

add_error(error, item_context=None)[source]#

Add an error to the aggregation.

Return type:

None

Parameters:

error (Exception)
item_context (Dict[str, Any] | None)

get_summary()[source]#

Get summary of batch operation results.

Return type:: Dict[str, Any]

has_errors()[source]#

Check if any errors were recorded.

Return type:: bool

get_critical_errors()[source]#

Get only critical errors from the batch.

Return type:: List[MarcusBaseError]

raise_if_critical()[source]#

Raise exception if any critical errors were encountered.

Return type:: None

src.core.error_strategies.with_retry(config=None)[source]#

Add retry logic to functions.

Return type:: Callable[[Callable[..., Any]], Callable[..., Any]]
Parameters:: config (RetryConfig | None)

src.core.error_strategies.with_circuit_breaker(name, config=None)[source]#

Add circuit breaker protection to functions.

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

Parameters:

name (str)
config (CircuitBreakerConfig | None)

src.core.error_strategies.with_fallback(*fallback_functions)[source]#

Add fallback functions.

Return type:: Callable[[Callable[..., Any]], Callable[..., Any]]
Parameters:: fallback_functions (Callable[[...], Any])

class src.core.error_strategies.ErrorStrategyRegistry[source]#

Bases: object

Registry for managing error handling strategies across the application.

__init__()[source]#

Return type:: None

circuit_breakers: Dict[str, CircuitBreaker]#

fallback_handlers: Dict[str, FallbackHandler]#

retry_configs: Dict[str, RetryConfig]#

get_circuit_breaker(name, config=None)[source]#

Get or create a circuit breaker for a service.

Return type:

CircuitBreaker

Parameters:

name (str)
config (CircuitBreakerConfig | None)

get_fallback_handler(name)[source]#

Get or create a fallback handler.

Return type:: FallbackHandler
Parameters:: name (str)

register_retry_config(operation, config)[source]#

Register a retry configuration for an operation.

Return type:

None

Parameters:

operation (str)
config (RetryConfig)

get_retry_config(operation)[source]#

Get retry configuration for an operation.

Return type:: RetryConfig
Parameters:: operation (str)

get_health_status()[source]#

Get health status of all circuit breakers.

Return type:: Dict[str, Any]