"""YAML Agent registry loader with validation.""" from __future__ import annotations from pathlib import Path from typing import Literal import yaml from pydantic import BaseModel, field_validator class PersonalityConfig(BaseModel, frozen=True): tone: str = "professional and helpful" greeting: str = "Hello! How can I help you today?" escalation_message: str = "Let me connect you with a human agent." class AgentConfig(BaseModel, frozen=True): name: str description: str permission: Literal["read", "write"] personality: PersonalityConfig = PersonalityConfig() tools: list[str] @field_validator("name") @classmethod def name_not_empty(cls, v: str) -> str: if not v.strip(): raise ValueError("Agent name must not be empty") return v.strip() @field_validator("tools") @classmethod def tools_not_empty(cls, v: list[str]) -> list[str]: if not v: raise ValueError("Agent must have at least one tool") return v class AgentRegistry: """Immutable registry of agent configurations loaded from YAML.""" def __init__(self, agents: tuple[AgentConfig, ...]) -> None: self._agents = {agent.name: agent for agent in agents} @classmethod def load(cls, yaml_path: str | Path) -> AgentRegistry: """Load and validate agent configurations from a YAML file.""" path = Path(yaml_path) if not path.exists(): raise FileNotFoundError(f"Agent config file not found: {path}") raw_text = path.read_text(encoding="utf-8") if not raw_text.strip(): raise ValueError(f"Agent config file is empty: {path}") try: data = yaml.safe_load(raw_text) except yaml.YAMLError as exc: msg = f"Invalid YAML in {path}" if hasattr(exc, "problem_mark") and exc.problem_mark is not None: mark = exc.problem_mark msg += f" at line {mark.line + 1}, column {mark.column + 1}" raise ValueError(msg) from exc if not isinstance(data, dict) or "agents" not in data: raise ValueError(f"Agent config must have a top-level 'agents' key in {path}") raw_agents = data["agents"] if not isinstance(raw_agents, list) or not raw_agents: raise ValueError(f"'agents' must be a non-empty list in {path}") agents: list[AgentConfig] = [] seen_names: set[str] = set() for i, raw in enumerate(raw_agents): if not isinstance(raw, dict): raise ValueError(f"Agent at index {i} must be a mapping in {path}") try: agent = AgentConfig(**raw) except Exception as exc: raise ValueError(f"Invalid agent config at index {i} in {path}: {exc}") from exc if agent.name in seen_names: raise ValueError(f"Duplicate agent name '{agent.name}' in {path}") seen_names.add(agent.name) agents.append(agent) return cls(agents=tuple(agents)) def get_agent(self, name: str) -> AgentConfig: if name not in self._agents: available = ", ".join(sorted(self._agents.keys())) raise KeyError(f"Agent '{name}' not found. Available: {available}") return self._agents[name] def list_agents(self) -> tuple[AgentConfig, ...]: return tuple(self._agents.values()) def get_agents_by_permission(self, permission: str) -> tuple[AgentConfig, ...]: return tuple(a for a in self._agents.values() if a.permission == permission) def __len__(self) -> int: return len(self._agents)