I Built a JARVIS Dashboard for My AI Agent Fleet — Here's the Code

#ai #python #flask #opensource

When you're running 5 AI agents on scheduled launchd jobs, you need visibility without grepping log files.

So we built atlas-ops — a lightweight Flask dashboard that shows every agent's status, live output via SSE, launchd health, and session transcripts. Terminal aesthetic. Runs on localhost. Zero external dependencies beyond Flask.

Repo: github.com/Wh0FF24/whoff-automation → atlas-ops/

What It Does

Agent monitor — live view of every Claude Code subagent transcript
Session feed — SSE stream of vault file writes
launchd status — green/red for each scheduled job
Skill panel — 46 Claude Code skills, one-click trigger
Command input — streams claude -p output live

Architecture

atlas-ops/
├── app.py           # Flask backend (~400 lines)
├── templates/
│   └── index.html   # Single-page dashboard
└── static/
    └── style.css    # Terminal aesthetic

No webpack. No npm. No build step.

How It Finds Running Agents

Claude Code stores task output files in /private/tmp/claude-{uid}/. We watch those directly — no instrumentation required in the agents:

TASKS_ROOT = Path("/private/tmp/claude-501/-Users-you-projects-your-project")
RUNNING_THRESHOLD_SEC = 90

def _iter_agent_files():
    if not TASKS_ROOT.exists():
        return
    for session_dir in TASKS_ROOT.iterdir():
        tasks_dir = session_dir / "tasks"
        if not tasks_dir.is_dir():
            continue
        for p in tasks_dir.iterdir():
            if p.name.endswith(".output"):
                yield p

def _agent_summary(path: Path) -> dict | None:
    stat = path.stat()
    age = time.time() - stat.st_mtime
    is_running = age < RUNNING_THRESHOLD_SEC and stat.st_size > 0
    prompt = ""
    try:
        with path.open() as f:
            for line in f:
                event = json.loads(line)
                if event.get("type") == "user" and event.get("message"):
                    prompt = event["message"]["content"][:120]
                    break
    except Exception:
        pass
    return {"id": path.stem, "running": is_running, "age_sec": int(age), "prompt": prompt}

SSE Live Feed

@app.route("/api/stream")
def stream():
    def event_generator():
        seen = set()
        vault = Path("~/Desktop/Agents").expanduser()
        while True:
            for f in vault.rglob("*.md"):
                key = (str(f), f.stat().st_mtime)
                if key not in seen:
                    seen.add(key)
                    payload = json.dumps({
                        "file": f.name, "agent": f.parent.name,
                        "size": f.stat().st_size, "ts": datetime.now().isoformat()
                    })
                    yield f"data: {payload}\n\n"
            time.sleep(2)
    return Response(event_generator(), mimetype="text/event-stream",
                    headers={"Cache-Control": "no-cache"})

const es = new EventSource("/api/stream");
es.onmessage = (e) => {
    const ev = JSON.parse(e.data);
    appendToFeed(`[${ev.agent}] ${ev.file} — ${ev.size} bytes`);
};

launchd Health

@app.route("/api/launchd")
def launchd_status():
    labels = ["com.whoff.agent.midnight", "com.whoff.agent.morning",
              "com.whoff.agent.afternoon", "com.whoff.agent.night"]
    results = {}
    for label in labels:
        proc = subprocess.run(["launchctl", "list", label], capture_output=True, text=True)
        if proc.returncode == 0:
            has_pid = any("PID" in line for line in proc.stdout.splitlines())
            results[label] = {"status": "running" if has_pid else "idle"}
        else:
            results[label] = {"status": "not_loaded"}
    return jsonify(results)

Streaming Skill Runner

@app.route("/api/skill/run", methods=["POST"])
def run_skill():
    skill = request.json.get("skill", "")
    def stream_output():
        proc = subprocess.Popen(["claude", "-p", f"/{skill}"],
                                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                text=True, cwd=str(REPO_DIR))
        for line in proc.stdout:
            yield f"data: {json.dumps({'line': line})}\n\n"
        yield f"data: {json.dumps({'done': True})}\n\n"
    return Response(stream_output(), mimetype="text/event-stream")

Running It

git clone https://github.com/Wh0FF24/whoff-automation.git
cd whoff-automation
pip install flask
python atlas-ops/app.py
# → http://localhost:4100/atlas

No config. Auto-discovers agents from the tmp path and vault.