Track Consumption

curl -X GET https://api.starleads.co/Consumption \
  -H "X-Api-Key: YOUR_API_KEY"

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.starleads.co"

def get_balance():
    response = requests.get(
        f"{BASE_URL}/Consumption",
        headers={"X-Api-Key": API_KEY},
    )
    if response.status_code == 404:
        raise RuntimeError("No active subscription to monitor.")
    response.raise_for_status()
    return response.json()

balance = get_balance()
print(f"Remaining: {balance['remainingCredits']} ({balance['usagePercent']}% used)")
print(f"Status: {balance['status']}, recharges on {balance['periodEnd']}")

# Fetch the balance, then evaluate the alert rules with jq.
curl -s -X GET https://api.starleads.co/Consumption \
  -H "X-Api-Key: YOUR_API_KEY" \
| jq '
    if .status != "Active" then "ALERT: subscription is \(.status)"
    elif .remainingCredits < 500 then "ALERT: only \(.remainingCredits) credits left"
    elif .usagePercent >= 80 then "ALERT: \(.usagePercent)% of credits used"
    else "OK"
    end
  '

THRESHOLD_CREDITS = 500      # alert when fewer credits remain
THRESHOLD_PERCENT = 80       # alert when usage crosses this percentage

def should_alert(balance):
    if balance["status"] != "Active":
        return f"Subscription is {balance['status']}"
    if balance["remainingCredits"] < 0:
        return f"Overage: {balance['remainingCredits']} credits (over the included pool)"
    if balance["remainingCredits"] < THRESHOLD_CREDITS:
        return f"Low balance: {balance['remainingCredits']} credits remaining"
    if balance["usagePercent"] >= THRESHOLD_PERCENT:
        return f"High usage: {balance['usagePercent']}% of credits consumed"
    return None

reason = should_alert(balance)
if reason:
    print(f"ALERT -- {reason} (period ends {balance['periodEnd']})")
else:
    print("Balance is healthy.")

curl -X GET "https://api.starleads.co/Consumption/breakdown?from=2026-01-01&to=2026-01-31" \
  -H "X-Api-Key: YOUR_API_KEY"

from datetime import date, timedelta

def get_breakdown(from_date, to_date):
    response = requests.get(
        f"{BASE_URL}/Consumption/breakdown",
        headers={"X-Api-Key": API_KEY},
        params={"from": from_date.isoformat(), "to": to_date.isoformat()},
    )
    response.raise_for_status()
    return response.json()

# When an alert fires, look at the last 30 days to see where credits went.
if reason:
    today = date.today()
    breakdown = get_breakdown(today - timedelta(days=30), today)

    print(f"Total consumed: {breakdown['totalCredits']}")

    print("By channel:")
    for row in breakdown["byChannel"]:
        print(f"  {row['channel']}: {row['credits']}")

    print("Top agents:")
    top = sorted(breakdown["byAgent"], key=lambda a: a["credits"], reverse=True)
    for agent in top[:5]:
        label = agent["agentId"] or agent["nameStatus"]
        print(f"  {label} ({agent['nameStatus']}): {agent['credits']}")

from datetime import date, timedelta

def monitor():
    balance = get_balance()
    reason = should_alert(balance)
    if not reason:
        return

    # Alert is firing -- attach a 30-day breakdown for context.
    today = date.today()
    breakdown = get_breakdown(today - timedelta(days=30), today)
    top_channel = max(breakdown["byChannel"], key=lambda c: c["credits"], default=None)

    message = (
        f"[Starleads] {reason}. "
        f"Remaining: {balance['remainingCredits']} credits, "
        f"period ends {balance['periodEnd']}."
    )
    if top_channel:
        message += f" Biggest driver (30d): {top_channel['channel']} = {top_channel['credits']} credits."

    # Send `message` to your alerting channel (email, Slack, PagerDuty, ...).
    print(message)

monitor()