Understanding AI Agent Cost Explosions
AI agents — autonomous systems that chain multiple LLM calls, tool invocations, and reasoning steps — can generate costs at a velocity that catches even experienced developers off guard. A single agentic workflow might invoke an LLM dozens of times, each call carrying its own token-based price tag. Without deliberate monitoring, a runaway loop, an overly verbose chain-of-thought, or a misconfigured retry strategy can turn a $5 daily bill into a $500 catastrophe overnight.
An AI Agent Cost Alert is a programmatic guardrail that continuously tracks token consumption, estimates real-time dollar costs, and fires notifications — Slack messages, emails, PagerDuty alerts, or even automated circuit breakers — when spending crosses predefined thresholds. It transforms opaque API billing into actionable, time-sensitive signals.
Why Cost Alerts Matter for Agentic Workloads
Traditional LLM applications (single-prompt chatbots, classification endpoints) have predictable per-request costs. Agentic workloads break that predictability. Consider these real failure modes:
- Infinite reasoning loops: An agent stuck in a plan-execute-observe cycle may call GPT-4o hundreds of times before hitting a max_iter limit.
- Runaway tool cascades: A search agent might recursively expand queries, each expansion spawning fresh LLM calls with ever-growing context windows.
- Context bloat: Agents that accumulate conversation history, tool outputs, and reflection logs can push token counts per call from 2K to 50K+ without warning.
- Retry storms: Exponential backoff on rate-limited APIs can mask the fact that you're paying for every failed attempt.
Cost alerts give you the early warning system needed to catch these scenarios within minutes, not days — before your CFO asks uncomfortable questions about a cloud services line item.
Architecture of a Cost Alert System
A production-grade cost alert pipeline has four layers:
- Instrumentation: Wrapping every LLM call to capture token usage (prompt_tokens + completion_tokens) and model pricing metadata.
- Aggregation: Rolling windows (per-session, per-agent, per-hour, per-day) that accumulate costs in near real-time.
- Threshold evaluation: Comparing aggregated costs against configurable limits.
- Notification & action: Dispatching alerts and optionally triggering circuit breakers (pausing the agent, degrading to cheaper models, or killing the session).
Step 1: Instrumenting LLM Calls for Token Tracking
The foundation is a thin wrapper around your LLM provider's SDK. Here's a production-oriented example using OpenAI's API with a cost-tracking decorator pattern:
import json
import time
from dataclasses import dataclass, field
from typing import List, Callable, Optional
from openai import OpenAI
# Pricing table (USD per 1K tokens) — keep this updated
MODEL_PRICING = {
"gpt-4o": {"prompt": 0.005, "completion": 0.015},
"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
"gpt-4-turbo": {"prompt": 0.01, "completion": 0.03},
"gpt-3.5-turbo": {"prompt": 0.0005, "completion": 0.0015},
"claude-3-5-sonnet-20241022": {"prompt": 0.003, "completion": 0.015},
"claude-3-haiku-20240307": {"prompt": 0.00025, "completion": 0.00125},
}
@dataclass
class CostEvent:
"""Immutable record of a single LLM call's cost."""
timestamp: float
model: str
prompt_tokens: int
completion_tokens: int
cost_usd: float
agent_session_id: str
call_duration_ms: float
@dataclass
class CostTracker:
"""Accumulates cost events and evaluates thresholds."""
session_id: str
events: List[CostEvent] = field(default_factory=list)
on_threshold_breach: Optional[Callable] = None
max_session_cost: float = 10.0 # default: $10 per agent session
max_hourly_cost: float = 50.0 # default: $50 per hour
def record(self, event: CostEvent) -> None:
self.events.append(event)
self._evaluate_thresholds()
def total_session_cost(self) -> float:
return sum(e.cost_usd for e in self.events)
def hourly_cost(self) -> float:
now = time.time()
window_start = now - 3600
return sum(e.cost_usd for e in self.events if e.timestamp >= window_start)
def _evaluate_thresholds(self) -> None:
session_cost = self.total_session_cost()
hour_cost = self.hourly_cost()
if session_cost >= self.max_session_cost:
if self.on_threshold_breach:
self.on_threshold_breach(
"session", session_cost, self.max_session_cost, self.session_id
)
if hour_cost >= self.max_hourly_cost:
if self.on_threshold_breach:
self.on_threshold_breach(
"hourly", hour_cost, self.max_hourly_cost, self.session_id
)
def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Calculate USD cost from token counts and pricing table."""
if model not in MODEL_PRICING:
# Unknown model — log warning and use a conservative estimate
print(f"WARNING: Unknown model {model}, using fallback pricing")
return (prompt_tokens + completion_tokens) * 0.01 / 1000
pricing = MODEL_PRICING[model]
prompt_cost = (prompt_tokens / 1000) * pricing["prompt"]
completion_cost = (completion_tokens / 1000) * pricing["completion"]
return prompt_cost + completion_costStep 2: Creating the Instrumented LLM Client
Now we build a client class that wraps OpenAI calls, extracts usage metadata from the response, and pushes cost events into the tracker:
class MonitoredLLMClient:
"""Wrapper around OpenAI client that tracks every call's cost."""
def __init__(self, tracker: CostTracker, openai_client: OpenAI):
self.tracker = tracker
self.client = openai_client
def chat_completion(
self,
model: str = "gpt-4o",
messages: list = None,
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> dict:
"""Drop-in replacement for chat.completions.create with cost tracking."""
start_time = time.time()
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
duration_ms = (time.time() - start_time) * 1000
# Extract token counts from the response
usage = response.usage
if usage:
prompt_tokens = usage.prompt_tokens
completion_tokens = usage.completion_tokens
cost = estimate_cost(model, prompt_tokens, completion_tokens)
event = CostEvent(
timestamp=time.time(),
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
cost_usd=cost,
agent_session_id=self.tracker.session_id,
call_duration_ms=duration_ms
)
self.tracker.record(event)
return {
"content": response.choices[0].message.content,
"usage": {
"prompt_tokens": usage.prompt_tokens if usage else 0,
"completion_tokens": usage.completion_tokens if usage else 0,
"total_tokens": usage.total_tokens if usage else 0,
"estimated_cost_usd": round(cost, 6) if usage else 0
}
}
def streamed_completion(self, model: str = "gpt-4o", messages: list = None, **kwargs):
"""Generator that yields chunks AND tracks cost after completion.
Note: streaming responses report usage only on the final chunk.
"""
start_time = time.time()
stream = self.client.chat.completions.create(
model=model,
messages=messages,
stream=True,
stream_options={"include_usage": True}, # OpenAI beta feature
**kwargs
)
collected_content = []
final_usage = None
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
collected_content.append(chunk.choices[0].delta.content)
yield {"delta": chunk.choices[0].delta.content, "done": False}
# The final chunk may include usage when stream_options is set
if hasattr(chunk, 'usage') and chunk.usage:
final_usage = chunk.usage
duration_ms = (time.time() - start_time) * 1000
if final_usage:
prompt_tokens = final_usage.prompt_tokens
completion_tokens = final_usage.completion_tokens
cost = estimate_cost(model, prompt_tokens, completion_tokens)
event = CostEvent(
timestamp=time.time(),
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
cost_usd=cost,
agent_session_id=self.tracker.session_id,
call_duration_ms=duration_ms
)
self.tracker.record(event)
yield {"content": "".join(collected_content), "done": True}Step 3: Building the Alert Dispatcher
The alert dispatcher listens for threshold breaches and routes notifications to the appropriate channels. Here's a modular implementation supporting Slack, email, and a circuit breaker:
import os
import smtplib
import requests
from email.mime.text import MIMEText
from datetime import datetime
class AlertDispatcher:
"""Routes cost alerts to Slack, email, and/or programmatic circuit breakers."""
def __init__(self, slack_webhook_url: str = None, email_config: dict = None):
self.slack_webhook_url = slack_webhook_url
self.email_config = email_config
self.circuit_breakers = {} # session_id -> bool (True = open/blocked)
def handle_breach(self, breach_type: str, current_cost: float,
threshold: float, session_id: str) -> None:
"""Entry point called by CostTracker when a threshold is breached."""
alert_message = (
f"⚠️ **AI Agent Cost Alert**\n"
f"Breach Type: `{breach_type}`\n"
f"Session ID: `{session_id}`\n"
f"Current Cost: `${current_cost:.4f}`\n"
f"Threshold: `${threshold:.2f}`\n"
f"Timestamp: {datetime.now().isoformat()}\n"
f"Overrun: `${current_cost - threshold:.4f}` ({(current_cost/threshold - 1)*100:.1f}%)\n"
)
if self.slack_webhook_url:
self._send_slack(alert_message)
if self.email_config:
self._send_email(alert_message)
# Open circuit breaker for runaway sessions
if breach_type == "session" and current_cost >= threshold * 1.5:
self.circuit_breakers[session_id] = True
print(f"CIRCUIT BREAKER OPENED for session {session_id}")
def is_circuit_open(self, session_id: str) -> bool:
return self.circuit_breakers.get(session_id, False)
def _send_slack(self, message: str) -> None:
payload = {
"text": message,
"username": "AI Cost Monitor",
"icon_emoji": ":warning:"
}
try:
response = requests.post(
self.slack_webhook_url,
json=payload,
timeout=5
)
if response.status_code != 200:
print(f"Slack webhook failed: {response.status_code} {response.text}")
except Exception as e:
print(f"Failed to send Slack alert: {e}")
def _send_email(self, message: str) -> None:
if not self.email_config:
return
msg = MIMEText(message.replace('`', '').replace('*', ''))
msg['Subject'] = f"AI Agent Cost Alert - ${message.split('$')[1].split('\\n')[0]}"
msg['From'] = self.email_config.get('from', 'ai-monitor@company.com')
msg['To'] = self.email_config.get('to', 'devops@company.com')
try:
with smtplib.SMTP(self.email_config['smtp_host'],
self.email_config.get('smtp_port', 587)) as server:
server.starttls()
server.login(
self.email_config['smtp_user'],
self.email_config['smtp_password']
)
server.send_message(msg)
except Exception as e:
print(f"Failed to send email alert: {e}")Step 4: Tying It All Together in an Agent Loop
Here's how to integrate cost monitoring into a typical agentic workflow — a ReAct-style agent that reasons, acts, and observes. Notice the circuit breaker check before each LLM call:
class CostMonitoredAgent:
"""ReAct agent with per-step cost tracking and circuit breaker protection."""
def __init__(self, system_prompt: str, tools: list,
model: str = "gpt-4o-mini",
max_steps: int = 20,
session_cost_limit: float = 5.0,
hourly_cost_limit: float = 25.0):
self.tools = {t.__name__: t for t in tools}
self.model = model
self.max_steps = max_steps
self.system_prompt = system_prompt
# Set up cost monitoring infrastructure
self.tracker = CostTracker(
session_id=f"agent-{id(self)}-{int(time.time())}",
max_session_cost=session_cost_limit,
max_hourly_cost=hourly_cost_limit
)
self.dispatcher = AlertDispatcher(
slack_webhook_url=os.environ.get("SLACK_WEBHOOK_URL")
)
self.tracker.on_threshold_breach = self.dispatcher.handle_breach
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
self.llm = MonitoredLLMClient(self.tracker, openai_client)
self.messages = [{"role": "system", "content": system_prompt}]
def run(self, user_query: str) -> str:
self.messages.append({"role": "user", "content": user_query})
for step in range(self.max_steps):
# Circuit breaker check before any LLM call
if self.dispatcher.is_circuit_open(self.tracker.session_id):
return (
f"Agent halted: cost circuit breaker open. "
f"Session cost: ${self.tracker.total_session_cost():.4f}. "
f"Please investigate."
)
# Optional: dynamic model downgrade when costs are high
active_model = self.model
if self.tracker.hourly_cost() > self.tracker.max_hourly_cost * 0.7:
active_model = "gpt-4o-mini" # downgrade to cheaper model
print(f"⚠️ Downgrading to {active_model} due to cost pressure")
response = self.llm.chat_completion(
model=active_model,
messages=self.messages,
temperature=0.7
)
assistant_message = response["content"]
call_info = (
f"Step {step+1} | "
f"Tokens: {response['usage']['total_tokens']} | "
f"Cost: ${response['usage']['estimated_cost_usd']:.4f} | "
f"Session Total: ${self.tracker.total_session_cost():.4f}"
)
print(call_info)
# Parse for tool calls (simplified — adapt to your agent framework)
tool_name, tool_args = self._parse_tool_call(assistant_message)
if tool_name is None:
# Agent provided final answer
return assistant_message
if tool_name in self.tools:
tool_result = self.tools[tool_name](**tool_args)
self.messages.append({
"role": "assistant",
"content": assistant_message
})
self.messages.append({
"role": "user",
"content": f"Tool {tool_name} returned: {tool_result}"
})
else:
self.messages.append({
"role": "assistant",
"content": assistant_message
})
self.messages.append({
"role": "user",
"content": f"Error: tool '{tool_name}' not found. Available: {list(self.tools.keys())}"
})
return "Agent reached maximum steps without final answer."
def _parse_tool_call(self, text: str):
"""Basic tool call parser — replace with structured output in production."""
import re
match = re.search(r'Action:\s*(\w+)\s*\((.*?)\)', text, re.DOTALL)
if match:
tool_name = match.group(1)
args_str = match.group(2)
try:
args = json.loads(args_str)
except json.JSONDecodeError:
args = {"input": args_str.strip()}
return tool_name, args
return None, None
def get_cost_report(self) -> dict:
return {
"session_id": self.tracker.session_id,
"total_cost": self.tracker.total_session_cost(),
"call_count": len(self.tracker.events),
"hourly_cost": self.tracker.hourly_cost(),
"circuit_open": self.dispatcher.is_circuit_open(self.tracker.session_id),
"events": [
{
"model": e.model,
"cost": e.cost_usd,
"tokens": e.prompt_tokens + e.completion_tokens
}
for e in self.tracker.events[-10:] # last 10 for brevity
]
}Step 5: Running the Agent with Cost Visibility
Here's a complete example demonstrating the agent with two simulated tools and full cost monitoring:
import os
import time
# Simulated tools
def search_web(query: str) -> str:
"""Simulate a web search — returns truncated results."""
return f"Search results for '{query}': Found 3 relevant articles about {query[:30]}..."
def calculator(expression: str) -> str:
"""Evaluate a mathematical expression safely."""
try:
result = eval(expression, {"__builtins__": {}}, {})
return f"Result: {result}"
except Exception as e:
return f"Error: {str(e)}"
# Configure environment (set these in your .env or secrets manager)
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
os.environ["SLACK_WEBHOOK_URL"] = "https://hooks.slack.com/services/..."
# Initialize the agent
agent = CostMonitoredAgent(
system_prompt="You are a helpful assistant. Use tools when needed. "
"Format tool calls as: Action: tool_name({\"arg\": \"value\"})",
tools=[search_web, calculator],
model="gpt-4o-mini",
max_steps=10,
session_cost_limit=5.0,
hourly_cost_limit=25.0
)
# Run a query
print("=" * 60)
print("STARTING AGENT RUN — Cost monitoring active")
print("=" * 60)
result = agent.run(
"What is the population of Tokyo in 2024? "
"Search for the data, then calculate the population density "
"if the area is 2,194 square kilometers."
)
print("\\n" + "=" * 60)
print("AGENT RESULT:")
print(result)
print("\\n" + "=" * 60)
# Print cost report
import json
cost_report = agent.get_cost_report()
print("\\nCOST REPORT:")
print(json.dumps(cost_report, indent=2))
print(f"\\nTotal Session Cost: ${cost_report['total_cost']:.4f}")
print(f"Number of LLM Calls: {cost_report['call_count']}")
print(f"Average Cost Per Call: ${cost_report['total_cost']/max(1,cost_report['call_count']):.4f}")Advanced: Multi-Provider Cost Aggregation Service
In production, you'll likely have multiple agents running concurrently across different LLM providers (OpenAI, Anthropic, Google, Azure). A centralized cost aggregation service with persistent storage is essential. Here's a lightweight version using Redis for real-time counters:
import redis
import json
from datetime import datetime
class RedisCostAggregator:
"""Multi-agent, multi-provider cost tracking backed by Redis."""
def __init__(self, redis_client: redis.Redis, retention_hours: int = 24):
self.redis = redis_client
self.retention_hours = retention_hours
def record_cost(self, agent_id: str, model: str,
prompt_tokens: int, completion_tokens: int,
cost_usd: float) -> None:
"""Atomically update multiple cost counters."""
now = datetime.now()
hour_bucket = now.strftime("%Y-%m-%dT%H")
minute_bucket = now.strftime("%Y-%m-%dT%H:%M")
pipeline = self.redis.pipeline()
# Per-agent running totals
pipeline.incrbyfloat(f"agent:{agent_id}:total_cost", cost_usd)
pipeline.incrby(f"agent:{agent_id}:call_count", 1)
pipeline.incrby(f"agent:{agent_id}:total_tokens",
prompt_tokens + completion_tokens)
# Per-model global totals
pipeline.incrbyfloat(f"model:{model}:total_cost", cost_usd)
pipeline.incrby(f"model:{model}:call_count", 1)
# Time-bucketed costs for dashboarding
pipeline.incrbyfloat(f"costs:hourly:{hour_bucket}", cost_usd)
pipeline.incrbyfloat(f"costs:minutely:{minute_bucket}", cost_usd)
# Add to agent's event log with TTL
event_key = f"agent:{agent_id}:events"
event_data = json.dumps({
"timestamp": now.isoformat(),
"model": model,
"cost_usd": cost_usd,
"tokens": prompt_tokens + completion_tokens
})
pipeline.rpush(event_key, event_data)
pipeline.expire(event_key, self.retention_hours * 3600)
pipeline.execute()
def get_agent_cost(self, agent_id: str) -> dict:
"""Retrieve current cost state for an agent."""
total_cost = float(self.redis.get(f"agent:{agent_id}:total_cost") or 0)
call_count = int(self.redis.get(f"agent:{agent_id}:call_count") or 0)
total_tokens = int(self.redis.get(f"agent:{agent_id}:total_tokens") or 0)
return {
"agent_id": agent_id,
"total_cost_usd": total_cost,
"call_count": call_count,
"total_tokens": total_tokens,
"avg_cost_per_call": total_cost / max(1, call_count),
"avg_tokens_per_call": total_tokens / max(1, call_count)
}
def get_hourly_cost(self, hour_bucket: str = None) -> float:
"""Get cost for a specific hour bucket, defaulting to current."""
if hour_bucket is None:
hour_bucket = datetime.now().strftime("%Y-%m-%dT%H")
return float(self.redis.get(f"costs:hourly:{hour_bucket}") or 0)
def check_and_alert(self, agent_id: str,
session_limit: float = 10.0,
hourly_limit: float = 50.0) -> list:
"""Evaluate thresholds and return triggered alerts."""
alerts = []
agent_cost = self.get_agent_cost(agent_id)
current_hour = datetime.now().strftime("%Y-%m-%dT%H")
hourly_cost = self.get_hourly_cost(current_hour)
if agent_cost["total_cost_usd"] >= session_limit:
alerts.append({
"type": "session",
"agent_id": agent_id,
"current": agent_cost["total_cost_usd"],
"limit": session_limit,
"severity": "critical" if agent_cost["total_cost_usd"] >= session_limit * 1.5 else "warning"
})
if hourly_cost >= hourly_limit:
alerts.append({
"type": "hourly",
"hour_bucket": current_hour,
"current": hourly_cost,
"limit": hourly_limit,
"severity": "critical" if hourly_cost >= hourly_limit * 1.5 else "warning"
})
return alertsSetting Up a Cron-Based Alert Sweeper
For a lightweight production setup, run a periodic sweeper that checks all active agents and dispatches alerts. This pairs well with the Redis aggregator above:
import redis
import os
import requests
from datetime import datetime
def sweep_and_alert():
"""Periodic job — run every 60 seconds via cron or a background thread."""
r = redis.Redis(
host=os.environ.get("REDIS_HOST", "localhost"),
port=int(os.environ.get("REDIS_PORT", 6379)),
decode_responses=True
)
aggregator = RedisCostAggregator(r)
slack_url = os.environ.get("SLACK_WEBHOOK_URL")
# Find all active agent keys
agent_keys = r.keys("agent:*:total_cost")
alerted_agents = set()
for key in agent_keys:
agent_id = key.decode() if isinstance(key, bytes) else key
agent_id = agent_id.split(":")[1] # Extract agent ID
alerts = aggregator.check_and_alert(
agent_id,
session_limit=float(os.environ.get("SESSION_COST_LIMIT", "10.0")),
hourly_limit=float(os.environ.get("HOURLY_COST_LIMIT", "50.0"))
)
for alert in alerts:
# Deduplicate: don't re-alert the same agent within 5 minutes
dedup_key = f"{agent_id}:{alert['type']}"
if dedup_key in alerted_agents:
continue
alerted_agents.add(dedup_key)
# Set dedup TTL in Redis
r.setex(f"alert_dedup:{dedup_key}", 300, "1")
if slack_url and alert['severity'] == 'critical':
payload = {
"text": (
f"🚨 *CRITICAL COST ALERT*\n"
f"Agent: `{agent_id}`\n"
f"Type: {alert['type']}\n"
f"Current: ${alert['current']:.4f}\n"
f"Limit: ${alert['limit']:.2f}\n"
f"Time: {datetime.now().isoformat()}"
),
"channel": "#ai-cost-alerts"
}
requests.post(slack_url, json=payload, timeout=5)
print(f"[{datetime.now().isoformat()}] ALERT: {alert}")
if __name__ == "__main__":
sweep_and_alert()Best Practices for AI Agent Cost Alerts
- Set tiered thresholds, not binary ones. Use warning thresholds at 70% and critical thresholds at 100% of your limit. Warnings give you time to investigate before the circuit breaker trips and disrupts users.
- Track costs per agent session, per user, and globally. A single agent might stay within limits, but 500 concurrent sessions could silently bankrupt you. Aggregate at multiple levels.
- Implement graceful degradation. When costs approach thresholds, automatically switch to cheaper models (gpt-4o → gpt-4o-mini), reduce context window sizes, or limit tool call depth rather than hard-killing the agent.
- Log every call's metadata before the LLM invocation. If your code crashes mid-response, you still want the record of what was sent. Log estimated input token counts and model selection before calling the API.
- Use idempotent alert deduplication. A cost overrun that persists across multiple sweeps should not generate a new Slack message every 60 seconds. Use Redis TTL-based dedup keys (shown above) or a proper incident management system.
- Monitor token count trends, not just dollar costs. A gradual increase in average tokens per call can signal prompt bloat or context pollution. Alert on token-per-call anomalies even if the dollar cost hasn't yet breached limits.
- Keep pricing tables in version control and sync them regularly. LLM prices change (usually downward, but occasionally new models launch at premium prices). A stale pricing table leads to inaccurate cost estimates. Consider using a provider's /usage API for authoritative dollar figures where available.
- Test your alerting pipeline with synthetic cost events. Create a "cost canary" agent that deliberately burns through tokens and verify that Slack messages arrive, circuit breakers engage, and on-call rotations are paged before relying on the system in production.
- Expose cost metrics to your agent's decision loop. Advanced agents can be taught to reason about their own costs. Include current session cost in the system prompt so the agent can choose cheaper strategies when it's approaching limits.
Handling Edge Cases
Several edge cases deserve explicit attention in production systems:
- Streaming responses with delayed usage data: When using streaming, token counts arrive only at the end. Buffer the last known cost estimate and flag sessions that have been streaming for unusually long durations (potential runaway generation).
- Multi-turn tool calls with growing context: Each tool result appended to the conversation increases the prompt tokens for subsequent calls. Track the rate of context growth and alert if it's exponential.
- Retry logic that compounds costs: If your agent retries failed LLM calls 3 times with exponential backoff, you've multiplied costs by up to 3x. Always wrap retry logic in cost-aware guards that cap total retry spend per call.
- Provider outages and fallback models: When failing over from OpenAI to Anthropic, ensure your cost tracker correctly maps the fallback model to its pricing tier. A mismatch here can silently under-report or over-report costs.
Dashboard Integration: Making Costs Visible
Alerts are the reactive layer. A dashboard gives you proactive visibility. Here's a minimal FastAPI endpoint that serves cost data for a Grafana or Datadog integration:
from fastapi import FastAPI
import redis
import os
from datetime import datetime, timedelta
app = FastAPI(title="AI Cost Monitor API")
redis_client = redis.Redis(
host=os.environ.get("REDIS_HOST", "localhost"),
decode_responses=True
)
@app.get("/costs/current")
def get_current_costs():
"""Return snapshot of all agent costs for dashboard widgets."""
agent_keys = redis_client.keys("agent:*:total_cost")
agents = []
for key in agent_keys:
agent_id = key.split(":")[1]
total_cost = float(redis_client.get(key) or 0)
call_count = int(redis_client.get(f"agent:{agent_id}:call_count") or 0)
agents.append({
"agent_id": agent_id,
"total_cost_usd": round(total_cost, 4),
"call_count": call_count,
"avg_cost_per_call": round(total_cost / max(1, call_count), 4)
})
global_total = sum(a["total_cost_usd"] for a in agents)
return {
"timestamp": datetime.now().isoformat(),
"total_cost_all_agents": round(global_total, 4),
"active_agents": len(agents),
"agents": agents
}
@app.get("/costs/timeseries")
def get_timeseries(hours: int = 24):
"""Return hourly cost buckets for the last N hours."""
buckets = []
now = datetime.now()
for i in range(hours):
bucket_time = now - timedelta(hours=i)
bucket_key = bucket_time.strftime("costs:hourly:%Y-%m-%dT%H")
cost = float(redis_client.get(bucket_key) or 0)
buckets.append({
"hour": bucket_time.strftime("%Y-%m-%dT%H:00"),
"cost_usd": round(cost, 4)
})
return {
"timeseries": sorted(buckets, key=lambda b: b["hour"]),
"total_in_window": round(sum(b["cost_usd"] for b in buckets), 4)
}
@app.get("/costs/anomalies")
def get_anomalies():
"""Detect sessions with abnormal cost patterns."""
agent_keys = redis_client.keys("agent:*:total_cost")
anomalies = []
for key in agent_keys:
agent_id = key.split(":")[1]
call_count = int(redis_client.get(f"agent:{agent_id}:call_count") or 0)
total_tokens = int(redis