75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""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}:
|
|
# Clave deliberadamente distinta de "error": el cuerpo de un AgentExecution
|
|
# correcto ya trae error=None, así que las páginas no podrían distinguir
|
|
# "ejecución sin error" de "la API rechazó la petición" si reusáramos "error".
|
|
return {"api_error": r.json()}
|
|
r.raise_for_status()
|
|
return r.json()
|