feat(FN-003): add tool manifest registry with FastAPI CRUD and built-in manifests
- 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
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user