diff --git a/src/toad/agent_schema.py b/src/toad/agent_schema.py index 65e1d19f..659d08a9 100644 --- a/src/toad/agent_schema.py +++ b/src/toad/agent_schema.py @@ -1,4 +1,7 @@ -from typing import TypedDict, Literal, NotRequired +from pathlib import Path +from typing import Any, TypedDict, Literal, NotRequired, cast + +from typeguard import check_type, TypeCheckError type Tag = str """A tag used for categorizing the agent. For example: 'open-source', 'reasoning'.""" @@ -68,3 +71,40 @@ class Agent(TypedDict): """Command to run the agent, by OS or wildcard.""" actions: dict[OS, dict[Action, Command]] """Scripts to perform actions, typically at least to install the agent.""" + + +AGENT_KEYS = frozenset(Agent.__annotations__.keys()) +REQUIRED_KEYS = frozenset(Agent.__required_keys__) + + +class ValidationError(Exception): + """Agent configuration validation failed.""" + + +def validate_agent(agent: dict[str, Any], file_path: Path | None = None) -> "Agent": + """Validate that a dict conforms to the Agent TypedDict schema. + + Args: + agent: The dictionary to validate. + file_path: Optional path to the source file (for error messages). + + Raises: + ValidationError: If validation fails. + + Returns: + The validated Agent. + """ + missing = REQUIRED_KEYS - agent.keys() + if missing: + fields = ", ".join(sorted(missing)) + raise ValidationError(f"missing required field(s): {fields}") + + filtered = {k: v for k, v in agent.items() if k in AGENT_KEYS} + try: + check_type(filtered, Agent) + except TypeCheckError as e: + msg = str(e) + if "is not an instance of" in msg: + raise ValidationError(msg.replace("value of key", "field")) + raise ValidationError(msg) + return cast(Agent, agent) diff --git a/src/toad/agents.py b/src/toad/agents.py index ab5e45e7..2751f319 100644 --- a/src/toad/agents.py +++ b/src/toad/agents.py @@ -1,45 +1,106 @@ +from dataclasses import dataclass, field from importlib.resources import files import asyncio +from pathlib import Path -from toad.agent_schema import Agent +from toad.agent_schema import Agent, ValidationError, validate_agent +from toad.paths import get_config class AgentReadError(Exception): """Problem reading the agents.""" -async def read_agents() -> dict[str, Agent]: - """Read agent information from data/agents +@dataclass +class AgentValidationError: + """Represents a validation error for a custom agent.""" + + file_path: Path + error_message: str + + +@dataclass +class AgentReadResult: + """Result of reading agents, including any validation errors.""" + + agents: dict[str, Agent] = field(default_factory=dict) + validation_errors: list[AgentValidationError] = field(default_factory=list) + + +async def read_agents() -> AgentReadResult: + """Read agent information from data/agents and /agents/ Raises: - AgentReadError: If the files could not be read. + AgentReadError: If the built-in agents could not be read. Returns: - A mapping of identity on to Agent dict. + AgentReadResult containing valid agents and any validation errors from custom agents. """ import tomllib - def read_agents() -> list[Agent]: + def load_agent(file, validate: bool = False) -> Agent | None: + """Load an agent from a TOML file, returning None if inactive. + + Args: + file: The file to load. + validate: If True, validate against the Agent schema. + + Raises: + ValidationError: If validation is enabled and fails. + """ + with file.open("rb") as f: + agent = tomllib.load(f) + if not agent.get("active", True): + return None + if validate: + return validate_agent(agent, file) + return agent # type: ignore[return-value] + + def read_agents() -> AgentReadResult: """Read agent information. - Stored in data/agents + Loads built-in agents from data/agents, then custom agents from + /agents/. Custom agents with the same identity + will override built-in ones. Returns: - List of agent dicts. + AgentReadResult with valid agents and validation errors. """ - agents: list[Agent] = [] + result = AgentReadResult() + + def add_agent(agent: Agent) -> None: + identity = agent.get("identity") + if identity: + result.agents[identity] = agent + try: for file in files("toad.data").joinpath("agents").iterdir(): - agent: Agent = tomllib.load(file.open("rb")) - if agent.get("active", True): - agents.append(agent) - + if agent := load_agent(file): + add_agent(agent) except Exception as error: - raise AgentReadError(f"Failed to read agents; {error}") + raise AgentReadError(f"Failed to read built-in agents; {error}") - return agents + custom_agents_dir = get_config() / "agents" + if custom_agents_dir.exists(): + for file in custom_agents_dir.glob("*.toml"): + try: + if agent := load_agent(file, validate=True): + add_agent(agent) + except ValidationError as error: + result.validation_errors.append( + AgentValidationError( + file_path=file, + error_message=str(error), + ) + ) + except Exception as error: + result.validation_errors.append( + AgentValidationError( + file_path=file, + error_message=f"Failed to load agent: {error}", + ) + ) - agents = await asyncio.to_thread(read_agents) - agent_map = {agent["identity"]: agent for agent in agents} + return result - return agent_map + return await asyncio.to_thread(read_agents) diff --git a/src/toad/cli.py b/src/toad/cli.py index d8de3bce..98f1107a 100644 --- a/src/toad/cli.py +++ b/src/toad/cli.py @@ -24,7 +24,8 @@ async def get_agent_data(launch_agent) -> Agent | None: from toad.agents import read_agents, AgentReadError try: - agents = await read_agents() + result = await read_agents() + agents = result.agents except AgentReadError: agents = {} diff --git a/src/toad/screens/store.py b/src/toad/screens/store.py index 6a0ae650..2e32aae8 100644 --- a/src/toad/screens/store.py +++ b/src/toad/screens/store.py @@ -24,7 +24,7 @@ from toad.widgets.mandelbrot import Mandelbrot from toad.widgets.grid_select import GridSelect from toad.agent_schema import Agent -from toad.agents import read_agents +from toad.agents import read_agents, AgentReadResult QR = """\ @@ -419,7 +419,15 @@ def on_launch_agent(self, message: LaunchAgent) -> None: async def on_mount(self) -> None: self.app.settings_changed_signal.subscribe(self, self.setting_updated) try: - self._agents = await read_agents() + result = await read_agents() + self._agents = result.agents + for error in result.validation_errors: + self.notify( + f"[b]{error.file_path.name}[/b]: {error.error_message}", + title="Custom agent validation error", + severity="warning", + timeout=10, + ) except Exception as error: self.notify( f"Failed to read agents data ({error})",