Multi-Agent Task Orchestrator

Production-Grade Agentic AI Execution & Self-Correction Platform

The Multi-Agent Task Orchestrator is an autonomous software development platform powered by a graph-based state machine, Google Gemini API, LangGraph, Docker sandboxing, FastAPI, WebSockets, and a modern React execution dashboard.

Given a natural language request (such as "Create a Python program to calculate the average marks of students from a CSV file"), the system coordinates specialized agents to plan, code, audit security, execute tests in a Docker sandbox, review outcomes, and autonomously fix errors until the solution is approved.

---

🏛️ Architecture Overview

                         USER
                           │
              ┌────────────┴────────────┐
              │                         │
             CLI                    React UI
              │                         │
              └────────────┬────────────┘
                           │
                        FastAPI
                           │
                    ┌──────▼──────┐
                    │ Orchestrator│
                    └──────┬──────┘
                           │
                     ┌─────▼─────┐
                     │   Graph   │
                     │  Engine   │
                     └─────┬─────┘
                           │
                    ┌──────▼──────┐
                    │  TaskState  │
                    └──────┬──────┘
                           │
                     ┌─────▼─────┐
                     │  Planner  │
                     └─────┬─────┘
                           ▼
                     ┌───────────┐
                     │   Coder   │◄──────────────┐
                     └─────┬─────┘               │
                           ▼                      │
                     ┌───────────┐               │
                     │ Security  │               │
                     └─────┬─────┘               │
                           │                      │
                         SAFE                     │
                           ▼                      │
                     ┌───────────┐               │
                     │  Tester   │               │
                     └─────┬─────┘               │
                           ▼                      │
                     ┌───────────┐               │
                     │ Reviewer  │               │
                     └─────┬─────┘               │
                           ▼                      │
                       ROUTER                     │
                    ┌──────┼────────┐             │
                    │      │        │             │
                    ▼      ▼        ▼             │
                 SUCCESS   FIX   EXHAUSTED        │
                    │      │        │             │
                    ▼      └────────┘             │
                   END            └───────────────┘

---

🚀 Key Features

• State Machine Workflow: LangGraph state machine coordinating Planner, Coder, Security, Tester, and Reviewer agents.
• Structured Pydantic State & Outputs: Communication takes place entirely through strongly-typed state objects (TaskState, PlannerResult, CoderResult, SecurityResult, TesterResult, ReviewResult).
• Autonomous Self-Correction Loop: Automatically captures pytest tracebacks, reviewer feedback, and security alerts to retry and refine code (up to MAX_ITERATIONS = 5).
• AST & LLM Security Audit: Dual static AST analysis and LLM inspection catch dangerous calls (os.system, subprocess, shutil.rmtree, credential access) and transition state to BLOCKED.
• Docker Sandbox Execution: Generated code is executed inside isolated Docker containers (python:3.11-slim) with memory limits (256m), CPU quotas, and timeouts (30s). Fail-safe policy prevents silent fallback to host execution unless explicitly requested via --unsafe-local-execution.
• FastAPI REST API & WebSockets: Asynchronous task submission, state querying, and live WebSocket streaming of lifecycle event logs.
• Rich CLI: Terminal runner (python app/cli.py) with rich syntax highlighting and execution progress displays.
• React Dashboard: Interactive UI visualizing active workflow nodes, real-time log streams, code editor, and test pass/fail statistics.

---

🛠️ Installation & Setup

Prerequisites

• Python 3.11+
• Node.js 18+ & npm
• Docker Desktop (optional, recommended for sandboxing)

1. Clone & Setup Backend

cd "Multi-Agent Task Orchestrator"

# Install Python dependencies
python -m pip install -r requirements.txt

# Copy environment variables configuration
cp .env.example .env

Set your Gemini API Key in .env:

GEMINI_API_KEY=your_actual_gemini_api_key_here
GEMINI_MODEL=gemini-2.5-flash
MAX_ITERATIONS=5

2. Setup React Frontend

cd frontend
npm install
npm run build

---

💻 Usage

1. Command Line Interface (CLI)

Run an autonomous orchestration task directly from terminal:

# Run task using Docker Sandbox (or with --unsafe-local-execution fallback)
python app/cli.py "Create a Python program that reads students.csv and calculates the average marks." -u

# Check status of a task
python app/cli.py status task-12345678

# View lifecycle logs for a task
python app/cli.py logs task-12345678

2. FastAPI Backend Server

Start the API server:

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Access API Documentation:

• Swagger UI: http://localhost:8000/docs
• Redoc: http://localhost:8000/redoc

API Endpoints:

• POST /api/tasks — Submit task
• GET /api/tasks — List all tasks
• GET /api/tasks/{task_id} — Get full task detail
• GET /api/tasks/{task_id}/status — Get status payload
• GET /api/tasks/{task_id}/logs — Get event lifecycle logs
• GET /api/tasks/{task_id}/result — Get generated code and evaluation outputs
• WS /api/tasks/{task_id}/ws — WebSocket event stream

3. React Execution Dashboard

Start the frontend development server:

cd frontend
npm run dev

Open http://localhost:3000 in your browser.

---

🧪 Running the Test Suite

Run the full automated test suite (unit, integration, graph, router, execution, state, and failure injection tests):

python -m pytest

Expected Output:

============================= test session starts =============================
platform win32 -- Python 3.11.7, pytest-9.1.1, pluggy-1.6.0
collected 31 items

tests\test_agents.py ..........                                          [ 32%]
tests\test_api.py .                                                      [ 35%]
tests\test_db.py .                                                       [ 38%]
tests\test_execution.py ....                                            [ 51%]
tests\test_failure_injection.py ....                                    [ 64%]
tests\test_graph.py ....                                                 [ 77%]
tests\test_router.py .......                                            [100%]

============================== 31 passed in 4.09s ==============================

---

📁 Directory Structure

multi-agent-orchestrator/
│
├── app/
│   ├── main.py                  # FastAPI Application & WebSockets
│   ├── cli.py                   # Typer CLI Application
│   ├── config.py                # Environment & Settings
│   ├── orchestrator/
│   │   ├── state.py             # TaskState Pydantic Model
│   │   ├── graph.py             # LangGraph Workflow Engine
│   │   ├── router.py            # Decision Routers (Security & Review)
│   │   └── orchestrator.py      # TaskOrchestrator Facade
│   ├── agents/
│   │   ├── base.py              # BaseAgent (Gemini LLM Wrapper)
│   │   ├── planner.py           # Planner Agent
│   │   ├── coder.py             # Coder Agent
│   │   ├── security.py          # Security Agent (AST + LLM)
│   │   ├── tester.py            # Tester Agent
│   │   └── reviewer.py          # Reviewer Agent
│   ├── schemas/
│   │   ├── task.py              # TaskState & TaskStatus Schemas
│   │   ├── agent.py             # Planner, Coder, Security, Reviewer Schemas
│   │   ├── execution.py         # ExecutionResult & TestFailureDetail
│   │   └── results.py           # API Response Models
│   ├── execution/
│   │   ├── sandbox.py           # ExecutionManager (Fail-safe router)
│   │   ├── docker_runner.py     # Docker Container Sandbox Runner
│   │   ├── process_runner.py    # Local Process Runner (Unsafe Fallback)
│   │   ├── resource_limits.py   # Container Resource Quotas
│   │   └── result_parser.py     # Pytest Output Parser
│   ├── db/
│   │   └── database.py          # SQLite Persistence (SQLAlchemy + aiosqlite)
│   └── utils/
│       └── logger.py            # Structured EventLogger & Broadcaster
│
├── sandbox/
│   ├── Dockerfile               # Python 3.11-slim pytest Sandbox Image
│   └── requirements.txt
│
├── frontend/
│   ├── src/
│   │   ├── components/
│   │   │   ├── TaskInput.jsx
│   │   │   ├── ExecutionGraph.jsx
│   │   │   ├── CodeEditorView.jsx
│   │   │   ├── TestResultsCard.jsx
│   │   │   ├── LogTerminal.jsx
│   │   │   └── TaskHistory.jsx
│   │   ├── App.jsx
│   │   └── index.css
│   ├── package.json
│   └── vite.config.js
│
├── tests/
│   ├── test_state.py
│   ├── test_agents.py
│   ├── test_execution.py
│   ├── test_router.py
│   ├── test_graph.py
│   ├── test_db.py
│   ├── test_api.py
│   └── test_failure_injection.py
│
├── requirements.txt
├── .env.example
├── .gitignore
└── README.md

---

📜 License

MIT License. Built for Autonomous Agentic AI Software Development.