Orchestration configuration definitions and validation utilities.
This module defines the top-level orchestration settings used to choose the
internal coordination pattern for MADA and, optionally, constrain which
specialist agents participate in that pattern.
OrchestrationConfig
dataclass
Configuration for MADA orchestration behavior.
Attributes:
| Name |
Type |
Description |
mode |
str
|
Internal orchestration pattern to use.
|
participants |
list[str] | None
|
Optional ordered list of specialist agent names to include in the
orchestration pattern. When omitted, all non-PlanningAgent
agents participate.
|
Source code in src/mada/core/config/orchestration.py
| @dataclass
class OrchestrationConfig:
"""
Configuration for MADA orchestration behavior.
Attributes:
mode:
Internal orchestration pattern to use.
participants:
Optional ordered list of specialist agent names to include in the
orchestration pattern. When omitted, all non-`PlanningAgent`
agents participate.
"""
mode: str = DEFAULT_ORCHESTRATION_MODE
participants: list[str] | None = None
def __post_init__(self) -> None:
normalized_mode = expand_env_vars(self.mode or "").strip().lower()
self.mode = normalized_mode or DEFAULT_ORCHESTRATION_MODE
if self.mode not in SUPPORTED_ORCHESTRATION_MODES:
raise ValueError(f"unsupported orchestration mode: {self.mode}")
if self.participants is None:
return
if not isinstance(self.participants, list):
raise ValueError(
"'orchestration.participants' must be a list of agent names"
)
normalized_participants = []
for participant in self.participants:
if not isinstance(participant, str):
raise ValueError(
"'orchestration.participants' must contain only agent names"
)
participant_name = participant.strip()
if not participant_name:
raise ValueError(
"'orchestration.participants' must not contain empty agent names"
)
normalized_participants.append(participant_name)
self.participants = normalized_participants
def validate_participants(self, agent_names: Iterable[str]) -> None:
"""
Validate configured participants against the configured agents.
Args:
agent_names: Names of configured agents.
Raises:
ValueError: If `participants` names do not match configured
specialist agents.
"""
if self.participants is None:
return
if "PlanningAgent" in self.participants:
raise ValueError(
"PlanningAgent cannot be selected as an orchestration participant"
)
available_names = {name for name in agent_names if name != "PlanningAgent"}
unknown_names = [
participant
for participant in self.participants
if participant not in available_names
]
if unknown_names:
missing = ", ".join(unknown_names)
raise ValueError(f"unknown orchestration participants: {missing}")
|
validate_participants(agent_names)
Validate configured participants against the configured agents.
Parameters:
| Name |
Type |
Description |
Default |
agent_names
|
Iterable[str]
|
Names of configured agents.
|
required
|
Raises:
| Type |
Description |
ValueError
|
If participants names do not match configured
specialist agents.
|
Source code in src/mada/core/config/orchestration.py
| def validate_participants(self, agent_names: Iterable[str]) -> None:
"""
Validate configured participants against the configured agents.
Args:
agent_names: Names of configured agents.
Raises:
ValueError: If `participants` names do not match configured
specialist agents.
"""
if self.participants is None:
return
if "PlanningAgent" in self.participants:
raise ValueError(
"PlanningAgent cannot be selected as an orchestration participant"
)
available_names = {name for name in agent_names if name != "PlanningAgent"}
unknown_names = [
participant
for participant in self.participants
if participant not in available_names
]
if unknown_names:
missing = ", ".join(unknown_names)
raise ValueError(f"unknown orchestration participants: {missing}")
|
load_orchestration_config(config_dict)
Load orchestration configuration from a dictionary.
Parameters:
| Name |
Type |
Description |
Default |
config_dict
|
dict[str, Any] | None
|
Serialized orchestration settings, or None.
|
required
|
Returns:
Source code in src/mada/core/config/orchestration.py
| def load_orchestration_config(
config_dict: dict[str, Any] | None,
) -> OrchestrationConfig:
"""
Load orchestration configuration from a dictionary.
Args:
config_dict: Serialized orchestration settings, or `None`.
Returns:
A validated orchestration configuration object.
"""
if config_dict is None:
return OrchestrationConfig()
if not isinstance(config_dict, dict):
raise ValueError("'orchestration' must be an object")
return OrchestrationConfig(**config_dict)
|