File size: 3,177 Bytes
02117ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """
God Agent Orchestrator API Routes
"""
from fastapi import APIRouter, Request, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any
router = APIRouter()
class OrchestrateRequest(BaseModel):
message: str
session_id: str = ""
task_id: str = ""
context: Dict[str, Any] = {}
class SandboxExecRequest(BaseModel):
command: str
cwd: str = ""
timeout: int = 30
class FileWriteRequest(BaseModel):
filename: str
content: str
@router.post("/orchestrate")
async def orchestrate(req: OrchestrateRequest, request: Request):
"""Route message through God Agent Orchestrator."""
orchestrator = getattr(request.app.state, "orchestrator", None)
if not orchestrator:
raise HTTPException(500, "Orchestrator not initialized")
result = await orchestrator.orchestrate(
user_message=req.message,
session_id=req.session_id,
task_id=req.task_id,
context=req.context,
)
return {"result": result, "session_id": req.session_id}
@router.get("/status")
async def agent_status(request: Request):
"""Get all agent statuses."""
orchestrator = getattr(request.app.state, "orchestrator", None)
if not orchestrator:
return {"status": "not_initialized"}
return orchestrator.get_status()
@router.post("/sandbox/execute")
async def sandbox_execute(req: SandboxExecRequest, request: Request):
"""Execute command in sandbox."""
orchestrator = getattr(request.app.state, "orchestrator", None)
if not orchestrator:
raise HTTPException(500, "Orchestrator not initialized")
sandbox = orchestrator.get_agent("sandbox")
if not sandbox:
raise HTTPException(503, "SandboxAgent not available")
result = await sandbox.execute(req.command, cwd=req.cwd, timeout=req.timeout)
return {"result": result, "command": req.command}
@router.post("/sandbox/file")
async def sandbox_write_file(req: FileWriteRequest, request: Request):
"""Write file to sandbox workspace."""
orchestrator = getattr(request.app.state, "orchestrator", None)
if not orchestrator:
raise HTTPException(500, "Orchestrator not initialized")
sandbox = orchestrator.get_agent("sandbox")
if not sandbox:
raise HTTPException(503, "SandboxAgent not available")
result = await sandbox.write_file(req.filename, req.content)
return {"result": result, "filename": req.filename}
@router.get("/sandbox/workspace")
async def sandbox_workspace(request: Request):
"""Get workspace info."""
orchestrator = getattr(request.app.state, "orchestrator", None)
if not orchestrator:
raise HTTPException(500, "Orchestrator not initialized")
sandbox = orchestrator.get_agent("sandbox")
if not sandbox:
raise HTTPException(503, "SandboxAgent not available")
info = await sandbox.get_workspace_info()
return info
@router.get("/ai-router/stats")
async def ai_router_stats(request: Request):
"""Get AI router statistics."""
ai_router = getattr(request.app.state, "ai_router", None)
if not ai_router:
return {"stats": {}}
return {"stats": ai_router.get_stats()}
|