sysops
548 lines
19 KiB
Python
548 lines
19 KiB
Python
"""Agents API — souls & activity."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.db import fetch_all, fetch_one
|
|
from app.services import agent_approvals, agent_integration, agent_souls
|
|
from app.services.agent_names import normalize_agent_key
|
|
from app.services.agent_terminal import execute_terminal_command
|
|
|
|
router = APIRouter(prefix="/api/agents", tags=["agents-api"])
|
|
|
|
|
|
class SoulUpdate(BaseModel):
|
|
display_name: Optional[str] = None
|
|
role_title: Optional[str] = None
|
|
soul_md: Optional[str] = None
|
|
responsibilities: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class ActionRequestBody(BaseModel):
|
|
agent_key: str
|
|
title: str = Field(..., min_length=1, max_length=255)
|
|
action_type: str = "query"
|
|
query_payload: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class RejectBody(BaseModel):
|
|
reason: str = ""
|
|
|
|
|
|
class ExecuteBody(BaseModel):
|
|
result: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class HandoffBody(BaseModel):
|
|
from_agent: str
|
|
to_agent: str
|
|
handoff_type: str = "partner"
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
correlation_id: Optional[str] = None
|
|
|
|
|
|
class TerminalCommandBody(BaseModel):
|
|
command: str = Field(..., min_length=1, max_length=2000)
|
|
|
|
|
|
def _aggregate_agent_stats() -> dict[str, dict[str, Any]]:
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT agent_name, title, status, created_at
|
|
FROM agent_events
|
|
WHERE created_at >= NOW() - INTERVAL '7 days'
|
|
ORDER BY created_at DESC
|
|
"""
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
by_key: dict[str, dict[str, Any]] = {}
|
|
for r in rows:
|
|
key = normalize_agent_key(str(r.get("agent_name") or ""))
|
|
if not key:
|
|
continue
|
|
bucket = by_key.setdefault(
|
|
key,
|
|
{"events_6h": 0, "errors_24h": 0, "last_event_at": None, "last_event_title": None},
|
|
)
|
|
created = r.get("created_at")
|
|
if bucket["last_event_at"] is None and created is not None:
|
|
bucket["last_event_at"] = created
|
|
bucket["last_event_title"] = r.get("title")
|
|
if created is not None:
|
|
if hasattr(created, "tzinfo") and created.tzinfo is None:
|
|
created = created.replace(tzinfo=timezone.utc)
|
|
if created >= now - timedelta(hours=6):
|
|
bucket["events_6h"] += 1
|
|
if created >= now - timedelta(hours=24) and str(r.get("status") or "") in ("error", "rejected"):
|
|
bucket["errors_24h"] += 1
|
|
return by_key
|
|
|
|
|
|
def _node_health(events_6h: int, errors_24h: int, event_count: int) -> str:
|
|
if events_6h > 0 and errors_24h == 0:
|
|
return "healthy"
|
|
if events_6h > 0:
|
|
return "warn"
|
|
if event_count > 0:
|
|
return "idle"
|
|
return "offline"
|
|
|
|
|
|
@router.get("/collaboration")
|
|
def api_collaboration() -> dict[str, Any]:
|
|
items = agent_integration.list_collaboration()
|
|
return {"items": items, "count": len(items)}
|
|
|
|
|
|
@router.post("/handoff")
|
|
def api_create_handoff(body: HandoffBody) -> dict[str, Any]:
|
|
try:
|
|
handoff = agent_integration.create_handoff(
|
|
body.from_agent,
|
|
body.to_agent,
|
|
handoff_type=body.handoff_type,
|
|
payload=body.payload,
|
|
correlation_id=body.correlation_id,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"ok": True, "handoff": handoff}
|
|
|
|
|
|
def _terminal_payload(ev: dict[str, Any]) -> dict[str, Any]:
|
|
if ev.get("created_at") and hasattr(ev["created_at"], "isoformat"):
|
|
ev["created_at"] = ev["created_at"].isoformat()
|
|
meta = ev.get("metadata")
|
|
if isinstance(meta, str):
|
|
try:
|
|
meta = json.loads(meta)
|
|
except Exception:
|
|
meta = {}
|
|
ev["metadata"] = meta or {}
|
|
line_type = "handoff" if "handoff" in str(ev.get("event_type") or "") else "action"
|
|
if str(ev.get("event_type") or "").startswith("terminal_"):
|
|
line_type = "command" if ev.get("event_type") == "terminal_in" else "output"
|
|
if ev["metadata"].get("target_agent"):
|
|
line_type = "handoff_out"
|
|
if ev["metadata"].get("source_agent"):
|
|
line_type = "handoff_in"
|
|
if str(ev.get("status") or "") in ("error", "rejected"):
|
|
line_type = "error"
|
|
return {
|
|
"type": line_type,
|
|
"id": ev.get("id"),
|
|
"agent": normalize_agent_key(ev.get("agent_name")),
|
|
"message": ev.get("title") or ev.get("event_type"),
|
|
"detail": (ev.get("body") or "")[:500],
|
|
"status": ev.get("status"),
|
|
"correlation_id": ev["metadata"].get("correlation_id"),
|
|
"at": ev.get("created_at"),
|
|
}
|
|
|
|
|
|
@router.get("/souls/{agent_key}/terminal/stream")
|
|
async def api_agent_terminal_stream(agent_key: str, correlation_id: Optional[str] = None):
|
|
key = normalize_agent_key(agent_key)
|
|
soul = agent_souls.get_soul(key)
|
|
if not soul:
|
|
raise HTTPException(404, "Agent not found")
|
|
|
|
async def event_gen():
|
|
last_id = 0
|
|
try:
|
|
hist = fetch_all(
|
|
"""
|
|
SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
|
|
FROM agent_events
|
|
WHERE LOWER(agent_name) = %s
|
|
ORDER BY id DESC
|
|
LIMIT 30
|
|
""",
|
|
(key,),
|
|
)
|
|
for row in reversed(hist):
|
|
last_id = max(last_id, int(row["id"]))
|
|
payload = _terminal_payload(dict(row))
|
|
yield f"data: {json.dumps(payload, default=str)}\n\n"
|
|
except Exception:
|
|
pass
|
|
|
|
while True:
|
|
try:
|
|
if correlation_id:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
|
|
FROM agent_events
|
|
WHERE metadata->>'correlation_id' = %s
|
|
AND id > %s
|
|
ORDER BY id ASC
|
|
LIMIT 50
|
|
""",
|
|
(correlation_id, last_id),
|
|
)
|
|
else:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
|
|
FROM agent_events
|
|
WHERE LOWER(agent_name) = %s AND id > %s
|
|
ORDER BY id ASC
|
|
LIMIT 50
|
|
""",
|
|
(key, last_id),
|
|
)
|
|
for row in rows:
|
|
last_id = max(last_id, int(row["id"]))
|
|
payload = _terminal_payload(dict(row))
|
|
yield f"data: {json.dumps(payload, default=str)}\n\n"
|
|
except Exception as exc:
|
|
err = {"type": "error", "message": str(exc)}
|
|
yield f"data: {json.dumps(err)}\n\n"
|
|
await asyncio.sleep(0.4)
|
|
|
|
return StreamingResponse(
|
|
event_gen(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
@router.get("/live/stream")
|
|
async def api_agents_live_stream():
|
|
"""Single realtime stream for all agent terminals."""
|
|
|
|
async def event_gen():
|
|
last_id = 0
|
|
try:
|
|
recent = fetch_all(
|
|
"""
|
|
SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
|
|
FROM agent_events
|
|
WHERE created_at >= NOW() - INTERVAL '30 minutes'
|
|
ORDER BY id ASC
|
|
LIMIT 80
|
|
"""
|
|
)
|
|
for row in recent:
|
|
last_id = max(last_id, int(row["id"]))
|
|
payload = _terminal_payload(dict(row))
|
|
payload["replay"] = True
|
|
yield f"data: {json.dumps(payload, default=str)}\n\n"
|
|
except Exception:
|
|
pass
|
|
yield f"data: {json.dumps({'type': 'connected', 'last_id': last_id})}\n\n"
|
|
|
|
while True:
|
|
try:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
|
|
FROM agent_events
|
|
WHERE id > %s
|
|
ORDER BY id ASC
|
|
LIMIT 100
|
|
""",
|
|
(last_id,),
|
|
)
|
|
for row in rows:
|
|
last_id = max(last_id, int(row["id"]))
|
|
payload = _terminal_payload(dict(row))
|
|
yield f"data: {json.dumps(payload, default=str)}\n\n"
|
|
except Exception as exc:
|
|
err = {"type": "error", "message": str(exc)}
|
|
yield f"data: {json.dumps(err)}\n\n"
|
|
await asyncio.sleep(0.25)
|
|
|
|
return StreamingResponse(
|
|
event_gen(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
@router.get("/souls")
|
|
def api_list_souls() -> dict[str, Any]:
|
|
return {"items": agent_souls.list_souls(), "count": len(agent_souls.list_souls())}
|
|
|
|
|
|
@router.get("/souls/{agent_key}")
|
|
def api_get_soul(agent_key: str) -> dict[str, Any]:
|
|
soul = agent_souls.get_soul(agent_key)
|
|
if not soul:
|
|
raise HTTPException(404, "Agent not found")
|
|
return {"soul": soul}
|
|
|
|
|
|
@router.get("/souls/{agent_key}/events")
|
|
def api_list_agent_events(agent_key: str, limit: int = 50) -> dict[str, Any]:
|
|
soul = agent_souls.get_soul(agent_key)
|
|
if not soul:
|
|
raise HTTPException(404, "Agent not found")
|
|
items = agent_souls.list_agent_events(agent_key, limit=min(limit, 100))
|
|
return {"agent_key": agent_key.lower(), "items": items, "count": len(items)}
|
|
|
|
|
|
@router.post("/souls/{agent_key}/terminal/command")
|
|
async def api_terminal_command(agent_key: str, body: TerminalCommandBody) -> dict[str, Any]:
|
|
try:
|
|
return await execute_terminal_command(agent_key, body.command.strip())
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
|
|
|
|
@router.put("/souls/{agent_key}")
|
|
def api_update_soul(agent_key: str, body: SoulUpdate) -> dict[str, Any]:
|
|
try:
|
|
soul = agent_souls.update_soul(agent_key, **body.model_dump(exclude_none=True))
|
|
except ValueError as exc:
|
|
raise HTTPException(404, str(exc)) from exc
|
|
return {"soul": soul}
|
|
|
|
|
|
@router.get("/mesh")
|
|
def api_agents_mesh() -> dict[str, Any]:
|
|
souls = agent_souls.list_souls()
|
|
by_key = _aggregate_agent_stats()
|
|
|
|
nodes: list[dict[str, Any]] = []
|
|
for soul in souls:
|
|
key = str(soul.get("agent_key") or "").lower()
|
|
row = by_key.get(key, {})
|
|
events_6h = int(row.get("events_6h") or 0)
|
|
errors_24h = int(row.get("errors_24h") or 0)
|
|
health = _node_health(events_6h, errors_24h, int(soul.get("event_count") or 0))
|
|
is_active = health == "healthy" and events_6h > 0
|
|
node = dict(soul)
|
|
node["health"] = health
|
|
node["events_6h"] = events_6h
|
|
node["errors_24h"] = errors_24h
|
|
node["is_active"] = is_active
|
|
node["last_event_title"] = row.get("last_event_title")
|
|
lat = row.get("last_event_at")
|
|
if lat is not None and hasattr(lat, "isoformat"):
|
|
node["last_event_at"] = lat.isoformat()
|
|
nodes.append(node)
|
|
|
|
report_edges: list[dict[str, Any]] = []
|
|
for node in nodes:
|
|
key = str(node.get("agent_key") or "").lower()
|
|
if key == "herman" or not node.get("is_active"):
|
|
continue
|
|
report_edges.append(
|
|
{
|
|
"source": key,
|
|
"target": "herman",
|
|
"type": "report",
|
|
"active": True,
|
|
"weight": int(node.get("events_6h") or 1),
|
|
}
|
|
)
|
|
|
|
delegate_rows = fetch_all(
|
|
"""
|
|
SELECT metadata, created_at
|
|
FROM agent_events
|
|
WHERE agent_name = 'herman'
|
|
AND created_at >= NOW() - INTERVAL '6 hours'
|
|
AND (
|
|
event_type IN ('openswarm_delegation', 'delegate', 'packaging_delivered', 'telegram_delegation')
|
|
OR metadata ? 'delegated'
|
|
)
|
|
ORDER BY created_at DESC
|
|
LIMIT 100
|
|
"""
|
|
)
|
|
delegate_edges: list[dict[str, Any]] = []
|
|
seen_delegate: set[tuple[str, str]] = set()
|
|
for r in delegate_rows:
|
|
meta = r.get("metadata") or {}
|
|
if isinstance(meta, str):
|
|
try:
|
|
meta = json.loads(meta)
|
|
except Exception:
|
|
meta = {}
|
|
delegated = meta.get("delegated") or meta.get("delegated_agents") or []
|
|
if isinstance(delegated, str):
|
|
delegated = [delegated]
|
|
channel = meta.get("channel") or "herman"
|
|
for agent in delegated:
|
|
tgt = normalize_agent_key(str(agent))
|
|
if not tgt or tgt == "herman":
|
|
continue
|
|
pair = ("herman", tgt)
|
|
if pair in seen_delegate:
|
|
continue
|
|
seen_delegate.add(pair)
|
|
delegate_edges.append(
|
|
{
|
|
"source": "herman",
|
|
"target": tgt,
|
|
"type": "delegate",
|
|
"active": True,
|
|
"channel": channel,
|
|
}
|
|
)
|
|
|
|
peer_static = [
|
|
{
|
|
"source": str(r["from_agent"]),
|
|
"target": str(r["to_agent"]),
|
|
"type": "static",
|
|
"handoff_type": r.get("handoff_type"),
|
|
}
|
|
for r in agent_integration.list_collaboration()
|
|
]
|
|
peer_live = agent_integration.peer_edges_live(hours=6)
|
|
peer_edges = peer_static + peer_live
|
|
|
|
herman_active = any(n.get("agent_key") == "herman" and n.get("is_active") for n in nodes)
|
|
herman_active = herman_active or bool(delegate_edges) or bool(report_edges)
|
|
executive_edges = []
|
|
if herman_active:
|
|
executive_edges = [
|
|
{"source": "herman", "target": "ceo", "type": "executive", "active": True},
|
|
{"source": "herman", "target": "cto", "type": "executive", "active": True},
|
|
]
|
|
|
|
return {
|
|
"nodes": nodes,
|
|
"report_edges": report_edges,
|
|
"delegate_edges": delegate_edges,
|
|
"peer_edges": peer_edges,
|
|
"executive_edges": executive_edges,
|
|
"edges": report_edges,
|
|
"executives": [
|
|
{"id": "ceo", "label": "CEO", "role": "Aissa"},
|
|
{"id": "cto", "label": "CTO", "role": "Platform"},
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/approvals")
|
|
def api_list_approvals(status: Optional[str] = None, limit: int = 50) -> dict[str, Any]:
|
|
try:
|
|
items = agent_approvals.list_requests(status=status, limit=limit)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
return {"items": items, "count": len(items)}
|
|
|
|
|
|
@router.post("/requests")
|
|
def api_create_action_request(body: ActionRequestBody) -> dict[str, Any]:
|
|
try:
|
|
req = agent_approvals.create_request(
|
|
body.agent_key,
|
|
body.title,
|
|
action_type=body.action_type,
|
|
query_payload=body.query_payload,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
auto = None
|
|
rid = (req or {}).get("id")
|
|
if rid:
|
|
try:
|
|
auto = agent_approvals.try_auto_approve_and_execute(int(rid))
|
|
except Exception:
|
|
auto = None
|
|
msg = "Wacht op goedkeuring voordat query uitgevoerd mag worden"
|
|
if auto:
|
|
msg = "Auto-goedgekeurd en uitgevoerd (SysOps policy)"
|
|
return {"ok": True, "request": req, "auto_executed": bool(auto), "message": msg}
|
|
|
|
|
|
@router.post("/approvals/{request_id}/approve")
|
|
def api_approve_request(request_id: int) -> dict[str, Any]:
|
|
import os
|
|
|
|
import httpx
|
|
|
|
try:
|
|
req = agent_approvals.approve_request(request_id, approved_by="ceo")
|
|
action = req.get("action_type") or ""
|
|
auto_result = None
|
|
|
|
if action in ("config_backup", "maintenance_scan"):
|
|
tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
|
|
path = "/ops/backup/run" if action == "config_backup" else "/ops/maintenance/scan"
|
|
payload = {"approval_request_id": request_id} if action == "config_backup" else None
|
|
with httpx.Client(timeout=180.0) as client:
|
|
resp = client.post(f"{tools_url}{path}", json=payload or {})
|
|
try:
|
|
auto_result = resp.json()
|
|
except Exception:
|
|
auto_result = {"ok": False, "detail": resp.text}
|
|
req = agent_approvals.mark_executed(request_id, auto_result or {})
|
|
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
return {"ok": True, "request": req, "executed": auto_result is not None, "result": auto_result}
|
|
|
|
|
|
@router.post("/approvals/{request_id}/reject")
|
|
def api_reject_request(request_id: int, body: RejectBody) -> dict[str, Any]:
|
|
try:
|
|
req = agent_approvals.reject_request(request_id, reason=body.reason)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
return {"ok": True, "request": req}
|
|
|
|
|
|
@router.post("/approvals/{request_id}/execute")
|
|
def api_execute_approved_request(request_id: int, body: ExecuteBody) -> dict[str, Any]:
|
|
import os
|
|
|
|
import httpx
|
|
|
|
try:
|
|
req = agent_approvals.require_approved(request_id)
|
|
result_payload = body.result or {}
|
|
action = req.get("action_type") or ""
|
|
|
|
if action == "config_backup":
|
|
tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
|
|
with httpx.Client(timeout=180.0) as client:
|
|
resp = client.post(
|
|
f"{tools_url}/ops/backup/run",
|
|
json={"approval_request_id": request_id},
|
|
)
|
|
try:
|
|
result_payload = resp.json()
|
|
except Exception:
|
|
result_payload = {"ok": False, "detail": resp.text}
|
|
if resp.status_code >= 400:
|
|
raise HTTPException(status_code=502, detail=result_payload.get("detail") or resp.text)
|
|
|
|
elif action == "maintenance_scan":
|
|
tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
|
|
with httpx.Client(timeout=120.0) as client:
|
|
resp = client.post(f"{tools_url}/ops/maintenance/scan")
|
|
result_payload = resp.json() if resp.status_code < 400 else {"ok": False, "detail": resp.text}
|
|
|
|
req = agent_approvals.mark_executed(request_id, result_payload)
|
|
except HTTPException:
|
|
raise
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc)) from exc
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
return {"ok": True, "request": req}
|