53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""API for agent orchestration observability."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
try:
|
|
from fastapi import APIRouter
|
|
except Exception:
|
|
APIRouter = None
|
|
|
|
from .agent_observability import AgentExecutionRecord
|
|
from .agent_observability_dashboard import AgentObservabilityDashboard
|
|
from .agent_observability_store import AgentObservabilityStore
|
|
|
|
store = AgentObservabilityStore()
|
|
dashboard = AgentObservabilityDashboard(store.load())
|
|
|
|
|
|
def record_agent_execution(payload: dict[str, Any]):
|
|
record = AgentExecutionRecord(**payload)
|
|
return {"status": "ok", "record": store.append(record)}
|
|
|
|
|
|
def get_agent_observability(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
|
|
records = store.query(user_id, group_id, role)
|
|
return {
|
|
"status": "ok",
|
|
"records": records,
|
|
"summary": store.load().summary(user_id if role != "owner" else None),
|
|
}
|
|
|
|
|
|
def build_router():
|
|
if APIRouter is None:
|
|
return None
|
|
router = APIRouter(prefix="/admin/ai-evolution/agents", tags=["agent-observability"])
|
|
|
|
@router.post("/execution", operation_id="recordAgentExecution")
|
|
def record(payload: dict[str, Any]):
|
|
return record_agent_execution(payload)
|
|
|
|
@router.get("/overview", operation_id="getAgentObservabilityOverview")
|
|
def overview(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
|
|
return get_agent_observability(user_id, group_id, role)
|
|
|
|
return router
|
|
|
|
|
|
router = build_router()
|
|
|
|
__all__ = ["router", "record_agent_execution", "get_agent_observability"]
|