78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
"""Actions API for unified Agent Event Bus."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
try:
|
|
from fastapi import APIRouter
|
|
except Exception:
|
|
APIRouter = None
|
|
|
|
from .agent_event_bus import AgentEventBus
|
|
|
|
bus = AgentEventBus()
|
|
|
|
|
|
def create_agent_event(payload: dict[str, Any]) -> dict[str, Any]:
|
|
return {"status": "ok", "event": bus.publish(**payload)}
|
|
|
|
|
|
def list_agent_events(kind: str | None = None, status: str = "pending") -> dict[str, Any]:
|
|
return {"status": "ok", "events": bus.list_events(kind, status)}
|
|
|
|
|
|
def get_agent_event(event_id: str) -> dict[str, Any]:
|
|
return {"status": "ok", "event": bus.get(event_id)}
|
|
|
|
|
|
def transition_agent_event(event_id: str, status: str, actor: str = "system") -> dict[str, Any]:
|
|
return {"status": "ok", "event": bus.transition(event_id, status, actor)}
|
|
|
|
|
|
def agent_event_stats() -> dict[str, Any]:
|
|
return {"status": "ok", "stats": bus.stats()}
|
|
|
|
|
|
# Compatibility adapters
|
|
|
|
def list_gpt_inbox_messages():
|
|
return list_agent_events(kind="gpt_inbox")
|
|
|
|
|
|
def ack_gpt_inbox_message(event_id: str):
|
|
return transition_agent_event(event_id, "completed", "custom-gpt")
|
|
|
|
|
|
def build_router():
|
|
if APIRouter is None:
|
|
return None
|
|
router = APIRouter(prefix="/agent-events", tags=["agent-events"])
|
|
|
|
@router.post("", operation_id="createAgentEvent")
|
|
def create(payload: dict[str, Any]):
|
|
return create_agent_event(payload)
|
|
|
|
@router.get("", operation_id="listAgentEvents")
|
|
def listing(kind: str | None = None, status: str = "pending"):
|
|
return list_agent_events(kind, status)
|
|
|
|
@router.get("/{event_id}", operation_id="getAgentEvent")
|
|
def get(event_id: str):
|
|
return get_agent_event(event_id)
|
|
|
|
@router.post("/{event_id}/transition", operation_id="transitionAgentEvent")
|
|
def transition(event_id: str, payload: dict[str, Any]):
|
|
return transition_agent_event(event_id, payload["status"], payload.get("actor", "system"))
|
|
|
|
@router.get("/stats/overview", operation_id="getAgentEventStats")
|
|
def stats():
|
|
return agent_event_stats()
|
|
|
|
return router
|
|
|
|
|
|
router = build_router()
|
|
|
|
__all__ = ["router", "create_agent_event", "list_agent_events", "get_agent_event", "transition_agent_event"]
|