superbizon007/orchestrator-agent repository overview

⁠orchestrator-agent

Execution engine that accepts a validated JSON plan over the Google A2A protocol⁠, executes it as a directed acyclic graph (DAG) against external A2A agents, persists run and step state in SQLite, and returns structured results.

The orchestrator is a pure execution engine — no LLM, no planning logic. Upstream plan sources (plan_self_ask_agent, plan_judge_agent) submit plans as A2A tasks; the orchestrator validates, schedules, and drives them to completion.

⁠What it does

1. Receives a plan JSON payload via A2A (message/send)
2. Validates schema and DAG semantics (cycles, unreachable nodes, unknown agents, etc.)
3. Executes each step in topological order, invoking external A2A agents via HTTP
4. Persists run and step state in SQLite after every attempt
5. Retries on transient failures (HTTP 429/502/503/504, timeouts) with exponential backoff
6. Falls back to an alternate agent when configured
7. Replans on step failure when on_failure is "replan" or "replan_with_approval"
8. Accepts mid-run amendments via the A2A inbox (send a new plan JSON to the running task)
9. Returns run_id and final status as an A2A artifact

⁠Environment variables

⁠Secret resolution

Secret refs in the plan (e.g. secrets/gmail/password) are resolved at execution time from environment variables:

SECRET_SECRETS_GMAIL_PASSWORD=hunter2

The ref is uppercased and / and - are replaced with _, then the SECRET_ prefix is prepended.

⁠Trajectory recording

When TRAJECTORY_DIR is set, per-step files are written under TRAJECTORY_DIR/<run_id>/step_<step_id>/:

• resolved_inputs.json — inputs after secret injection and upstream output propagation
• agent_request.json — full A2A JSON-RPC body sent to the agent
• agent_response.json — raw A2A response received
• step_result.json — final status, outputs, error, duration in milliseconds

⁠Build

Build from inside this directory (apps/orchestrator_agent/). The shared packages/a2a_agent package is passed as a named build context using Docker BuildKit's --build-context flag (Docker Desktop or Docker Engine ≥ 23).

# From apps/orchestrator_agent/
docker build \
  --build-context a2a_agent=../../packages/a2a_agent \
  -t orchestrator-agent:latest \
  .

⁠Run

⁠Minimal (no auth, plain HTTP)

docker run --rm \
  -p 8080:8080 \
  orchestrator-agent:latest

⁠With API key authentication

docker run --rm \
  -e API_KEY=my-secret-token \
  -p 8080:8080 \
  orchestrator-agent:latest

⁠With trajectory recording

docker run --rm \
  -e TRAJECTORY_DIR=/trajectories \
  -v /tmp/orch-trajectories:/trajectories \
  -p 8080:8080 \
  orchestrator-agent:latest

⁠With agent secrets

docker run --rm \
  -e SECRET_SECRETS_GMAIL_PASSWORD=hunter2 \
  -e SECRET_SECRETS_CHROME_AGENT_TOKEN=tok-xyz \
  -p 8080:8080 \
  orchestrator-agent:latest

⁠Persistent database

docker run --rm \
  -e DB_URL=sqlite+aiosqlite:////data/orchestrator.db \
  -v /tmp/orch-data:/data \
  -p 8080:8080 \
  orchestrator-agent:latest

⁠API

⁠A2A inbound interface

The orchestrator exposes a standard A2A endpoint at POST /. Upstream agents submit plans as A2A tasks.

Agent card: GET /.well-known/agent.json

Health check: GET /health → {"status": "ok"} (no auth required)

⁠Submit a plan

curl -s http://localhost:8080/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "id": "req-001",
    "params": {
      "message": {
        "role": "user",
        "messageId": "msg-001",
        "parts": [{"type": "text", "text": "<plan JSON here>"}]
      }
    }
  }'

On success the task artifact contains:

{"run_id": "uuid", "status": "running"}

On validation failure the task is marked failed and the artifact contains:

{
  "valid": false,
  "errors": [
    {"code": "CYCLE_DETECTED", "message": "...", "location": "step_graph.edges", "severity": "error"}
  ]
}

⁠Validate only (no execution)

Set execution_handoff.mode to plan_only in the plan. The orchestrator returns validation results without creating a run.

⁠Plan contract (summary)

A minimal valid plan:

{
  "plan_id": "550e8400-e29b-41d4-a716-446655440000",
  "summary": "Example plan",
  "original_request": "Find the latest invoice in Gmail",
  "assumptions": [],
  "pipeline_stages": ["research", "process"],
  "required_agents": [
    {
      "name": "web.browser.chrome",
      "url": "https://chrome-agent:8080",
      "role": "research",
      "price_per_turn": 0.0,
      "auth": {"type": "none"}
    }
  ],
  "step_graph": {
    "is_dag": true,
    "entry_node_ids": ["S1"],
    "exit_node_ids": ["S1"],
    "nodes": [
      {
        "id": "S1",
        "title": "Find invoice",
        "stage": "research",
        "assigned_agent_name": "web.browser.chrome",
        "inputs": {
          "request_fragment": "Open Gmail and return the filename of the latest invoice attachment.",
          "context": {},
          "secret_refs": {}
        },
        "outputs": ["attachment_filename"],
        "risk_level": "high",
        "execution_policy": {
          "timeout_seconds": 120,
          "max_retries": 1,
          "on_failure": "fail_run"
        }
      }
    ],
    "edges": []
  },
  "execution_handoff": {
    "mode": "execute",
    "ready_for_orchestrator": true
  }
}

⁠Top-level plan fields

⁠execution_policy.on_failure values

Replans are capped at MAX_REPLANS per run (default 3). Mid-run amendments sent via A2A do not count against this cap.

⁠Mid-run amendment

While a run is executing, send an amended plan JSON to the running A2A task to substitute the plan in-flight:

# Send amended plan to a running task (replace <task_id> and <plan_json>)
curl -s https://orchestrator:8080/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "id": "amend-001",
    "params": {
      "message": {
        "role": "user",
        "taskId": "<task_id>",
        "messageId": "msg-amend-001",
        "parts": [{"type": "text", "text": "<amended plan JSON>"}]
      }
    }
  }'

The engine drains the inbox before each step. If the message parses as a valid Plan, it substitutes the running plan and restarts execution from the beginning of the amended plan. Steps already completed are not re-run — only remaining steps are executed.

⁠Pause and resume

Send a message with command: "pause" to suspend execution between steps:

{"parts": [{"type": "text", "text": "{\"command\": \"pause\"}"}]}

The engine completes the current step, then blocks until the next message arrives. The resume message may optionally carry an amended plan JSON to inject before resuming.

See INSTRUCTIONS.md for the full plan contract specification.

⁠Development

⁠Install dependencies

uv sync --extra server --extra dev

⁠Run tests

uv run pytest tests/ -v

⁠Run locally (without Docker)

uv run python -m app.server

The server starts on port 8080 with a self-signed TLS certificate at /tmp/agent.crt / /tmp/agent.key.

⁠MCP tools

The agent image exposes an MCP (Model Context Protocol) server on the same port as A2A at /mcp. Any MCP client — Claude Desktop, Cursor, Windsurf, ChatGPT Connectors, OpenAI Agents SDK — can list and invoke these tools using the pod's API_KEY as Bearer.

In production, Core proxies https://orchestrator-agent.agents.forfetch.ai/mcp → the worker pod's /mcp.

Input schemas are authored in apps/agents_mcp/app/seed.py. Skills without an explicit schema advertise a single {task: str} freeform parameter.