Fusion
f33a563003
- Add ToolManifest Pydantic models with validators for ports, mounts, health checks, and traefik config - Implement in-memory ToolRegistry with YAML loading and built-in manifest scanning - Add FastAPI CRUD routes for listing, retrieving, and creating tool manifests - Include built-in manifests for runfusion and code-server - Harden web Dockerfile with unprivileged nginx and port 8080 - Add tool manifest specification documentation and architecture updates Fusion-Task-Id: FN-003
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
"""In-memory tool manifest registry with YAML file loading."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from app.tools.models import ToolManifest
|
|
|
|
|
|
class ToolRegistry:
|
|
"""In-memory registry for tool manifests."""
|
|
|
|
def __init__(self) -> None:
|
|
self._manifests: dict[str, ToolManifest] = {}
|
|
|
|
def load_builtin_manifests(self) -> None:
|
|
"""Scan the built-in manifests directory and register all *.yml files."""
|
|
manifests_dir = Path(__file__).parent / "manifests"
|
|
if not manifests_dir.exists():
|
|
return
|
|
for file_path in sorted(manifests_dir.glob("*.yml")):
|
|
self.load_file(file_path)
|
|
|
|
def load_file(self, path: Path) -> ToolManifest:
|
|
"""Load a single YAML manifest file, validate it, and register it."""
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
manifest = ToolManifest.model_validate(data)
|
|
self.register(manifest)
|
|
return manifest
|
|
|
|
def register(self, manifest: ToolManifest) -> None:
|
|
"""Store a manifest in the registry (idempotent upsert)."""
|
|
self._manifests[manifest.id] = manifest
|
|
|
|
def get(self, tool_id: str) -> ToolManifest | None:
|
|
"""Retrieve a manifest by tool id, or None if not found."""
|
|
return self._manifests.get(tool_id)
|
|
|
|
def list(self) -> list[ToolManifest]:
|
|
"""Return all registered manifests."""
|
|
return list(self._manifests.values())
|
|
|
|
def remove(self, tool_id: str) -> ToolManifest | None:
|
|
"""Remove a manifest by tool id and return it, or None if not found."""
|
|
return self._manifests.pop(tool_id, None)
|
|
|
|
|
|
# Module-level singleton — callers must explicitly bootstrap via
|
|
# registry.load_builtin_manifests() (typically in a FastAPI lifespan).
|
|
registry = ToolRegistry()
|