Spaces:
Configuration error
Configuration error
| """ | |
| Agent-Q3 — Plugin auto-loader. | |
| Plugins are drop-in FastAPI router modules placed under /app/plugins/<name>/. | |
| Each plugin directory must contain a `plugin.py` that exposes either: | |
| - `router: APIRouter` — auto-mounted at /plugins/<name>/... | |
| - `register(app: FastAPI)` — called with the host app for custom wiring | |
| Plugins are discovered at service startup. A `manifest.json` is optional but | |
| recommended; if present its keys are exposed at /plugins/list. | |
| """ | |
| import importlib.util | |
| import json | |
| from collections.abc import Callable | |
| from pathlib import Path | |
| import structlog | |
| from fastapi import APIRouter, FastAPI | |
| log = structlog.get_logger(__name__) | |
| PLUGINS_DIR = Path("/app/plugins") | |
| class Plugin: | |
| __slots__ = ("name", "manifest", "module", "router", "register_fn", "path") | |
| def __init__( | |
| self, name: str, manifest: dict, module, router: APIRouter | None, register_fn: Callable | None, path: str | |
| ): | |
| self.name = name | |
| self.manifest = manifest | |
| self.module = module | |
| self.router = router | |
| self.register_fn = register_fn | |
| self.path = path | |
| def to_dict(self) -> dict: | |
| return { | |
| "name": self.name, | |
| "manifest": self.manifest, | |
| "has_router": self.router is not None, | |
| "has_register_fn": self.register_fn is not None, | |
| "path": self.path, | |
| } | |
| class PluginRegistry: | |
| def __init__(self, plugins_dir: Path = PLUGINS_DIR): | |
| self.plugins_dir = plugins_dir | |
| self._plugins: list[Plugin] = [] | |
| def discover(self) -> int: | |
| self._plugins.clear() | |
| if not self.plugins_dir.exists(): | |
| log.info("plugins directory missing — none loaded", path=str(self.plugins_dir)) | |
| return 0 | |
| for child in sorted(self.plugins_dir.iterdir()): | |
| if not child.is_dir() or child.name.startswith((".", "_")): | |
| continue | |
| plugin_py = child / "plugin.py" | |
| if not plugin_py.exists(): | |
| continue | |
| try: | |
| plug = self._load(child, plugin_py) | |
| if plug: | |
| self._plugins.append(plug) | |
| except Exception as e: | |
| log.warning("plugin load failed", name=child.name, error=str(e)) | |
| log.info("plugins discovered", count=len(self._plugins), names=[p.name for p in self._plugins]) | |
| return len(self._plugins) | |
| def _load(plugin_dir: Path, plugin_py: Path) -> Plugin | None: | |
| manifest = {} | |
| mf = plugin_dir / "manifest.json" | |
| if mf.exists(): | |
| try: | |
| manifest = json.loads(mf.read_text(encoding="utf-8")) | |
| except Exception: | |
| pass | |
| name = manifest.get("name", plugin_dir.name) | |
| spec = importlib.util.spec_from_file_location(f"agent_q3_plugin_{name}", str(plugin_py)) | |
| if not spec or not spec.loader: | |
| return None | |
| module = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(module) | |
| router = getattr(module, "router", None) | |
| register_fn = getattr(module, "register", None) | |
| if not (isinstance(router, APIRouter) or callable(register_fn)): | |
| log.warning("plugin has no router or register()", name=name) | |
| return None | |
| return Plugin(name, manifest, module, router, register_fn, str(plugin_dir)) | |
| def mount_into(self, app: FastAPI) -> int: | |
| mounted = 0 | |
| for p in self._plugins: | |
| try: | |
| if p.router is not None: | |
| app.include_router(p.router, prefix=f"/plugins/{p.name}", tags=[f"plugin:{p.name}"]) | |
| mounted += 1 | |
| if callable(p.register_fn): | |
| p.register_fn(app) | |
| mounted += 1 | |
| except Exception as e: | |
| log.error("plugin mount failed", name=p.name, error=str(e)) | |
| log.info("plugins mounted", count=mounted) | |
| return mounted | |
| def all(self) -> list[Plugin]: | |
| return list(self._plugins) | |
| plugins = PluginRegistry() | |