| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from urllib.parse import unquote |
|
|
| from .profiles import InvalidAgentId, UnknownAgent |
| from .service import BackendService |
|
|
|
|
| @dataclass(frozen=True) |
| class ApiResponse: |
| status: int |
| body: dict | list |
|
|
|
|
| class BackendApi: |
| ENDPOINTS = [ |
| "GET /", |
| "GET /api/health", |
| "GET /api/agents", |
| "GET /api/context", |
| "POST /api/context", |
| "POST /api/run-agent", |
| "GET /api/runs", |
| "GET /api/runs/<run_id>", |
| ] |
|
|
| def __init__(self, service: BackendService) -> None: |
| self.service = service |
|
|
| def handle(self, method: str, path: str, payload: dict | None = None) -> ApiResponse: |
| payload = payload or {} |
| normalized_method = method.upper() |
| normalized_path = self._normalize_path(path) |
| try: |
| return self._handle(normalized_method, normalized_path, payload) |
| except InvalidAgentId as exc: |
| return ApiResponse(400, {"error": str(exc)}) |
| except UnknownAgent as exc: |
| return ApiResponse(404, {"error": str(exc)}) |
| except ValueError as exc: |
| return ApiResponse(400, {"error": str(exc)}) |
|
|
| def _handle(self, method: str, path: str, payload: dict) -> ApiResponse: |
| if method == "GET" and path == "/": |
| return ApiResponse(200, {"service": "clfgui-api", "endpoints": self.ENDPOINTS}) |
|
|
| if method == "GET" and path == "/api/health": |
| return ApiResponse(200, {"status": "ok"}) |
|
|
| if method == "GET" and path == "/api/agents": |
| return ApiResponse(200, {"agents": self.service.list_agents()}) |
|
|
| if method == "GET" and path == "/api/context": |
| return ApiResponse(200, self.service.get_context()) |
|
|
| if method == "POST" and path == "/api/context": |
| context = payload.get("context") |
| if not isinstance(context, str): |
| raise ValueError("context must be a string.") |
| return ApiResponse(200, self.service.set_context(context)) |
|
|
| if method == "POST" and path == "/api/run-agent": |
| agent_id = payload.get("agent_id") |
| if not isinstance(agent_id, str): |
| raise ValueError("agent_id must be a string.") |
| status, body = self.service.run_agent(agent_id) |
| return ApiResponse(status, body) |
|
|
| if method == "GET" and path == "/api/runs": |
| return ApiResponse(200, {"runs": self.service.list_runs()}) |
|
|
| if method == "GET" and path.startswith("/api/runs/"): |
| run_id = unquote(path.removeprefix("/api/runs/")) |
| run = self.service.get_run(run_id) |
| if run is None: |
| return ApiResponse(404, {"error": "Run not found."}) |
| return ApiResponse(200, run) |
|
|
| return ApiResponse( |
| 404, |
| { |
| "error": "Not found.", |
| "method": method, |
| "path": path, |
| "endpoints": self.ENDPOINTS, |
| }, |
| ) |
|
|
| @staticmethod |
| def _normalize_path(path: str) -> str: |
| if not path: |
| return "/" |
| if path != "/" and path.endswith("/"): |
| return path.rstrip("/") |
| return path |
|
|
