Usage & cost Admin API cookbook Aug 2025 • Observability Programmatically access and analyze your Claude API usage and cost data via Admin API.

import os
from datetime import datetime, time, timedelta
from typing import Any
import requests

class AnthropicAdminAPI:
    """Secure wrapper for Anthropic Admin API endpoints."""
    
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or os.getenv("ANTHROPIC_ADMIN_API_KEY")
        if not self.api_key:
            raise ValueError("Admin API key required. Set ANTHROPIC_ADMIN_API_KEY environment variable.")
        if not self.api_key.startswith("sk-ant-admin"):
            raise ValueError("Invalid Admin API key format.")
        self.base_url = "https://api.anthropic.com/v1/organizations"
        self.headers = {
            "anthropic-version": "2023-06-01",
            "x-api-key": self.api_key,
            "Content-Type": "application/json",
        }
    
    def _make_request(self, endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
        """Make authenticated request with basic error handling."""
        url = f"{self.base_url}/{endpoint}"
        try:
            response = requests.get(url, headers=self.headers, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise ValueError("Invalid API key or insufficient permissions") from e
            elif response.status_code == 429:
                raise requests.exceptions.RequestException("Rate limit exceeded - try again later") from e
            else:
                raise requests.exceptions.RequestException(f"API error: {e}") from e
    
    def test_connection(self):
        """Test Admin API connection."""
        try:
            params = {
                "starting_at": (datetime.combine(datetime.utcnow(), time.min) - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%SZ"),
                "ending_at": datetime.combine(datetime.utcnow(), time.min).strftime("%Y-%m-%dT%H:%M:%SZ"),
                "bucket_width": "1d",
                "limit": 1,
            }
            self._make_request("usage_report/messages", params)
            print("✅ Connection successful!")
            return True
        except Exception as e:
            print(f"❌ Connection failed: {e}")
            return False

def get_daily_usage(client, days_back=7):
    """Get usage data for the last N days."""
    end_time = datetime.combine(datetime.utcnow(), time.min)
    start_time = end_time - timedelta(days=days_back)
    params = {
        "starting_at": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "ending_at": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "bucket_width": "1d",
        "limit": days_back,
    }
    return client._make_request("usage_report/messages", params)

def analyze_usage_data(response):
    """Process and display usage data."""
    if not response or not response.get("data"):
        print("No usage data found.")
        return
    
    total_uncached_input = total_output = total_cache_creation = 0
    total_cache_reads = total_web_searches = 0
    daily_data = []
    
    for bucket in response["data"]:
        date = bucket["starting_at"][:10]
        bucket_uncached = bucket_output = bucket_cache_creation = 0
        bucket_cache_reads = bucket_web_searches = 0
        
        for result in bucket["results"]:
            bucket_uncached += result.get("uncached_input_tokens", 0)
            bucket_output += result.get("output_tokens", 0)
            cache_creation = result.get("cache_creation", {})
            bucket_cache_creation += cache_creation.get("ephemeral_1h_input_tokens", 0) + cache_creation.get("ephemeral_5m_input_tokens", 0)
            bucket_cache_reads += result.get("cache_read_input_tokens", 0)
            server_tools = result.get("server_tool_use", {})
            bucket_web_searches += server_tools.get("web_search_requests", 0)
        
        daily_data.append({
            "date": date,
            "uncached_input_tokens": bucket_uncached,
            "output_tokens": bucket_output,
            "cache_creation": bucket_cache_creation,
            "cache_reads": bucket_cache_reads,
            "web_searches": bucket_web_searches,
            "total_tokens": bucket_uncached + bucket_output,
        })
        
        total_uncached_input += bucket_uncached
        total_output += bucket_output
        total_cache_creation += bucket_cache_creation
        total_cache_reads += bucket_cache_reads
        total_web_searches += bucket_web_searches
    
    total_input_tokens = total_uncached_input + total_cache_creation + total_cache_reads
    cache_efficiency = ((total_cache_reads / total_input_tokens * 100) if total_input_tokens > 0 else 0)
    
    print("📊 Usage Summary:")
    print(f"Uncached input tokens: {total_uncached_input:,}")
    print(f"Output tokens: {total_output:,}")
    print(f"Cache creation: {total_cache_creation:,}")
    print(f"Cache reads: {total_cache_reads:,}")
    print(f"Cache efficiency: {cache_efficiency:.1f}%")
    print(f"Web searches: {total_web_searches:,}")
    
    return daily_data

def get_daily_costs(client, days_back=7):
    """Get cost data for the last N days."""
    end_time = datetime.combine(datetime.utcnow(), time.min)
    start_time = end_time - timedelta(days=days_back)
    params = {
        "starting_at": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "ending_at": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "bucket_width": "1d",
        "limit": min(days_back, 31),
    }
    return client._make_request("cost_report", params)

def analyze_cost_data(response):
    """Process and display cost data."""
    if not response or not response.get("data"):
        print("No cost data found.")
        return
    
    total_cost_minor_units = 0
    daily_costs = []
    
    for bucket in response["data"]:
        date = bucket["starting_at"][:10]
        bucket_cost = 0
        
        for result in bucket["results"]:
            amount = result.get("amount", 0)
            if isinstance(amount, str):
                try:
                    amount = float(amount)
                except (ValueError, TypeError):
                    amount = 0
            bucket_cost += amount
        
        daily_costs.append({
            "date": date,
            "cost_usd": bucket_cost / 100,
        })
        total_cost_minor_units += bucket_cost
    
    total_cost_usd = total_cost_minor_units / 100
    
    print("💰 Cost Summary:")
    print(f"Total cost: ${total_cost_usd:.2f}")
    print(f"Daily breakdown:")
    for day in daily_costs:
        print(f"  {day['date']}: ${day['cost_usd']:.2f}")
    
    return daily_costs