Spaces:
Configuration error
Configuration error
| """registry_loader.py - load registry/*.yaml into typed objects. | |
| This is the single entry point used by the orchestrator, codegen, and CLI. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from pathlib import Path | |
| from typing import Any | |
| try: | |
| import yaml | |
| except ImportError as e: # pragma: no cover | |
| raise SystemExit("pip install pyyaml") from e | |
| REGISTRY = Path(__file__).resolve().parent.parent / "registry" | |
| def _load(name: str) -> dict: | |
| path = REGISTRY / name | |
| if not path.exists(): | |
| return {} | |
| with open(path) as f: | |
| return yaml.safe_load(f) or {} | |
| def providers() -> dict[str, dict]: | |
| return _load("providers.yaml").get("providers", {}) | |
| def models() -> dict[str, dict]: | |
| return _load("models.yaml").get("models", {}) | |
| def tools() -> dict[str, dict]: | |
| return _load("tools.yaml").get("tools", {}) | |
| def endpoints() -> dict[str, dict]: | |
| return _load("endpoints.yaml").get("endpoints", {}) | |
| def capabilities() -> dict[str, dict]: | |
| return _load("capabilities.yaml").get("capabilities", {}) | |
| def agents() -> dict[str, dict]: | |
| return _load("agents.yaml").get("agents", {}) | |
| def packages() -> dict[str, dict]: | |
| return _load("packages.yaml").get("packages", {}) | |
| def distributed_models() -> dict[str, dict]: | |
| return _load("distributed-models.yaml").get("models", {}) | |
| def rollout_plans() -> dict[str, dict]: | |
| return _load("rollout-plans.yaml").get("rollouts", {}) | |
| def worker_profiles() -> dict[str, dict]: | |
| return _load("worker-profiles.yaml").get("workers", {}) | |
| def resolve_base(provider_name: str, api_name: str) -> str: | |
| """Resolve an API base URL from env (base_env) or the registry default.""" | |
| prov = providers().get(provider_name, {}) | |
| spec = prov.get("apis", {}).get(api_name, {}) | |
| if not isinstance(spec, dict): | |
| return "" | |
| env_name = spec.get("base_env") | |
| if env_name and os.getenv(env_name): | |
| return os.getenv(env_name) # type: ignore[return-value] | |
| return spec.get("default_base", "") | |
| def enabled(capability: str) -> bool: | |
| """Capability feature-flag check (honors premium flags + env overrides).""" | |
| cfg = capabilities().get(capability) | |
| if not cfg: | |
| return False | |
| env_flag = cfg.get("env_flag") | |
| if env_flag and env_flag in os.environ: | |
| return os.getenv(env_flag, "").lower() in ("1", "true", "yes") | |
| return bool(cfg.get("default_on", False)) | |