Production Observability with Swarms

import os
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

def fetch_logs():
    resp = requests.get(f"{BASE_URL}/v1/account/logs", headers=headers, timeout=30)
    resp.raise_for_status()
    return resp.json()


def last_24h_breakdown():
    data = fetch_logs()
    cutoff = datetime.now(timezone.utc) - timedelta(hours=24)

    by_type = defaultdict(lambda: {"count": 0, "errors": 0, "cost": 0.0, "tokens": 0})
    untyped = {"count": 0, "errors": 0, "cost": 0.0, "tokens": 0}

    for log in data.get("logs", []):
        ts_raw = log.get("timestamp")
        if not ts_raw:
            continue
        ts = datetime.fromisoformat(ts_raw.replace("Z", "+00:00"))
        if ts < cutoff:
            continue

        # swarm_type lives inside the request payload echoed in the log
        swarm_type = (
            log.get("data", {}).get("swarm_type")
            or log.get("swarm_type")
            or None
        )
        bucket = by_type[swarm_type] if swarm_type else untyped

        bucket["count"] += 1
        bucket["cost"] += float(log.get("cost", 0) or 0)
        bucket["tokens"] += int(log.get("tokens_used", 0) or 0)
        if (log.get("status_code") or 200) >= 400:
            bucket["errors"] += 1

    print("Last 24 hours, by swarm_type")
    print("-" * 70)
    print(f"{'swarm_type':<22} {'reqs':>6} {'errs':>6} {'tokens':>10} {'cost':>10}")
    print("-" * 70)
    for swarm_type, b in sorted(by_type.items(), key=lambda kv: kv[1]["cost"], reverse=True):
        print(
            f"{(swarm_type or '-'):<22} "
            f"{b['count']:>6} {b['errors']:>6} "
            f"{b['tokens']:>10,} ${b['cost']:>8.4f}"
        )
    if untyped["count"]:
        print(
            f"{'(single-agent)':<22} "
            f"{untyped['count']:>6} {untyped['errors']:>6} "
            f"{untyped['tokens']:>10,} ${untyped['cost']:>8.4f}"
        )

    total_cost = sum(b["cost"] for b in by_type.values()) + untyped["cost"]
    total_err = sum(b["errors"] for b in by_type.values()) + untyped["errors"]
    total_req = sum(b["count"] for b in by_type.values()) + untyped["count"]
    print("-" * 70)
    print(f"Total: {total_req} requests, {total_err} errors, ${total_cost:.4f}")


if __name__ == "__main__":
    last_24h_breakdown()

def fetch_current_pricing() -> dict:
    resp = requests.get(f"{BASE_URL}/v1/usage/costs", headers=headers, timeout=30)
    resp.raise_for_status()
    return resp.json()["usage_pricing"]


def reconcile_today():
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    # Logs side: sum all of today's entries (the only durable record of spend)
    logs = fetch_logs().get("logs", [])
    logs_cost = sum(
        float(log.get("cost", 0) or 0)
        for log in logs
        if (log.get("timestamp") or "").startswith(today)
    )

    # Rates side: pull the live pricing rate card
    pricing = fetch_current_pricing()

    print(f"Logs total ({today}):       ${logs_cost:.4f}")
    print(f"Swarm input rate (live):    ${pricing['swarm_completions_input_cost_per_1m']}/1M tokens")
    print(f"Swarm output rate (live):   ${pricing['swarm_completions_output_cost_per_1m']}/1M tokens")
    print(f"Per-agent rate (live):      ${pricing['swarm_completions_agent_cost']}/agent")
    print(
        "Sanity-check: if the per-request costs in your logs don't line up "
        "with what these rates would produce, either the rate card changed "
        "mid-day or your client-side cost estimate is stale."
    )

def credits_remaining() -> float:
    resp = requests.get(f"{BASE_URL}/v1/account/credits", headers=headers, timeout=10)
    resp.raise_for_status()
    return float(resp.json().get("total_credits", 0))


def guard_batch(estimated_cost: float, safety_margin: float = 1.5):
    """Refuse to start a batch if credits < estimated_cost * safety_margin."""
    available = credits_remaining()
    required = estimated_cost * safety_margin
    if available < required:
        raise RuntimeError(
            f"Insufficient credits: have ${available:.2f}, "
            f"need ${required:.2f} (estimated ${estimated_cost:.2f} x {safety_margin})."
        )
    print(f"Credit check OK: ${available:.2f} available, ${required:.2f} required.")

def call_with_headers_logged(payload: dict):
    resp = requests.post(
        f"{BASE_URL}/v1/swarm/completions",
        headers=headers,
        json=payload,
        timeout=600,
    )

    rl = {k: v for k, v in resp.headers.items() if k.lower().startswith("x-ratelimit")}
    print(
        f"tier={rl.get('X-RateLimit-Tier')} "
        f"min={rl.get('X-RateLimit-Remaining-Minute')}/{rl.get('X-RateLimit-Limit-Minute')} "
        f"day={rl.get('X-RateLimit-Remaining-Day')}/{rl.get('X-RateLimit-Limit-Day')}"
    )

    if resp.status_code == 429:
        retry_after = int(resp.headers.get("Retry-After", "60"))
        print(f"Rate limited. Retry after {retry_after}s.")
        return None

    resp.raise_for_status()
    return resp.json()