102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""Load the user's ``/config/config.py``.
|
|
|
|
The config file is regular Python. We ``exec`` it in a namespace seeded
|
|
with the public surface the user is expected to use (``ClaudeAgent``,
|
|
``RaycastAgent``, ``McpServer``, ``ExposedMcp``, ``Gateway``). The file
|
|
must assign a top-level ``gateway = Gateway(...)``.
|
|
|
|
Element-level validation already happens at construction time — agents
|
|
and ``McpServer`` factories are pydantic models that reject garbage.
|
|
What we *can't* validate at construction is the ``Gateway`` container
|
|
itself (deliberately a plain dataclass per PRD), so we type-check its
|
|
contents here before handing it back.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from beaver_gateway.agents.base import BaseAgent, ExposedMcp
|
|
from beaver_gateway.agents.claude import ClaudeAgent
|
|
from beaver_gateway.agents.raycast import RaycastAgent
|
|
from beaver_gateway.core.conversations import ConversationTexts
|
|
from beaver_gateway.core.registry import Gateway
|
|
from beaver_gateway.frontends.base import Frontend
|
|
from beaver_gateway.mcp.types import HttpMcp, McpServer, PythonToolMcp, StdioMcp
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""User config file is missing, unreadable, or structurally wrong."""
|
|
|
|
|
|
_PUBLIC_NAMES: dict[str, Any] = {
|
|
"ClaudeAgent": ClaudeAgent,
|
|
"RaycastAgent": RaycastAgent,
|
|
"McpServer": McpServer,
|
|
"ExposedMcp": ExposedMcp,
|
|
"Gateway": Gateway,
|
|
"ConversationTexts": ConversationTexts,
|
|
}
|
|
|
|
_McpInstance = StdioMcp | HttpMcp | PythonToolMcp
|
|
|
|
|
|
def load(path: Path) -> Gateway:
|
|
"""Execute ``path`` and return its top-level ``gateway`` object."""
|
|
try:
|
|
source = path.read_text(encoding="utf-8")
|
|
except FileNotFoundError as exc:
|
|
msg = f"config file not found: {path}"
|
|
raise ConfigError(msg) from exc
|
|
except OSError as exc:
|
|
msg = f"could not read config file {path}: {exc}"
|
|
raise ConfigError(msg) from exc
|
|
|
|
code = compile(source, str(path), "exec")
|
|
namespace: dict[str, Any] = {"__file__": str(path), **_PUBLIC_NAMES}
|
|
exec(code, namespace) # noqa: S102 - exec'ing user config is the feature
|
|
|
|
try:
|
|
gw = namespace["gateway"]
|
|
except KeyError as exc:
|
|
msg = f"{path}: no top-level `gateway = Gateway(...)` found"
|
|
raise ConfigError(msg) from exc
|
|
|
|
if not isinstance(gw, Gateway):
|
|
msg = (
|
|
f"{path}: top-level `gateway` must be a Gateway instance, "
|
|
f"got {type(gw).__name__}"
|
|
)
|
|
raise ConfigError(msg)
|
|
|
|
_validate(gw, path)
|
|
return gw
|
|
|
|
|
|
def _validate(gw: Gateway, path: Path) -> None:
|
|
for i, a in enumerate(gw.agents):
|
|
if not isinstance(a, BaseAgent):
|
|
msg = (
|
|
f"{path}: gateway.agents[{i}] must be a ClaudeAgent / "
|
|
f"RaycastAgent instance, got {type(a).__name__}"
|
|
)
|
|
raise ConfigError(msg)
|
|
for i, m in enumerate(gw.mcps):
|
|
if not isinstance(m, _McpInstance):
|
|
msg = (
|
|
f"{path}: gateway.mcps[{i}] must be built via "
|
|
f"McpServer.stdio/.http/.python_tool, "
|
|
f"got {type(m).__name__}"
|
|
)
|
|
raise ConfigError(msg)
|
|
for i, f in enumerate(gw.frontends):
|
|
if not isinstance(f, Frontend):
|
|
msg = (
|
|
f"{path}: gateway.frontends[{i}] must be a Frontend, "
|
|
f"got {type(f).__name__}"
|
|
)
|
|
raise ConfigError(msg)
|