| from __future__ import annotations |
|
|
| from .context_store import ContextStore |
| from .hermes_adapter import HermesAdapter, HermesRunError |
| from .profiles import ProfileDiscovery |
|
|
|
|
| class BackendService: |
| def __init__( |
| self, |
| profiles: ProfileDiscovery, |
| store: ContextStore, |
| hermes: HermesAdapter, |
| ) -> None: |
| self.profiles = profiles |
| self.store = store |
| self.hermes = hermes |
|
|
| def list_agents(self) -> list[dict[str, str]]: |
| return [profile.__dict__ for profile in self.profiles.list_agents()] |
|
|
| def get_context(self) -> dict: |
| context = self.store.get_context() |
| return {"context": context.context, "updated_at": context.updated_at} |
|
|
| def set_context(self, context: str) -> dict: |
| updated = self.store.set_context(context) |
| return {"context": updated.context, "updated_at": updated.updated_at} |
|
|
| def run_agent(self, agent_id: str) -> tuple[int, dict]: |
| self.profiles.validate_agent_id(agent_id) |
| old_context = self.store.get_context().context |
|
|
| try: |
| result = self.hermes.run_agent(agent_id, old_context) |
| except HermesRunError as exc: |
| run = self.store.append_run( |
| agent_id=agent_id, |
| status="failed", |
| input_context=old_context, |
| error=str(exc), |
| ) |
| return 502, {"error": str(exc), "run": run} |
|
|
| updated_context = self.store.set_context(result.output) |
| run = self.store.append_run( |
| agent_id=agent_id, |
| status="success", |
| input_context=old_context, |
| output_context=result.output, |
| ) |
| return 200, {"context": updated_context.context, "updated_at": updated_context.updated_at, "run": run} |
|
|
| def list_runs(self) -> list[dict]: |
| return self.store.list_runs() |
|
|
| def get_run(self, run_id: str) -> dict | None: |
| return self.store.get_run(run_id) |
|
|
