Error Handling#

Marcus error framework and recovery strategies.

Auto-generated from source code docstrings.

Error Framework#

Marcus Error Handling Framework.

Comprehensive error handling system designed for autonomous agent environments. Provides structured error types, rich context, and intelligent recovery strategies.

class src.core.error_framework.ErrorSeverity[source]#

Bases: Enum

Error severity levels for prioritization and alerting.

LOW = 'low'#
MEDIUM = 'medium'#
HIGH = 'high'#
CRITICAL = 'critical'#
class src.core.error_framework.ErrorCategory[source]#

Bases: Enum

High-level error categories for monitoring and classification.

TRANSIENT = 'transient'#
CONFIGURATION = 'configuration'#
BUSINESS_LOGIC = 'business_logic'#
INTEGRATION = 'integration'#
SECURITY = 'security'#
SYSTEM = 'system'#
class src.core.error_framework.ErrorContext[source]#

Bases: object

Rich context information for error debugging and recovery.

operation: str = ''#
operation_id: str#
agent_id: str | None = None#
task_id: str | None = None#
agent_state: Dict[str, Any] | None = None#
timestamp: datetime#
correlation_id: str#
system_state: Dict[str, Any] | None = None#
integration_name: str | None = None#
integration_state: Dict[str, Any] | None = None#
user_context: Dict[str, Any] | None = None#
custom_context: Dict[str, Any] | None = None#
__init__(operation='', operation_id=<factory>, agent_id=None, task_id=None, agent_state=None, timestamp=<factory>, correlation_id=<factory>, system_state=None, integration_name=None, integration_state=None, user_context=None, custom_context=None)#
Parameters:
Return type:

None

class src.core.error_framework.RemediationSuggestion[source]#

Bases: object

Actionable remediation suggestions for autonomous agents.

immediate_action: str | None = None#
long_term_solution: str | None = None#
fallback_strategy: str | None = None#
retry_strategy: str | None = None#
escalation_path: str | None = None#
__init__(immediate_action=None, long_term_solution=None, fallback_strategy=None, retry_strategy=None, escalation_path=None)#
Parameters:

  • immediate_action (str | None)

  • long_term_solution (str | None)

  • fallback_strategy (str | None)

  • retry_strategy (str | None)

  • escalation_path (str | None)

Return type:

None

exception src.core.error_framework.MarcusBaseError[source]#

Bases: Exception

Base exception class for Marcus-specific errors.

Provides rich context, categorization, and remediation suggestions for autonomous agent error handling.

__init__(message, error_code=None, context=None, remediation=None, severity=ErrorSeverity.MEDIUM, retryable=False, cause=None)[source]#
Parameters:
to_dict()[source]#

Convert error to dictionary for API responses.

Return type:

Dict[str, Any]

exception src.core.error_framework.TransientError[source]#

Bases: MarcusBaseError

Base class for transient errors that can be automatically retried.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.NetworkTimeoutError[source]#

Bases: TransientError

Network operation timed out.

__init__(service_name='unknown', timeout_seconds=30, *args, **kwargs)[source]#
Parameters:

  • service_name (str)

  • timeout_seconds (int)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.ServiceUnavailableError[source]#

Bases: TransientError

External service is temporarily unavailable.

__init__(service_name='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.RateLimitError[source]#

Bases: TransientError

Rate limit exceeded for API calls.

__init__(service_name='unknown', retry_after=60, *args, **kwargs)[source]#
Parameters:

  • service_name (str)

  • retry_after (int)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.TemporaryResourceError[source]#

Bases: TransientError

Temporary resource unavailability (memory, disk, etc.).

__init__(resource_type='unknown', *args, **kwargs)[source]#
Parameters:

  • resource_type (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.ConfigurationError[source]#

Bases: MarcusBaseError

Base class for configuration-related errors.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.MissingCredentialsError[source]#

Bases: ConfigurationError

Required API credentials are missing.

__init__(service_name='unknown', credential_type='API key', *args, **kwargs)[source]#
Parameters:

  • service_name (str)

  • credential_type (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.InvalidConfigurationError[source]#

Bases: ConfigurationError

Configuration values are invalid.

__init__(config_key='unknown', expected_format='', *args, **kwargs)[source]#
Parameters:

  • config_key (str)

  • expected_format (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.MissingDependencyError[source]#

Bases: ConfigurationError

Required dependency is missing.

__init__(dependency_name='unknown', *args, **kwargs)[source]#
Parameters:

  • dependency_name (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.EnvironmentError[source]#

Bases: ConfigurationError

Environment setup is incorrect.

__init__(environment_issue='unknown', *args, **kwargs)[source]#
Parameters:

  • environment_issue (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.BusinessLogicError[source]#

Bases: MarcusBaseError

Base class for business logic violations.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.TaskAssignmentError[source]#

Bases: BusinessLogicError

Error in task assignment logic.

__init__(task_id='unknown', agent_id='unknown', reason='', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.WorkflowViolationError[source]#

Bases: BusinessLogicError

Workflow rule violation.

__init__(workflow_rule='unknown', *args, **kwargs)[source]#
Parameters:

  • workflow_rule (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.ValidationError[source]#

Bases: BusinessLogicError

Data validation failed.

__init__(field_name='unknown', validation_rule='', *args, **kwargs)[source]#
Parameters:

  • field_name (str)

  • validation_rule (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.StateConflictError[source]#

Bases: BusinessLogicError

System state conflict detected.

__init__(conflict_description='unknown', *args, **kwargs)[source]#
Parameters:

  • conflict_description (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.TaskValidationError[source]#

Bases: BusinessLogicError

Task implementation validation failed.

__init__(task_id='unknown', task_name='', issues=None, *args, **kwargs)[source]#

Initialize task validation error.

Parameters:

  • task_id (str) – Task identifier

  • task_name (str) – Human-readable task name

  • issues (list[str] | None) – List of validation issues found

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.ProjectRootNotFoundError[source]#

Bases: ConfigurationError

Cannot determine project root directory.

__init__(task_id='unknown', *args, **kwargs)[source]#

Initialize project root not found error.

Parameters:

  • task_id (str) – Task identifier

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.IntegrationError[source]#

Bases: MarcusBaseError

Base class for external integration errors.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.KanbanIntegrationError[source]#

Bases: IntegrationError

Kanban board integration error.

__init__(board_name='unknown', operation='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.AIProviderError[source]#

Bases: IntegrationError

AI provider integration error.

__init__(provider_name='unknown', operation='unknown', *args, **kwargs)[source]#
Parameters:

  • provider_name (str)

  • operation (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.AuthenticationError[source]#

Bases: IntegrationError

Authentication failed for external service.

__init__(service_name='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.ExternalServiceError[source]#

Bases: IntegrationError

Generic external service error.

__init__(service_name='unknown', error_details='', *args, **kwargs)[source]#
Parameters:

  • service_name (str)

  • error_details (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.SecurityError[source]#

Bases: MarcusBaseError

Base class for security-related errors.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.AuthorizationError[source]#

Bases: SecurityError

Authorization failed - insufficient permissions.

__init__(resource='unknown', required_permission='unknown', *args, **kwargs)[source]#
Parameters:

  • resource (str)

  • required_permission (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.WorkspaceSecurityError[source]#

Bases: SecurityError

Workspace security violation.

__init__(path='unknown', violation_type='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.PermissionError[source]#

Bases: SecurityError

Permission denied for operation.

__init__(operation='unknown', resource='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.SystemError[source]#

Bases: MarcusBaseError

Base class for critical system errors.

__init__(*args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.ResourceExhaustionError[source]#

Bases: SystemError

System resource exhaustion.

__init__(resource_type='unknown', current_usage='', *args, **kwargs)[source]#
Parameters:

  • resource_type (str)

  • current_usage (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.CorruptedStateError[source]#

Bases: SystemError

System state corruption detected.

__init__(component='unknown', corruption_details='', *args, **kwargs)[source]#
Parameters:

  • component (str)

  • corruption_details (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

exception src.core.error_framework.DatabaseError[source]#

Bases: SystemError

Database operation failed.

__init__(operation='unknown', table='unknown', *args, **kwargs)[source]#
Parameters:
Return type:

None

exception src.core.error_framework.CriticalDependencyError[source]#

Bases: SystemError

Critical system dependency failure.

__init__(dependency_name='unknown', failure_reason='', *args, **kwargs)[source]#
Parameters:

  • dependency_name (str)

  • failure_reason (str)

  • args (Any)

  • kwargs (Any)

Return type:

None

src.core.error_framework.error_context(operation, agent_id=None, task_id=None, integration_name=None, custom_context=None)[source]#

Context manager for automatic error context injection.

Return type:

Any

Parameters:

  • operation (str)

  • agent_id (str | None)

  • task_id (str | None)

  • integration_name (str | None)

  • custom_context (Dict[str, Any] | None)

Usage:
with error_context(“create_task”, agent_id=”agent_123”, task_id=”task_456”):

# operations that might raise Marcus errors pass

class src.core.error_framework.ErrorResponseFormatter[source]#

Bases: object

Formats error responses for different output formats (MCP, JSON, etc.).

static format_for_mcp(error, include_debug=False)[source]#

Format error for MCP protocol response.

Return type:

Dict[str, Any]

Parameters:
static format_for_logging(error)[source]#

Format error for structured logging.

Return type:

Dict[str, Any]

Parameters:

error (MarcusBaseError)

static format_user_friendly(error)[source]#

Format error for user-friendly display.

Return type:

str

Parameters:

error (MarcusBaseError)

src.core.error_framework.MarcusError#

alias of MarcusBaseError

Error Strategies#

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:
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:
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:
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:
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:
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:
async execute_with_fallback(primary_func, *args, cache_key=None, context=None, **kwargs)[source]#

Execute primary function with fallback options.

Return type:

Any

Parameters:
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:
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:
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:
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:
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]

Error Responses#

Marcus Error Response System.

Standardized error response formatting for different contexts: - MCP protocol responses - JSON API responses - User-friendly messages - Logging formats - Monitoring/alerting formats

class src.core.error_responses.ResponseFormat[source]#

Bases: Enum

Supported response formats.

MCP = 'mcp'#
JSON_API = 'json_api'#
USER_FRIENDLY = 'user_friendly'#
LOGGING = 'logging'#
MONITORING = 'monitoring'#
DEBUG = 'debug'#
class src.core.error_responses.ErrorResponseConfig[source]#

Bases: object

Configuration for error response formatting.

include_debug_info: bool = False#
include_stack_trace: bool = False#
include_system_context: bool = False#
include_remediation: bool = True#
max_message_length: int = 500#
sanitize_sensitive_data: bool = True#
custom_fields: Dict[str, Any]#
__init__(include_debug_info=False, include_stack_trace=False, include_system_context=False, include_remediation=True, max_message_length=500, sanitize_sensitive_data=True, custom_fields=<factory>)#
Parameters:

  • include_debug_info (bool)

  • include_stack_trace (bool)

  • include_system_context (bool)

  • include_remediation (bool)

  • max_message_length (int)

  • sanitize_sensitive_data (bool)

  • custom_fields (Dict[str, Any])

Return type:

None

class src.core.error_responses.ErrorResponseFormatter[source]#

Bases: object

Comprehensive error response formatter for Marcus errors.

Provides standardized formatting across different contexts while maintaining security and usability.

__init__(config=None)[source]#
Parameters:

config (ErrorResponseConfig | None)

format_error(error, format_type, additional_context=None)[source]#

Format error for specified response type.

Parameters:
Return type:

Dict[str, Any]

Returns:

Formatted error response dictionary

class src.core.error_responses.BatchErrorResponseFormatter[source]#

Bases: object

Formats responses for batch operations with multiple errors.

Provides summary views and detailed breakdowns of batch operation results.

__init__(formatter=None)[source]#
Parameters:

formatter (ErrorResponseFormatter | None)

format_batch_response(operation_name, errors, successes, total_operations, format_type=ResponseFormat.JSON_API)[source]#

Format response for batch operation with multiple errors.

Return type:

Dict[str, Any]

Parameters:
format_error_summary(errors)[source]#

Create a summary view of multiple errors.

Return type:

Dict[str, Any]

Parameters:

errors (List[MarcusBaseError])

src.core.error_responses.create_success_response(data=None, message='Operation completed successfully', metadata=None)[source]#

Create a standardized success response.

Return type:

Dict[str, Any]

Parameters:
src.core.error_responses.create_error_response(error, format_type=ResponseFormat.MCP, config=None, additional_context=None)[source]#

Create a standardized error response.

Return type:

Dict[str, Any]

Parameters:
src.core.error_responses.handle_mcp_tool_error(error, tool_name, arguments=None)[source]#

Handle errors in MCP tool calls.

Return type:

Dict[str, Any]

Parameters:
On this page
Edit on GitHub
Show Source