Cost Dashboard Agent (Anthropic SDK Tutorial)

TL;DR — Build a self-monitoring agent that tracks every API call it makes, calculates real-time cost, enforces budget limits, and produces daily/weekly spending reports. Uses the Anthropic SDK to call Claude on SandBase. Implements the budget control patterns from our per-call pricing guide. Complete code included.

What we’re building

An agent that:

Performs useful tasks (answering questions, analyzing data)
Logs every API call with token counts and costs
Tracks spending against daily/weekly/monthly budgets
Produces cost reports on demand or on schedule
Self-throttles when approaching budget limits

This is the “observability-first” agent pattern — the agent understands its own economics and can make cost-aware decisions.

Why self-monitoring matters

Most agents run blind on cost. They make API calls, tokens accumulate, and the bill arrives at month-end. This creates problems:

Budget surprises: An agent in a reasoning loop can burn through $50 in minutes
No cost attribution: Which task or user drove the spending?
No adaptive behavior: Agent doesn’t know if it should use a cheaper model
No early warning: By the time you notice, the budget is already blown

A self-monitoring agent solves all four by making cost a first-class input to its decision-making.

Prerequisites

pip install anthropic pydantic

Required:

SandBase API key with access to Claude models
Anthropic SDK (works with SandBase’s Anthropic-compatible endpoint)

The complete agent

Core: Cost tracking infrastructure

"""
Cost Dashboard Agent — Anthropic SDK on SandBase
Self-monitoring agent that tracks and reports its own API spending.
"""

import json
import time
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
from anthropic import Anthropic
from pydantic import BaseModel

# --- Configuration ---
SANDBASE_API_KEY = "your-sandbase-api-key"
DAILY_BUDGET_USD = 5.00
WEEKLY_BUDGET_USD = 25.00
MONTHLY_BUDGET_USD = 80.00

# --- Anthropic Client (via SandBase) ---
client = Anthropic(
    base_url="https://api.sandbase.ai/anthropic",
    api_key=SANDBASE_API_KEY,
)

# --- Pricing Table (per million tokens, 2026) ---
PRICING = {
    "claude-sonnet-5": {"input": 3.00, "output": 15.00},
    "claude-haiku-4": {"input": 0.80, "output": 4.00},
    "claude-opus-5": {"input": 15.00, "output": 75.00},
}

# --- Data Models ---
@dataclass
class APICallLog:
    timestamp: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0
    cost_usd: float = 0.0
    task_label: str = ""
    duration_ms: int = 0

@dataclass
class BudgetState:
    daily_spent: float = 0.0
    weekly_spent: float = 0.0
    monthly_spent: float = 0.0
    daily_limit: float = 5.00
    weekly_limit: float = 25.00
    monthly_limit: float = 80.00
    total_calls: int = 0
    last_reset_daily: str = ""
    last_reset_weekly: str = ""

class CostReport(BaseModel):
    period: str
    total_cost: float
    total_calls: int
    avg_cost_per_call: float
    model_breakdown: dict
    top_tasks: list
    budget_remaining: dict
    recommendations: list

# --- Cost Calculator ---
class CostCalculator:
    """Calculate costs based on token usage and model pricing."""
    
    @staticmethod
    def calculate(model: str, input_tokens: int, output_tokens: int,
                  cache_read_tokens: int = 0, cache_write_tokens: int = 0) -> float:
        pricing = PRICING.get(model, PRICING["claude-haiku-4"])
        
        # Standard token costs
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        
        # Cache pricing (Anthropic-style)
        # Cache reads are 90% cheaper than regular input
        cache_read_cost = (cache_read_tokens / 1_000_000) * pricing["input"] * 0.1
        # Cache writes cost 25% more than regular input
        cache_write_cost = (cache_write_tokens / 1_000_000) * pricing["input"] * 1.25
        
        return input_cost + output_cost + cache_read_cost + cache_write_cost

Cost-aware agent wrapper

# --- Cost-Aware Agent ---
class CostDashboardAgent:
    """Agent that tracks its own spending and enforces budgets."""
    
    def __init__(self, model: str = "claude-sonnet-5"):
        self.model = model
        self.logs: list[APICallLog] = []
        self.budget = BudgetState(
            daily_limit=DAILY_BUDGET_USD,
            weekly_limit=WEEKLY_BUDGET_USD,
            monthly_limit=MONTHLY_BUDGET_USD,
            last_reset_daily=datetime.now().strftime("%Y-%m-%d"),
            last_reset_weekly=datetime.now().strftime("%Y-%W"),
        )
    
    def _check_budget(self) -> tuple[bool, str]:
        """Check if we're within budget. Returns (ok, reason)."""
        today = datetime.now().strftime("%Y-%m-%d")
        week = datetime.now().strftime("%Y-%W")
        
        # Reset daily counter if new day
        if self.budget.last_reset_daily != today:
            self.budget.daily_spent = 0.0
            self.budget.last_reset_daily = today
        
        # Reset weekly counter if new week
        if self.budget.last_reset_weekly != week:
            self.budget.weekly_spent = 0.0
            self.budget.last_reset_weekly = week
        
        if self.budget.daily_spent >= self.budget.daily_limit:
            return False, f"Daily budget exhausted (${self.budget.daily_spent:.2f}/${self.budget.daily_limit:.2f})"
        if self.budget.weekly_spent >= self.budget.weekly_limit:
            return False, f"Weekly budget exhausted (${self.budget.weekly_spent:.2f}/${self.budget.weekly_limit:.2f})"
        if self.budget.monthly_spent >= self.budget.monthly_limit:
            return False, f"Monthly budget exhausted (${self.budget.monthly_spent:.2f}/${self.budget.monthly_limit:.2f})"
        
        return True, "OK"
    
    def _select_model(self, task_complexity: str = "normal") -> str:
        """Dynamically select model based on budget pressure."""
        remaining_daily = self.budget.daily_limit - self.budget.daily_spent
        
        # If >80% of daily budget used, downgrade to cheaper model
        if remaining_daily < self.budget.daily_limit * 0.2:
            print(f"  ⚠ Budget pressure: switching to claude-haiku-4")
            return "claude-haiku-4"
        
        # For simple tasks, always use cheaper model
        if task_complexity == "simple":
            return "claude-haiku-4"
        
        return self.model
    
    def call(self, messages: list[dict], task_label: str = "",
             task_complexity: str = "normal",
             system: str = None) -> Optional[str]:
        """Make an API call with cost tracking and budget enforcement."""
        
        # Budget check
        within_budget, reason = self._check_budget()
        if not within_budget:
            print(f"  🛑 BLOCKED: {reason}")
            return None
        
        # Model selection
        model = self._select_model(task_complexity)
        
        # Make the call
        start = time.time()
        
        kwargs = {"model": model, "max_tokens": 1024, "messages": messages}
        if system:
            kwargs["system"] = system
        
        response = client.messages.create(**kwargs)
        
        duration_ms = int((time.time() - start) * 1000)
        
        # Extract usage
        usage = response.usage
        input_tokens = usage.input_tokens
        output_tokens = usage.output_tokens
        cache_read = getattr(usage, 'cache_read_input_tokens', 0) or 0
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0) or 0
        
        # Calculate cost
        cost = CostCalculator.calculate(
            model, input_tokens, output_tokens, cache_read, cache_write
        )
        
        # Log the call
        log = APICallLog(
            timestamp=datetime.now().isoformat(),
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read,
            cache_write_tokens=cache_write,
            cost_usd=cost,
            task_label=task_label,
            duration_ms=duration_ms,
        )
        self.logs.append(log)
        
        # Update budget
        self.budget.daily_spent += cost
        self.budget.weekly_spent += cost
        self.budget.monthly_spent += cost
        self.budget.total_calls += 1
        
        # Print real-time cost
        print(f"  💰 ${cost:.4f} | {model} | {input_tokens}→{output_tokens} tok | {duration_ms}ms | {task_label}")
        
        return response.content[0].text
    
    def get_daily_report(self) -> CostReport:
        """Generate a report for today's spending."""
        today = datetime.now().strftime("%Y-%m-%d")
        today_logs = [l for l in self.logs if l.timestamp.startswith(today)]
        
        # Model breakdown
        model_costs = {}
        for log in today_logs:
            if log.model not in model_costs:
                model_costs[log.model] = {"calls": 0, "cost": 0.0, "tokens": 0}
            model_costs[log.model]["calls"] += 1
            model_costs[log.model]["cost"] += log.cost_usd
            model_costs[log.model]["tokens"] += log.input_tokens + log.output_tokens
        
        # Top tasks by cost
        task_costs = {}
        for log in today_logs:
            label = log.task_label or "unlabeled"
            task_costs[label] = task_costs.get(label, 0) + log.cost_usd
        top_tasks = sorted(task_costs.items(), key=lambda x: x[1], reverse=True)[:5]
        
        total_cost = sum(l.cost_usd for l in today_logs)
        
        return CostReport(
            period=f"Daily ({today})",
            total_cost=total_cost,
            total_calls=len(today_logs),
            avg_cost_per_call=total_cost / len(today_logs) if today_logs else 0,
            model_breakdown=model_costs,
            top_tasks=[{"task": t, "cost": c} for t, c in top_tasks],
            budget_remaining={
                "daily": self.budget.daily_limit - self.budget.daily_spent,
                "weekly": self.budget.weekly_limit - self.budget.weekly_spent,
                "monthly": self.budget.monthly_limit - self.budget.monthly_spent,
            },
            recommendations=self._generate_recommendations(today_logs),
        )
    
    def _generate_recommendations(self, logs: list[APICallLog]) -> list[str]:
        """Generate cost optimization recommendations."""
        recs = []
        
        if not logs:
            return ["No data yet for recommendations."]
        
        # Check if expensive model is being used for simple tasks
        expensive_calls = [l for l in logs if l.model == "claude-opus-5"]
        if expensive_calls:
            avg_output = sum(l.output_tokens for l in expensive_calls) / len(expensive_calls)
            if avg_output < 200:
                recs.append(
                    f"Consider using claude-haiku-4 for short responses. "
                    f"{len(expensive_calls)} Opus calls averaged only {avg_output:.0f} output tokens."
                )
        
        # Check cache utilization
        total_input = sum(l.input_tokens for l in logs)
        total_cache_read = sum(l.cache_read_tokens for l in logs)
        if total_input > 10000 and total_cache_read < total_input * 0.1:
            recs.append(
                "Cache utilization is low (<10%). Enable prompt caching for repeated system prompts "
                "to reduce costs by up to 90% on cached content."
            )
        
        # Budget pressure warning
        daily_pct = self.budget.daily_spent / self.budget.daily_limit
        if daily_pct > 0.7:
            recs.append(
                f"Daily budget is {daily_pct*100:.0f}% used. "
                "Consider deferring non-urgent tasks or switching to cheaper models."
            )
        
        return recs if recs else ["Spending is within normal parameters."]

Running the agent

# --- Usage Example ---
def demo():
    """Demonstrate the cost dashboard agent."""
    
    agent = CostDashboardAgent(model="claude-sonnet-5")
    
    print("🤖 Cost Dashboard Agent — Demo")
    print(f"   Daily budget: ${DAILY_BUDGET_USD}")
    print(f"   Model: claude-sonnet-5 (auto-downgrades under pressure)")
    print()
    
    # Task 1: Simple question (should use cheap model)
    print("📌 Task 1: Simple question")
    result = agent.call(
        messages=[{"role": "user", "content": "What is 2+2?"}],
        task_label="simple_math",
        task_complexity="simple"
    )
    print(f"   → {result}\n")
    
    # Task 2: Complex analysis
    print("📌 Task 2: Complex analysis")
    result = agent.call(
        messages=[{"role": "user", "content": 
            "Analyze the cost implications of running 100 AI agents "
            "each making 50 API calls per day at $0.01 average per call. "
            "Include monthly projections and optimization strategies."}],
        task_label="cost_analysis",
        task_complexity="normal"
    )
    print(f"   → {result[:150]}...\n")
    
    # Task 3: Using cache (system prompt reuse)
    system_prompt = (
        "You are a financial analyst AI specializing in API cost optimization. "
        "Always provide specific dollar amounts and percentages."
    )
    
    print("📌 Task 3: Cached system prompt")
    for i, question in enumerate([
        "What's the cost difference between GPT-4.1 and Claude Sonnet 5?",
        "How much can prompt caching save on repeated queries?",
    ]):
        result = agent.call(
            messages=[{"role": "user", "content": question}],
            system=system_prompt,
            task_label=f"financial_q{i+1}",
            task_complexity="normal"
        )
        print(f"   → {result[:100]}...\n")
    
    # Generate report
    print("\n" + "="*60)
    print("📊 DAILY COST REPORT")
    print("="*60)
    report = agent.get_daily_report()
    print(f"  Period: {report.period}")
    print(f"  Total cost: ${report.total_cost:.4f}")
    print(f"  Total calls: {report.total_calls}")
    print(f"  Avg cost/call: ${report.avg_cost_per_call:.4f}")
    print(f"\n  Model breakdown:")
    for model, stats in report.model_breakdown.items():
        print(f"    {model}: {stats['calls']} calls, ${stats['cost']:.4f}")
    print(f"\n  Top tasks by cost:")
    for task in report.top_tasks:
        print(f"    {task['task']}: ${task['cost']:.4f}")
    print(f"\n  Budget remaining:")
    for period, remaining in report.budget_remaining.items():
        print(f"    {period}: ${remaining:.2f}")
    print(f"\n  Recommendations:")
    for rec in report.recommendations:
        print(f"    → {rec}")

if __name__ == "__main__":
    demo()

How it implements budget controls

The agent implements three levels of cost control from our per-call pricing article:

Level 1: Hard budget limits

if self.budget.daily_spent >= self.budget.daily_limit:
    return None  # Refuse to make the call

The agent will not make API calls once a budget is exhausted. This prevents runaway costs.

Level 2: Adaptive model selection

if remaining_daily < self.budget.daily_limit * 0.2:
    return "claude-haiku-4"  # 4x cheaper than Sonnet

When budget pressure increases, the agent automatically downgrades to cheaper models. Quality degrades gracefully rather than stopping entirely.

Level 3: Task-aware routing

if task_complexity == "simple":
    return "claude-haiku-4"  # Don't waste expensive models on simple tasks

Simple questions get routed to cheap models regardless of budget state. This is the cheapest optimization — it costs nothing to implement and saves 60-80% on trivial calls.

Cache pricing integration

The agent tracks Anthropic’s cache pricing to show real savings:

# Cache reads: 90% cheaper than regular input
cache_read_cost = (cache_read_tokens / 1_000_000) * pricing["input"] * 0.1

# Cache writes: 25% premium on first use
cache_write_cost = (cache_write_tokens / 1_000_000) * pricing["input"] * 1.25

For agents with repeated system prompts (most production agents), cache utilization can reduce input costs by 90%. The cost report surfaces this metric so you know if caching is working.

Weekly report automation

def generate_weekly_report_markdown(agent: CostDashboardAgent) -> str:
    """Generate a markdown weekly report for stakeholders."""
    
    # Use the agent itself to synthesize the report
    logs_summary = summarize_logs(agent.logs, days=7)
    
    result = agent.call(
        messages=[{"role": "user", "content": f"""
Generate a weekly cost report in markdown format.

Data:
{json.dumps(logs_summary, indent=2)}

Include:
1. Executive summary (one paragraph)
2. Cost trends (up/down vs last week)
3. Model usage breakdown table
4. Top 10 tasks by cost
5. Budget utilization chart (text-based)
6. Optimization recommendations

Format as clean markdown suitable for email or Slack."""}],
        task_label="weekly_report_generation",
        task_complexity="normal"
    )
    return result

Production deployment pattern

import schedule

def run_production_agent():
    """Production deployment with scheduled reports."""
    
    agent = CostDashboardAgent(model="claude-sonnet-5")
    
    # Schedule reports
    schedule.every().day.at("09:00").do(
        lambda: send_report(agent.get_daily_report(), channel="slack")
    )
    schedule.every().monday.at("09:00").do(
        lambda: send_report(generate_weekly_report_markdown(agent), channel="email")
    )
    
    # Main loop: agent handles tasks while self-monitoring
    while True:
        schedule.run_pending()
        task = get_next_task()  # Your task queue
        if task:
            agent.call(
                messages=[{"role": "user", "content": task.prompt}],
                task_label=task.label,
                task_complexity=task.complexity
            )
        time.sleep(1)

Cost comparison: With vs without self-monitoring

Scenario	Without monitoring	With monitoring	Savings
Normal week	$25.00	$18.00	28% (model routing)
Budget spike (loop bug)	$150+ (discovered at billing)	$5.00 (hard limit hit)	97%
Low-traffic day	$15.00 (same model always)	$4.00 (auto-downgrade)	73%
Repeated queries	$8.00 (no caching)	$2.40 (cache hits)	70%

The biggest savings come from preventing runaway spending and automatically routing simple tasks to cheaper models. The self-monitoring overhead (~$0.001 per tracked call for the logging itself) is negligible.

Key takeaways

Cost is a first-class agent input. The agent should know what it’s spending and adjust behavior accordingly.
Three control levels: Hard limits (prevent disasters), adaptive models (graceful degradation), task routing (eliminate waste).
Cache visibility matters. If you can’t measure cache hits, you can’t optimize them. See Anthropic cache pricing deep dive.
Reports close the loop. Daily/weekly reports turn cost from a surprise into a managed metric.
The Anthropic SDK on SandBase works identically to direct Anthropic access — same SDK, same patterns, same caching behavior — with unified billing across all models.