feat(dashboard): scaffolding Streamlit con cliente HTTP y health check

CoreClient (httpx sync, 2 retries) cubre todos los endpoints del core y mapea
404/409/422 a {"error": ...}. app.py monta la página raíz con sidebar de branding
y un health-check del core.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Juan
2026-05-11 12:39:16 +02:00
co-authored by Claude Opus 4.7
parent 00a5dc83fa
commit a27e972360
2 changed files with 118 additions and 0 deletions
@@ -0,0 +1,71 @@
"""Cliente HTTP tipado al core. Encapsula httpx y mapea respuestas a dicts."""
from __future__ import annotations
import os
from typing import Any
import httpx
class CoreClient:
"""Cliente sincrono (Streamlit es sync). 2 retries con httpx Transport."""
def __init__(self, base_url: str | None = None, timeout: float = 30.0) -> None:
self.base_url = base_url or os.getenv("AGENTFORGE_CORE_URL", "http://localhost:8000")
transport = httpx.HTTPTransport(retries=2)
self._client = httpx.Client(base_url=self.base_url, timeout=timeout, transport=transport)
def health(self) -> dict:
return self._get("/health")
def list_agents(self) -> list[dict]:
return self._get("/agents")
def get_agent(self, name: str) -> dict:
return self._get(f"/agents/{name}")
def list_versions(self, name: str) -> list[dict]:
return self._get(f"/agents/{name}/versions")
def diff_versions(self, name: str, v_from: str, v_to: str) -> dict:
return self._get(f"/agents/{name}/versions/{v_from}/diff/{v_to}")
def invoke_agent(self, name: str, body: dict) -> dict:
return self._post(f"/agents/{name}/invoke", body)
def get_execution(self, trace_id: str) -> dict:
return self._get(f"/executions/{trace_id}")
def list_executions(self) -> list[dict]:
return self._get("/executions")
def approve(self, trace_id: str, body: dict) -> dict:
return self._post(f"/executions/{trace_id}/approve", body)
def reject(self, trace_id: str, body: dict) -> dict:
return self._post(f"/executions/{trace_id}/reject", body)
def list_violations(self, **filters: Any) -> list[dict]:
params = {k: v for k, v in filters.items() if v is not None}
return self._get("/violations", params=params)
def list_policies(self) -> list[dict]:
return self._get("/policies")
def list_policy_versions(self, name: str) -> list[dict]:
return self._get(f"/policies/{name}/versions")
# ----- helpers privados -----
def _get(self, path: str, params: dict | None = None) -> Any:
r = self._client.get(path, params=params)
r.raise_for_status()
return r.json()
def _post(self, path: str, body: dict) -> Any:
r = self._client.post(path, json=body)
if r.status_code in {404, 409, 422}:
return {"error": r.json()}
r.raise_for_status()
return r.json()