Building Real-Time Dashboards for AI Agent Fleets

Running an AI agent fleet without observability is like flying blind. When your agents are managing real funds across Purple Flea's casino, trading, wallet, and escrow services, you need live visibility into what's happening — not a CSV export from yesterday.

This guide covers the full stack of real-time dashboard options for agent operators: from quick terminal views you can spin up in minutes, to production-grade Grafana setups with alerting. We'll build a live WebSocket P&L chart, a Python rich terminal dashboard, and a Telegram alert bot — all using Purple Flea's API.

1. Why Dashboards Matter for Agent Operators

Before building anything, it's worth being clear about what observability actually buys you. AI agents can make decisions at machine speed — a poorly configured agent can open positions, place bets, or initiate escrow contracts faster than any human can catch. Without real-time monitoring:

• Silent failures accumulate — an agent hitting API errors might stop operating for hours before you notice
• Risk exposure grows unchecked — position sizes or bet sequences can compound beyond intended limits
• Referral performance is invisible — you can't optimize what you can't see
• Anomalies go undetected — unusual behavior patterns that might indicate bugs or edge cases

A good dashboard makes the difference between being a passive agent operator and an active one who can respond to events in seconds.

2. Tech Stack Choices

There are four main approaches to agent dashboards, each with different trade-offs:

3. Key Metrics to Display

Before building any dashboard, decide what metrics matter. Here are the essential ones for Purple Flea agent operators:

Additional metrics worth tracking for advanced operators:

• Sharpe ratio (rolling 7-day) — risk-adjusted performance
• Max drawdown — worst peak-to-trough loss in the period
• Casino win rate — actual vs expected win rate (detects variance from expected)
• API latency (p50/p95/p99) — ensures agent is operating at full speed
• Escrow dispute rate — should trend toward zero for quality agents
• Referral conversion funnel — registrations → first transaction rate

4. Purple Flea API Polling Patterns

Efficient data fetching is critical for dashboards. Over-polling wastes API quota and can trigger rate limits. Under-polling gives stale data. Here's a recommended polling strategy:

"""
Purple Flea API Polling Manager
Implements tiered polling with caching and rate limit awareness.
"""

import asyncio
import time
from collections import defaultdict
import httpx

API_KEY = "pf_live_your_key_here"
BASE = "https://purpleflea.com/api/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# Polling intervals in seconds by data type
INTERVALS = {
    "balance":    10,   # Wallet balance — low change rate
    "positions":  5,    # Active positions — moderate change rate
    "pnl":        3,    # P&L — want fairly fresh
    "casino":     2,    # Casino bets — fast-moving
    "referrals":  30,   # Referral count — slow-moving
    "errors":     5,    # Error log — important to catch fast
}

class PollingManager:
    def __init__(self):
        self.cache = {}
        self.last_fetch = defaultdict(float)
        self.client = httpx.AsyncClient(headers=HEADERS, timeout=10)

    async def fetch(self, key, url):
        now = time.time()
        if now - self.last_fetch[key] < INTERVALS.get(key, 10):
            return self.cache.get(key)  # Return cached value

        try:
            r = await self.client.get(f"{BASE}/{url}")
            if r.status_code == 429:
                print(f"Rate limited on {key} — backing off 30s")
                INTERVALS[key] = min(INTERVALS[key] * 2, 120)
                return self.cache.get(key)

            if r.status_code == 200:
                # On success, restore normal interval
                INTERVALS[key] = max(INTERVALS[key] // 2,
                                     {"balance":10,"positions":5,"pnl":3,
                                      "casino":2,"referrals":30,"errors":5}.get(key, 10))
                self.cache[key] = r.json()
                self.last_fetch[key] = now
                return self.cache[key]

        except Exception as e:
            print(f"Fetch error for {key}: {e}")
            return self.cache.get(key)

    async def get_all(self):
        tasks = {
            "balance":   self.fetch("balance", "wallet/balance"),
            "positions": self.fetch("positions", "trading/positions"),
            "pnl":       self.fetch("pnl", "portfolio/pnl"),
            "casino":    self.fetch("casino", "casino/recent"),
            "referrals": self.fetch("referrals", "referrals/stats"),
        }
        results = await asyncio.gather(*tasks.values(), return_exceptions=True)
        return dict(zip(tasks.keys(), results))

    async def close(self):
        await self.client.aclose()

5. WebSocket-Based Live P&L Chart

For true real-time updates, a WebSocket connection is far more efficient than polling. Here's a complete implementation: a Python FastAPI backend that streams P&L updates, and a self-contained HTML/JS frontend that renders a live chart.

FastAPI Backend (server.py)

"""
Purple Flea Live P&L Dashboard Backend
FastAPI + WebSocket — streams real-time agent data to browser clients.
"""

import asyncio
import json
import time
from typing import List
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import httpx

app = FastAPI(title="PF Agent Dashboard")

API_KEY = "pf_live_your_key_here"
BASE = "https://purpleflea.com/api/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# In-memory store of connected dashboard clients
connected_clients: List[WebSocket] = []

# Historical P&L series (rolling 200 points)
pnl_history = []


async def broadcast(message: dict):
    """Send message to all connected dashboard clients."""
    payload = json.dumps(message)
    disconnected = []
    for ws in connected_clients:
        try:
            await ws.send_text(payload)
        except Exception:
            disconnected.append(ws)
    for ws in disconnected:
        connected_clients.remove(ws)


async def fetch_agent_snapshot():
    """Fetch current agent state from Purple Flea API."""
    async with httpx.AsyncClient(headers=HEADERS, timeout=8) as client:
        results = await asyncio.gather(
            client.get(f"{BASE}/wallet/balance"),
            client.get(f"{BASE}/portfolio/pnl"),
            client.get(f"{BASE}/trading/positions"),
            client.get(f"{BASE}/referrals/stats"),
            return_exceptions=True
        )

    snapshot = {"timestamp": int(time.time() * 1000)}
    labels = ["balance", "pnl", "positions", "referrals"]

    for label, result in zip(labels, results):
        if isinstance(result, Exception):
            snapshot[label] = None
        elif result.status_code == 200:
            snapshot[label] = result.json()
        else:
            snapshot[label] = None

    return snapshot


async def polling_loop():
    """Background task: fetch data every 2s, broadcast to clients."""
    while True:
        try:
            snapshot = await fetch_agent_snapshot()

            # Track P&L history
            pnl_val = None
            if snapshot.get("pnl"):
                pnl_val = snapshot["pnl"].get("total_usd", 0)
            pnl_history.append({
                "t": snapshot["timestamp"],
                "v": pnl_val
            })
            if len(pnl_history) > 200:
                pnl_history.pop(0)

            snapshot["pnl_history"] = pnl_history[-60:]  # last 60 points to chart

            if connected_clients:
                await broadcast({"type": "snapshot", "data": snapshot})

        except Exception as e:
            print(f"Polling loop error: {e}")

        await asyncio.sleep(2)


@app.on_event("startup")
async def startup():
    asyncio.create_task(polling_loop())


@app.websocket("/ws")
async def dashboard_ws(websocket: WebSocket):
    await websocket.accept()
    connected_clients.append(websocket)
    try:
        # Send current history immediately on connect
        await websocket.send_text(json.dumps({
            "type": "init",
            "pnl_history": pnl_history[-60:]
        }))
        while True:
            await websocket.receive_text()  # Keep connection alive
    except WebSocketDisconnect:
        connected_clients.remove(websocket)


@app.get("/", response_class=HTMLResponse)
async def dashboard_html():
    return DASHBOARD_HTML  # See inline HTML below

Browser Dashboard HTML (inline)

<!-- Paste this as DASHBOARD_HTML in the FastAPI server -->
<!DOCTYPE html>
<html>
<head>
  <title>Purple Flea Agent Dashboard</title>
  <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
  <style>
    body { background:#09090B; color:#FAFAFA; font-family:Inter,sans-serif; padding:24px; }
    .metrics { display:grid; grid-template-columns:repeat(4,1fr); gap:16px; margin-bottom:24px; }
    .metric { background:rgba(255,255,255,0.04); border:1px solid rgba(255,255,255,0.06);
              border-radius:12px; padding:16px; }
    .metric label { font-size:11px; text-transform:uppercase; letter-spacing:0.08em;
                    color:#71717A; display:block; margin-bottom:6px; }
    .metric span { font-size:24px; font-weight:700; font-family:monospace; }
    .pos { color:#22C55E; } .neg { color:#EF4444; }
    canvas { max-height:300px; }
  </style>
</head>
<body>
  <h2 style="margin-bottom:20px;letter-spacing:-0.02em;">
    🟣 Live Agent Dashboard
  </h2>
  <div class="metrics">
    <div class="metric"><label>Balance</label><span id="balance">--</span></div>
    <div class="metric"><label>P&L Today</label><span id="pnl">--</span></div>
    <div class="metric"><label>Positions</label><span id="positions">--</span></div>
    <div class="metric"><label>Referrals</label><span id="referrals">--</span></div>
  </div>
  <canvas id="pnlChart"></canvas>
  <script>
    const ctx = document.getElementById('pnlChart').getContext('2d');
    const chart = new Chart(ctx, {
      type: 'line',
      data: { labels: [], datasets: [{ label: 'P&L (USD)', data: [],
        borderColor: '#A855F7', backgroundColor: 'rgba(168,85,247,0.1)',
        tension: 0.4, fill: true, pointRadius: 0 }] },
      options: { animation: false, responsive: true,
        scales: { x: { display: false }, y: { grid: { color: 'rgba(255,255,255,0.05)' } } } }
    });

    const ws = new WebSocket(`ws://${location.host}/ws`);
    ws.onmessage = (e) => {
      const msg = JSON.parse(e.data);
      if (msg.pnl_history) {
        chart.data.labels = msg.pnl_history.map(p => new Date(p.t).toLocaleTimeString());
        chart.data.datasets[0].data = msg.pnl_history.map(p => p.v);
        chart.update('none');
      }
      if (msg.data) {
        const d = msg.data;
        if (d.balance) document.getElementById('balance').textContent =
          '$' + (d.balance.usd || 0).toFixed(2);
        if (d.pnl) {
          const pnl = d.pnl.total_usd || 0;
          const el = document.getElementById('pnl');
          el.textContent = (pnl >= 0 ? '+' : '') + '$' + pnl.toFixed(2);
          el.className = pnl >= 0 ? 'pos' : 'neg';
        }
        if (d.positions) document.getElementById('positions').textContent =
          d.positions.count || 0;
        if (d.referrals) document.getElementById('referrals').textContent =
          d.referrals.total || 0;
      }
    };
  </script>
</body>
</html>

6. Terminal Dashboard Using Python rich

For operators who live in the terminal or need to monitor agents over SSH, Python's rich library provides a surprisingly capable dashboard with zero browser dependencies.

"""
Purple Flea Terminal Dashboard
Uses Python rich library for live terminal monitoring.
Install: pip install rich httpx
"""

import asyncio
import time
from rich.console import Console
from rich.table import Table
from rich.layout import Layout
from rich.panel import Panel
from rich.live import Live
from rich.text import Text
from rich.columns import Columns
from rich import box
import httpx

API_KEY = "pf_live_your_key_here"
BASE = "https://purpleflea.com/api/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

console = Console()

# Rolling P&L history for sparkline
pnl_series = []
MAX_SPARK = 40


def sparkline(values, width=40):
    """Simple ASCII sparkline."""
    if not values or len(values) < 2:
        return "─" * width
    mn, mx = min(values), max(values)
    if mx == mn:
        return "─" * width
    chars = "▁▂▃▄▅▆▇█"
    normalized = [(v - mn) / (mx - mn) for v in values[-width:]]
    return "".join(chars[int(n * (len(chars) - 1))] for n in normalized)


def make_layout(data):
    layout = Layout()
    layout.split_column(
        Layout(name="header", size=3),
        Layout(name="metrics", size=8),
        Layout(name="positions"),
        Layout(name="footer", size=3)
    )

    # Header
    layout["header"].update(Panel(
        Text("🟣 Purple Flea Agent Dashboard  " +
             time.strftime("%Y-%m-%d %H:%M:%S"), style="bold white"),
        style="on #09090B"
    ))

    # Metrics row
    bal = data.get("balance", {}) or {}
    pnl = data.get("pnl", {}) or {}
    ref = data.get("referrals", {}) or {}
    casino = data.get("casino", {}) or {}

    pnl_val = pnl.get("total_usd", 0) or 0
    pnl_series.append(pnl_val)
    if len(pnl_series) > MAX_SPARK:
        pnl_series.pop(0)

    pnl_color = "green" if pnl_val >= 0 else "red"
    pnl_str = f"{'+'if pnl_val>=0 else ''}{pnl_val:.2f}"

    metrics = Columns([
        Panel(
            f"[bold]{bal.get('usd', '--'):.2f}[/bold] USDC\n"
            f"[dim]{bal.get('chain', 'TRON')}[/dim]",
            title="[cyan]Wallet Balance[/cyan]", width=22
        ),
        Panel(
            f"[bold {pnl_color}]{pnl_str}[/bold {pnl_color}] USD\n"
            f"[dim]{sparkline(pnl_series, 18)}[/dim]",
            title="[cyan]P&L Today[/cyan]", width=24
        ),
        Panel(
            f"[bold]{ref.get('total', 0)}[/bold] total\n"
            f"[green]+{ref.get('today', 0)} today[/green]",
            title="[cyan]Referrals[/cyan]", width=22
        ),
        Panel(
            f"[bold]{casino.get('win_rate', 0)*100:.1f}%[/bold] win rate\n"
            f"[dim]{casino.get('bets_today', 0)} bets today[/dim]",
            title="[cyan]Casino[/cyan]", width=22
        ),
    ], equal=True, expand=True)

    layout["metrics"].update(metrics)

    # Positions table
    positions = data.get("positions", {})
    pos_list = positions.get("items", []) if positions else []

    tbl = Table(box=box.SIMPLE, show_header=True, header_style="bold purple",
                border_style="dim white", expand=True)
    tbl.add_column("Pair", style="white", width=12)
    tbl.add_column("Side", width=8)
    tbl.add_column("Size", justify="right", width=12)
    tbl.add_column("Entry", justify="right", width=12)
    tbl.add_column("Current", justify="right", width=12)
    tbl.add_column("P&L", justify="right", width=12)

    for pos in pos_list[:10]:
        pnl_p = pos.get("pnl_usd", 0) or 0
        side_color = "green" if pos.get("side") == "long" else "red"
        tbl.add_row(
            pos.get("pair", "--"),
            f"[{side_color}]{pos.get('side', '--')}[/{side_color}]",
            f"{pos.get('size', 0):.4f}",
            f"${pos.get('entry_price', 0):.4f}",
            f"${pos.get('current_price', 0):.4f}",
            f"[{'green' if pnl_p>=0 else 'red'}]{'+'if pnl_p>=0 else ''}{pnl_p:.2f}[/]"
        )

    if not pos_list:
        tbl.add_row("[dim]No open positions[/dim]", "", "", "", "", "")

    layout["positions"].update(Panel(tbl, title="[bold]Open Positions[/bold]"))
    layout["footer"].update(Panel(
        "[dim]Refresh: 3s  |  pf_live_ key active  |  q to quit[/dim]"
    ))

    return layout


async def fetch_all():
    async with httpx.AsyncClient(headers=HEADERS, timeout=6) as client:
        results = await asyncio.gather(
            client.get(f"{BASE}/wallet/balance"),
            client.get(f"{BASE}/portfolio/pnl"),
            client.get(f"{BASE}/trading/positions"),
            client.get(f"{BASE}/referrals/stats"),
            client.get(f"{BASE}/casino/stats"),
            return_exceptions=True
        )

    labels = ["balance", "pnl", "positions", "referrals", "casino"]
    data = {}
    for label, r in zip(labels, results):
        if isinstance(r, Exception) or r.status_code != 200:
            data[label] = {}
        else:
            data[label] = r.json()
    return data


async def run_dashboard():
    with Live(console=console, refresh_per_second=4, screen=True) as live:
        while True:
            try:
                data = await fetch_all()
                live.update(make_layout(data))
            except Exception as e:
                live.update(Panel(f"[red]Error: {e}[/red]"))
            await asyncio.sleep(3)


if __name__ == "__main__":
    asyncio.run(run_dashboard())

7. Mobile Alerts — Telegram Bot for Critical Events

Dashboards are great when you're watching them. Telegram alerts are what catch problems when you're not. Here's a lightweight alert bot that monitors your agents and sends messages on critical events:

"""
Purple Flea Agent Alert Bot
Sends Telegram messages on critical agent events.
Install: pip install httpx
"""

import asyncio
import httpx

TELEGRAM_BOT_TOKEN = "your_bot_token_here"  # From @BotFather
TELEGRAM_CHAT_ID   = "your_chat_id_here"    # From @userinfobot

PF_API_KEY = "pf_live_your_key_here"
PF_BASE    = "https://purpleflea.com/api/v1"
PF_HEADERS = {"Authorization": f"Bearer {PF_API_KEY}"}

# Alert thresholds
THRESHOLDS = {
    "max_drawdown_pct":   10.0,   # Alert if drawdown exceeds 10%
    "balance_low_usd":    50.0,   # Alert if balance drops below $50
    "error_rate_pct":     5.0,    # Alert if error rate exceeds 5%
    "position_count_max": 20,     # Alert if more than 20 open positions
}

# State: track previously alerted conditions to avoid spam
alerted = set()


async def send_alert(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    async with httpx.AsyncClient() as client:
        await client.post(url, json={
            "chat_id": TELEGRAM_CHAT_ID,
            "text": message,
            "parse_mode": "HTML"
        })


async def check_and_alert():
    async with httpx.AsyncClient(headers=PF_HEADERS, timeout=10) as client:
        balance_r   = await client.get(f"{PF_BASE}/wallet/balance")
        pnl_r       = await client.get(f"{PF_BASE}/portfolio/pnl")
        positions_r = await client.get(f"{PF_BASE}/trading/positions")
        errors_r    = await client.get(f"{PF_BASE}/agent/error-stats")

    if balance_r.status_code == 200:
        bal = balance_r.json().get("usd", 999)
        if bal < THRESHOLDS["balance_low_usd"] and "low_balance" not in alerted:
            await send_alert(
                f"🔴 Low Balance Alert\n"
                f"Agent balance is ${bal:.2f} — "
                f"below threshold of ${THRESHOLDS['balance_low_usd']:.2f}\n"
                f"Fund your wallet at purpleflea.com/wallet"
            )
            alerted.add("low_balance")
        elif bal >= THRESHOLDS["balance_low_usd"] * 1.5:
            alerted.discard("low_balance")  # Reset alert when recovered

    if pnl_r.status_code == 200:
        dd = pnl_r.json().get("max_drawdown_pct", 0)
        if dd > THRESHOLDS["max_drawdown_pct"] and "drawdown" not in alerted:
            await send_alert(
                f"⚠️ Drawdown Alert\n"
                f"Max drawdown reached {dd:.1f}% — "
                f"threshold is {THRESHOLDS['max_drawdown_pct']}%\n"
                f"Review your risk settings."
            )
            alerted.add("drawdown")
        elif dd < THRESHOLDS["max_drawdown_pct"] * 0.5:
            alerted.discard("drawdown")

    if positions_r.status_code == 200:
        count = positions_r.json().get("count", 0)
        if count > THRESHOLDS["position_count_max"] and "pos_count" not in alerted:
            await send_alert(
                f"⚠️ Position Count Alert\n"
                f"Agent has {count} open positions — "
                f"limit is {THRESHOLDS['position_count_max']}"
            )
            alerted.add("pos_count")


async def run_alert_loop():
    print("Purple Flea alert bot started.")
    await send_alert("🟣 Purple Flea Alert Bot Online\nMonitoring your agent fleet.")
    while True:
        try:
            await check_and_alert()
        except Exception as e:
            print(f"Alert check error: {e}")
        await asyncio.sleep(30)  # Check every 30 seconds


if __name__ == "__main__":
    asyncio.run(run_alert_loop())

8. Multi-Agent Aggregation — Fleet View

When you operate multiple agents, you need a fleet-level view that aggregates across all of them. The key challenge is normalization — each agent may have different strategies, risk profiles, and starting balances, so raw P&L aggregation is misleading.

"""
Purple Flea Multi-Agent Fleet Aggregator
Aggregates metrics across multiple agent API keys.
"""

import asyncio
import httpx
from dataclasses import dataclass, field
from typing import List, Optional

BASE = "https://purpleflea.com/api/v1"


@dataclass
class AgentConfig:
    agent_id: str
    api_key: str
    label: str
    strategy: str
    initial_balance: float


@dataclass
class AgentSnapshot:
    agent_id: str
    label: str
    balance_usd: float = 0.0
    pnl_usd: float = 0.0
    pnl_pct: float = 0.0
    positions: int = 0
    error_rate: float = 0.0
    referral_count: int = 0
    status: str = "unknown"


async def fetch_agent_snapshot(config: AgentConfig) -> AgentSnapshot:
    headers = {"Authorization": f"Bearer {config.api_key}"}
    snap = AgentSnapshot(agent_id=config.agent_id, label=config.label)

    async with httpx.AsyncClient(headers=headers, timeout=8) as client:
        try:
            bal_r = await client.get(f"{BASE}/wallet/balance")
            pnl_r = await client.get(f"{BASE}/portfolio/pnl")
            pos_r = await client.get(f"{BASE}/trading/positions")
            ref_r = await client.get(f"{BASE}/referrals/stats")

            if bal_r.status_code == 200:
                snap.balance_usd = bal_r.json().get("usd", 0)

            if pnl_r.status_code == 200:
                snap.pnl_usd = pnl_r.json().get("total_usd", 0)
                snap.pnl_pct = (snap.pnl_usd / config.initial_balance * 100
                                if config.initial_balance > 0 else 0)

            if pos_r.status_code == 200:
                snap.positions = pos_r.json().get("count", 0)

            if ref_r.status_code == 200:
                snap.referral_count = ref_r.json().get("total", 0)

            snap.status = "healthy"

        except Exception as e:
            snap.status = f"error: {e}"

    return snap


async def fetch_fleet(agents: List[AgentConfig]) -> List[AgentSnapshot]:
    snapshots = await asyncio.gather(
        *[fetch_agent_snapshot(a) for a in agents],
        return_exceptions=True
    )
    return [s for s in snapshots if isinstance(s, AgentSnapshot)]


def print_fleet_summary(snapshots: List[AgentSnapshot]):
    total_balance = sum(s.balance_usd for s in snapshots)
    total_pnl = sum(s.pnl_usd for s in snapshots)
    total_positions = sum(s.positions for s in snapshots)
    total_referrals = sum(s.referral_count for s in snapshots)

    print(f"\n{'='*60}")
    print(f"  Purple Flea Fleet Summary — {len(snapshots)} agents")
    print(f"{'='*60}")
    print(f"  Total Balance:   ${total_balance:,.2f}")
    print(f"  Total P&L:       ${total_pnl:+,.2f}")
    print(f"  Total Positions: {total_positions}")
    print(f"  Total Referrals: {total_referrals}")
    print(f"{'='*60}")
    print(f"  {'Agent':<20} {'Balance':>10} {'P&L':>10} {'PnL%':>8} {'Status'}")
    print(f"  {'-'*20} {'-'*10} {'-'*10} {'-'*8} {'-'*10}")

    for s in sorted(snapshots, key=lambda x: x.pnl_usd, reverse=True):
        pnl_sign = "+" if s.pnl_usd >= 0 else ""
        print(f"  {s.label:<20} ${s.balance_usd:>9,.2f} "
              f"{pnl_sign}${s.pnl_usd:>8,.2f} "
              f"{s.pnl_pct:>+7.1f}% "
              f"{s.status}")
    print()


# Example usage:
FLEET = [
    AgentConfig("agent-001", "pf_live_key_001", "Momentum-Alpha", "momentum", 500.0),
    AgentConfig("agent-002", "pf_live_key_002", "MeanRev-Beta",   "mean-rev", 500.0),
    AgentConfig("agent-003", "pf_live_key_003", "Arb-Gamma",      "arbitrage", 1000.0),
]

if __name__ == "__main__":
    snapshots = asyncio.run(fetch_fleet(FLEET))
    print_fleet_summary(snapshots)

9. Dashboard as a Service — Monetizing Your Monitoring

An interesting business model enabled by agent dashboards: Dashboard as a Service (DaaS). If you operate a sophisticated monitoring stack for your own agents, you can sell access to other agent operators who lack the technical infrastructure to build their own.

The DaaS Architecture

The pattern is straightforward:

1. Build a multi-tenant dashboard — your WebSocket server accepts multiple client API keys
2. Authenticate clients — each client gets a dashboard access token scoped to their data
3. Charge for access — use Purple Flea's escrow for subscription payments: client locks $10/month in escrow, your service delivers monitoring, escrow auto-releases on schedule
4. Earn referral income — if clients register via your Purple Flea referral link, you earn 15% of their escrow fees perpetually

Further Reading

• Error Handling Patterns for Production Agents
• Bankroll Management — the Core Risk Dashboard Metric
• Purple Flea API Rate Limiting Guide
• Purple Flea API Reference