48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
class AgentRegistryError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentDefinition:
|
|
agent_id: str
|
|
label: str
|
|
|
|
|
|
class AgentRegistry:
|
|
def __init__(self, agents: list[AgentDefinition]) -> None:
|
|
self.agents = agents
|
|
self._by_id = {agent.agent_id: agent for agent in agents}
|
|
|
|
def get(self, agent_id: str) -> AgentDefinition:
|
|
try:
|
|
return self._by_id[agent_id]
|
|
except KeyError as exc:
|
|
raise AgentRegistryError(f"unknown agent id: {agent_id}") from exc
|
|
|
|
|
|
def load_agent_registry(path: str | Path) -> AgentRegistry:
|
|
raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or {}
|
|
entries = raw.get("agents")
|
|
if not isinstance(entries, list) or not entries:
|
|
raise AgentRegistryError("agents registry must contain a non-empty agents list")
|
|
|
|
agents: list[AgentDefinition] = []
|
|
seen: set[str] = set()
|
|
for entry in entries:
|
|
agent_id = str(entry.get("id", "")).strip()
|
|
label = str(entry.get("label", "")).strip()
|
|
if not agent_id or not label:
|
|
raise AgentRegistryError("each agent entry requires id and label")
|
|
if agent_id in seen:
|
|
raise AgentRegistryError(f"duplicate agent id: {agent_id}")
|
|
seen.add(agent_id)
|
|
agents.append(AgentDefinition(agent_id=agent_id, label=label))
|
|
return AgentRegistry(agents)
|