src.core.error_framework module#

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