src.config.settings module

Configuration management for Marcus.

This module provides the Settings class which handles loading, validation, and management of configuration settings from multiple sources including JSON files, environment variables, and default values.

class src.config.settings.Settings

Bases: object

Configuration management system for Marcus.

Manages configuration loading from multiple sources with support for hierarchical settings, environment variable overrides, and validation. Provides specialized getters for different configuration categories.

Parameters:

config_path (Optional[str]) – Path to the configuration file. If None, uses MARCUS_CONFIG environment variable or defaults to ‘config/marcus_config.json’

config_path

Path to the configuration file

Type:

str

defaults

Default configuration values

Type:

Dict[str, Any]

config

Current loaded configuration

Type:

Dict[str, Any]

Examples

>>> settings = Settings()
>>> monitoring_interval = settings.get('monitoring_interval')
>>> risk_config = settings.get_risk_thresholds()
>>> settings.set('custom_setting', 'value')
>>> settings.save()

Notes

Configuration loading priority (highest to lowest): 1. Environment variables 2. Configuration file 3. Default values

__init__(config_path=None)

Parameters:

config_path (str | None)

Return type:

None

get(key, default=None)

Get configuration value using dot notation key.

Supports nested key access using dot notation (e.g., ‘risk_thresholds.high_risk’).

Parameters:

• key (str) – Configuration key, supports dot notation for nested access

• default (Any) – Default value to return if key is not found

Returns:

Configuration value or default if not found

Return type:

Any

Examples

>>> value = settings.get('monitoring_interval')
>>> nested = settings.get('risk_thresholds.high_risk', 0.8)
>>> missing = settings.get('nonexistent.key', 'fallback')

set(key, value)

Set configuration value using dot notation key.

Creates nested dictionary structure as needed when using dot notation.

Parameters:

• key (str) – Configuration key, supports dot notation for nested setting

• value (Any) – Value to set

Return type:

None

Examples

>>> settings.set('monitoring_interval', 600)
>>> settings.set('custom.nested.setting', 'value')

Notes

Setting values does not automatically save to file. Call save() to persist.

save()

Save current configuration to file.

Writes the current configuration to the configured file path, creating directories as needed.

Raises:

• OSError – If file cannot be written or directory cannot be created

• json.JSONEncodeError – If configuration contains non-serializable values

Return type:

None

Examples

>>> settings.set('new_setting', 'value')
>>> settings.save()

Return type:

None

get_team_config(team_name='default')

Get configuration for a specific team.

Retrieves team-specific configuration including skills, work patterns, and communication preferences. Falls back to default team config.

Parameters:

team_name (str) – Name of the team to get configuration for

Returns:

Team configuration containing: - skills: List of team skills - work_patterns: Work scheduling preferences - communication_preferences: Notification settings

Return type:

Dict[str, Any]

Examples

>>> backend_config = settings.get_team_config("backend_team")
>>> default_config = settings.get_team_config()

get_risk_thresholds()

Get risk assessment thresholds for project monitoring.

Returns threshold values used for risk assessment calculations in project monitoring and alerting systems.

Returns:

Risk thresholds containing: - high_risk: Threshold for high risk classification (0.0-1.0) - medium_risk: Threshold for medium risk classification (0.0-1.0) - timeline_buffer: Buffer percentage for timeline calculations

Return type:

Dict[str, float]

Examples

>>> thresholds = settings.get_risk_thresholds()
>>> if project_risk > thresholds['high_risk']:
...     send_alert()

get_escalation_rules()

Get escalation rules for automated issue management.

Returns timing rules that determine when issues should be escalated to higher levels of management.

Returns:

Escalation rules containing: - stuck_task_hours: Hours before escalating stuck tasks - blocker_escalation_hours: Hours before escalating blockers - critical_path_delay_hours: Hours before escalating critical delays

Return type:

Dict[str, int]

Examples

>>> rules = settings.get_escalation_rules()
>>> if task_stuck_hours > rules['stuck_task_hours']:
...     escalate_task(task)

get_communication_rules()

Get communication timing rules for automated messaging.

Returns scheduling rules that determine when different types of automated communications should be sent.

Returns:

Communication timing rules containing: - daily_plan_time: Time to send daily work plans (HH:MM format) - progress_check_time: Time for progress check messages - end_of_day_summary: Time to send daily summaries

Return type:

Dict[str, str]

Examples

>>> rules = settings.get_communication_rules()
>>> daily_time = rules['daily_plan_time']  # "08:00"
>>> schedule_daily_plans(daily_time)

get_ai_settings()

Get AI model configuration settings.

Returns configuration for AI model integration including model selection, parameters, and retry behavior.

Returns:

AI settings containing: - model: Model identifier string - temperature: Sampling temperature (0.0-1.0) - max_tokens: Maximum response tokens - retry_attempts: Number of retry attempts - retry_delay: Delay between retries in seconds

Return type:

Dict[str, Any]

Examples

>>> ai_config = settings.get_ai_settings()
>>> client = AIClient(
...     model=ai_config['model'],
...     temperature=ai_config['temperature']
... )

is_subtasks_enabled()

Check if subtask functionality is enabled.

Returns whether automatic task decomposition and subtask assignment features are enabled.

Returns:

True if subtasks are enabled, False otherwise

Return type:

bool

Examples

>>> if settings.is_subtasks_enabled():
...     decompose_task(task)
>>> # Can also be set via environment:
>>> # export MARCUS_ENABLE_SUBTASKS=false

Notes

Can be controlled via: - Configuration file: features.enable_subtasks - Environment variable: MARCUS_ENABLE_SUBTASKS=true/false

validate()

Validate current configuration for consistency and completeness.

Performs validation checks on the loaded configuration to ensure required settings are present and values are within acceptable ranges. Prints warnings for potential issues.

Returns:

True if configuration is valid, False if critical issues found

Return type:

bool

Notes

Validation checks include: - Presence of required API keys - Reasonable monitoring intervals - At least one communication channel enabled

Warnings are printed to stdout for non-critical issues.

Examples

>>> if settings.validate():
...     start_marcus()
... else:
...     fix_configuration()