""" FastAPI Web Application ======================== Developer console and API endpoints for the multi-agent system. """ import sys import os from pathlib import Path # Add src to path sys.path.insert(0, str(Path(__file__).parent.parent)) from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import Any, Optional from orchestrator import Orchestrator from ledger.merkle import compute_merkle_root app = FastAPI( title="AgentMask Developer Console", description="Multi-agent system with audit trail", version="0.1.0" ) class TaskRequest(BaseModel): """Request model for running a task.""" query: str options: Optional[dict[str, Any]] = None class TaskResponse(BaseModel): """Response model for task execution.""" success: bool task: dict[str, Any] steps: list[dict[str, Any]] final_output: dict[str, Any] merkle_root: str total_steps: int # Global orchestrator instance orchestrator = Orchestrator() @app.get("/", response_class=HTMLResponse) async def get_console(): """ Serve the developer console HTML page. """ html_content = """ AgentMask Developer Console

⌘ AgentMask Terminal

> Awaiting command input...
> Type query and execute to initialize agent pipeline_

> AGENT PIPELINE GRAPH_

graph LR A[🔍 Input Query] --> B[📚 Research Agent] B --> C[📝 Summarizer Agent] C --> D[✅ Output]

sequenceDiagram participant U as User participant O as Orchestrator participant R as Research participant S as Summarizer U->>O: Submit Query O->>R: Execute Search R-->>O: Search Results O->>S: Summarize Results S-->>O: Summary O-->>U: Final Output

graph TB Root[🔐 Merkle Root] Root --> H1[Hash 1-2] Root --> H2[Hash 3-4] H1 --> L1[Step 1 Hash] H1 --> L2[Step 2 Hash]

""" return HTMLResponse(content=html_content) @app.post("/run", response_model=TaskResponse) async def run_task(request: TaskRequest): """ Execute a task through the multi-agent pipeline. Args: request: TaskRequest with query and optional parameters Returns: TaskResponse with execution results and merkle root """ try: # Run the orchestrator result = await orchestrator.run_task({"query": request.query}) # Compute merkle root from all step hashes step_hashes = [step["hash"] for step in result["steps"]] merkle_root = compute_merkle_root(step_hashes) if step_hashes else "" return TaskResponse( success=True, task=result["task"], steps=result["steps"], final_output=result["final_output"], merkle_root=merkle_root, total_steps=result["total_steps"] ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "version": "0.1.0"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)