from __future__ import annotations import json from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib.parse import parse_qs, urlparse from .service import MemoryService def _parse_bool(value: str | None, *, default: bool = False) -> bool: if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _first(query: dict[str, list[str]], key: str) -> str | None: values = query.get(key) if not values: return None value = values[0].strip() return value or None def _first_int(query: dict[str, list[str]], key: str, default: int) -> int: value = _first(query, key) if value is None: return default return int(value) def _metadata_filters(query: dict[str, list[str]]) -> dict[str, Any] | None: metadata: dict[str, Any] = {} for key, values in query.items(): if not key.startswith("metadata."): continue name = key.removeprefix("metadata.").strip() if not name: continue value = values[0].strip() if value: metadata[name] = value return metadata or None def build_handler(service: MemoryService): class MemoryAPIHandler(BaseHTTPRequestHandler): def do_OPTIONS(self) -> None: # noqa: N802 self.send_response(HTTPStatus.NO_CONTENT) self._send_common_headers(content_type="application/json; charset=utf-8", content_length=0) self.end_headers() def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) query = parse_qs(parsed.query) try: if parsed.path == "/api/status": payload = { "ok": True, "db_path": service.store.db_path, "dashboard": service.dashboard_snapshot(), } try: payload["audit"] = service.audit_v2() except Exception as exc: payload["audit_error"] = str(exc) self._send_json(payload) return if parsed.path == "/api/search": search_query = _first(query, "q") if search_query is None: self.send_error(HTTPStatus.BAD_REQUEST, "Missing query parameter 'q'") return scopes = [scope.strip() for scope in query.get("scope", []) if scope.strip()] filters = { key: value for key, value in { "type": _first(query, "type"), "subtype": _first(query, "subtype"), "project_id": _first(query, "project_id"), "repo_id": _first(query, "repo_id"), "source_ref": _first(query, "source_ref"), "evidence_ref": _first(query, "evidence_ref"), "run_id": _first(query, "run_id"), "task_id": _first(query, "task_id"), "origin_agent": _first(query, "origin_agent"), "url": _first(query, "url"), "domain": _first(query, "domain"), }.items() if value is not None } self._send_json( service.search( search_query, scopes or None, filters=filters or None, limit=_first_int(query, "limit", 10), include_inbox=_parse_bool(_first(query, "include_inbox")), ) ) return if parsed.path == "/api/context": self._send_json( service.context_for( project=_first(query, "project"), repo=_first(query, "repo"), agent=_first(query, "agent"), task=_first(query, "task"), ) ) return if parsed.path == "/api/memories": self._send_json( { "items": service.list_memories( status=_first(query, "status") if "status" in query else "active", scope=_first(query, "scope"), memory_type=_first(query, "type"), subtype=_first(query, "subtype"), project_id=_first(query, "project_id"), repo_id=_first(query, "repo_id"), source_ref=_first(query, "source_ref"), evidence_ref=_first(query, "evidence_ref"), run_id=_first(query, "run_id"), task_id=_first(query, "task_id"), origin_agent=_first(query, "origin_agent"), url=_first(query, "url"), domain=_first(query, "domain"), metadata=_metadata_filters(query), limit=_first_int(query, "limit", 50), ) } ) return if parsed.path.startswith("/api/memories/"): self._send_json(service.get_memory(parsed.path.split("/")[3])) return if parsed.path == "/api/tasks": requires_human_input = None if "requires_human_input" in query: requires_human_input = _parse_bool(_first(query, "requires_human_input")) self._send_json( { "items": service.list_tasks( status=_first(query, "status"), owner_agent=_first(query, "owner_agent"), requires_human_input=requires_human_input, run_id=_first(query, "run_id"), limit=_first_int(query, "limit", 50), ) } ) return if parsed.path.startswith("/api/tasks/") and parsed.path.endswith("/bundle"): self._send_json(service.task_bundle(parsed.path.split("/")[3])) return if parsed.path.startswith("/api/tasks/"): self._send_json(service.get_task(parsed.path.split("/")[3])) return if parsed.path == "/api/task-runs": self._send_json( { "items": service.list_task_runs( task_id=_first(query, "task_id"), status=_first(query, "status"), limit=_first_int(query, "limit", 50), ) } ) return if parsed.path.startswith("/api/task-runs/"): self._send_json(service.get_task_run(parsed.path.split("/")[3])) return if parsed.path == "/api/artifacts": self._send_json( { "items": service.list_artifacts( task_id=_first(query, "task_id"), limit=_first_int(query, "limit", 50), ) } ) return if parsed.path.startswith("/api/artifacts/"): self._send_json(service.get_artifact(parsed.path.split("/")[3])) return except KeyError as exc: self.send_error(HTTPStatus.NOT_FOUND, str(exc)) return except ValueError as exc: self.send_error(HTTPStatus.BAD_REQUEST, str(exc)) return except Exception as exc: self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) return self.send_error(HTTPStatus.NOT_FOUND) def do_POST(self) -> None: # noqa: N802 parsed = urlparse(self.path) try: payload = self._json_body() except (UnicodeDecodeError, json.JSONDecodeError, ValueError): self.send_error(HTTPStatus.BAD_REQUEST, "Invalid JSON body") return try: if parsed.path == "/api/ingest": self._send_json(service.ingest(payload), status=HTTPStatus.CREATED) return if parsed.path == "/api/tasks": self._send_json( service.create_task( title=payload["title"], intent=payload["intent"], kind=payload.get("kind", "task"), status=payload.get("status", "open"), priority=payload.get("priority", 3), project_id=payload.get("project_id"), repo_id=payload.get("repo_id"), parent_task_id=payload.get("parent_task_id"), origin=payload.get("origin"), owner_agent=payload.get("owner_agent"), blocked_reason=payload.get("blocked_reason"), requires_human_input=bool(payload.get("requires_human_input", False)), due_at=payload.get("due_at"), metadata=payload.get("metadata"), task_id=payload.get("task_id"), run_id=payload.get("run_id"), ), status=HTTPStatus.CREATED, ) return if parsed.path == "/api/task-runs": task_id = payload.get("task_id") agent_id = payload.get("agent_id") if not isinstance(task_id, str) or not task_id.strip(): self.send_error(HTTPStatus.BAD_REQUEST, "Missing or invalid 'task_id'") return if not isinstance(agent_id, str) or not agent_id.strip(): self.send_error(HTTPStatus.BAD_REQUEST, "Missing or invalid 'agent_id'") return self._send_json( service.start_task_run( task_id, agent_id, input_payload=payload.get("input_payload"), metadata=payload.get("metadata"), ), status=HTTPStatus.CREATED, ) return if parsed.path.startswith("/api/task-runs/") and parsed.path.endswith("/finish"): run_id = parsed.path.split("/")[3] status = payload.get("status") if not isinstance(status, str) or not status.strip(): self.send_error(HTTPStatus.BAD_REQUEST, "Missing or invalid 'status'") return self._send_json( service.finish_task_run( run_id, status=status, result_summary=payload.get("result_summary"), error_message=payload.get("error_message"), ) ) return if parsed.path == "/api/artifacts": task_id = payload.get("task_id") artifact_type = payload.get("artifact_type") title = payload.get("title") content = payload.get("content") if not all(isinstance(value, str) and value.strip() for value in (task_id, artifact_type, title, content)): self.send_error( HTTPStatus.BAD_REQUEST, "Missing or invalid required artifact fields", ) return self._send_json( service.create_artifact( task_id=task_id, artifact_type=artifact_type, title=title, content=content, metadata=payload.get("metadata"), source_ref=payload.get("source_ref"), ), status=HTTPStatus.CREATED, ) return except KeyError as exc: self.send_error(HTTPStatus.BAD_REQUEST, f"Missing field: {exc}") return except Exception as exc: self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) return self.send_error(HTTPStatus.NOT_FOUND) def do_PATCH(self) -> None: # noqa: N802 parsed = urlparse(self.path) if not parsed.path.startswith("/api/tasks/"): self.send_error(HTTPStatus.NOT_FOUND) return try: payload = self._json_body() except (UnicodeDecodeError, json.JSONDecodeError, ValueError): self.send_error(HTTPStatus.BAD_REQUEST, "Invalid JSON body") return try: self._send_json( service.update_task( parsed.path.split("/")[3], status=payload.get("status"), owner_agent=payload.get("owner_agent"), blocked_reason=payload.get("blocked_reason"), requires_human_input=payload.get("requires_human_input"), metadata=payload.get("metadata"), run_id=payload.get("run_id"), ) ) except KeyError as exc: self.send_error(HTTPStatus.NOT_FOUND, str(exc)) except Exception as exc: self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) def log_message(self, fmt: str, *args: Any) -> None: return def _json_body(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length) if length else b"{}" payload = json.loads(raw.decode("utf-8")) if not isinstance(payload, dict): raise ValueError("JSON body must be an object") return payload def _send_common_headers(self, *, content_type: str, content_length: int) -> None: self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(content_length)) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Allow-Methods", "GET,POST,PATCH,OPTIONS") def _send_json(self, payload: dict[str, Any], *, status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload, ensure_ascii=True).encode("utf-8") self.send_response(status) self._send_common_headers(content_type="application/json; charset=utf-8", content_length=len(body)) self.end_headers() self.wfile.write(body) return MemoryAPIHandler def run_http_server(*, db_path: str, host: str = "127.0.0.1", port: int = 8091) -> None: service = MemoryService(db_path) server = ThreadingHTTPServer((host, port), build_handler(service)) print(f"agents-database listening on http://{host}:{port}", flush=True) try: server.serve_forever() finally: server.server_close()