File size: 3,264 Bytes
a4514b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | 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
|