Track Usage Costs and Audit Admin API Events
Track Claude API token usage costs and audit workspace organization events with Anthropic Admin API and automated Prometheus metrics.
Contents
- The Challenge of Production LLM Cost Accounting
- Anthropic Pricing Architecture and Token Vectors
- Anthropic Admin API: Workspaces and Organization Audit Logs
- Prometheus Telemetry and Metrics Design
- Python Implementation: Cost Engine and Metrics Exporter
- TypeScript Implementation: Edge-Ready Telemetry Exporter
- Governance Edge Cases and Anomaly Detection
- FAQ
- Verifying Your Pipeline
The Challenge of Production LLM Cost Accounting
Operating AI agents and large-scale language model workloads across multi-tenant environments introduces severe financial and governance blind spots when organizations rely solely on monthly billing invoices. Without granular token tracking, teams cannot attribute costs to individual customers, developer workspaces, or operational pipelines.
Furthermore, Anthropic introduced prompt caching and extended reasoning tokens, fundamentally shifting pricing mechanics. In modern Claude applications, input tokens cannot be treated as a monolithic cost bucket:
- Base Input Tokens: Billed at standard baseline rates (for example, $3.00 per million tokens on Claude 3.5 and Claude 3.7 Sonnet).
- Cache Creation Tokens: Written to ephemeral memory at a 25% premium ($3.75 per million tokens on Sonnet).
- Cache Read Tokens: Read from warm cache at a 90% discount ($0.30 per million tokens on Sonnet).
- Generated Output Tokens: Billed at generation rates ($15.00 per million tokens on Sonnet), which includes visible output text and internal thinking reasoning tokens.
In this guide, we implement an enterprise cost tracking and organization audit engine that captures token deltas per request, computes real-time dollar expenditures, queries Anthropic Admin API organization events, and exports Prometheus metrics to Grafana.
Anthropic Pricing Architecture and Token Vectors
To compute exact financial impact, our billing engine maintains a normalized pricing matrix across the Claude model family:
| Model Tier | Base Input / 1M | Cache Creation / 1M | Cache Read / 1M | Output & Thinking / 1M |
|---|---|---|---|---|
| Claude 3.7 Sonnet | $3.00 | $3.75 | $0.30 | $15.00 |
| Claude 3.5 Sonnet | $3.00 | $3.75 | $0.30 | $15.00 |
| Claude 3.5 Haiku | $0.80 | $1.00 | $0.08 | $4.00 |
| Claude 3 Opus | $15.00 | $18.75 | $1.50 | $75.00 |
Mathematical Formula for Granular Turn Cost
For any completed request turn $i$ using model $M$, total turn cost $C_i$ is computed as:
$$C_i = \left(\frac{T_{\text{in}} \times R_{\text{in}} + T_{\text{out}} \times R_{\text{out}} + T_{\text{cw}} \times R_{\text{cw}} + T_{\text{cr}} \times R_{\text{cr}}}{1,000,000}\right)$$
Where:
- $T_{\text{in}}$ = Standard input tokens
- $T_{\text{out}}$ = Generated output tokens (including thinking tokens)
- $T_{\text{cw}}$ =
cache_creation_input_tokens - $T_{\text{cr}}$ =
cache_read_input_tokens - $R$ = Corresponding price per million tokens from the matrix
We also track the Cache Efficiency Ratio:
$$\text{Cache Ratio} = \frac{T_{\text{cr}}}{T_{\text{in}} + T_{\text{cw}} + T_{\text{cr}}}$$
A healthy multi-turn agent pipeline should maintain a cache read ratio above 70%, which slashes blended input token costs by upwards of 60% compared to uncached execution.
Anthropic Admin API: Workspaces and Organization Audit Logs
Beyond per-request token meters, enterprise governance demands visibility into organizational security changes. Anthropic exposes dedicated Admin API endpoints to organization administrators under the /v1/organizations namespace:
1. Workspace Inventory (/v1/organizations/workspaces)
Returns all active workspaces within an organization, allowing usage counters to tag metrics with accurate project, tenant, or environment identifiers.
2. Audit Logs (/v1/organizations/audit_logs)
Provides a paginated audit stream of all high-privilege activities:
api_key.created: An API key was provisioned.api_key.deleted: An API key was revoked.workspace_member.added: A new user or service account was granted workspace access.workspace_member.removed: Access was terminated.
Tracking these events in real time allows operations teams to detect compromised credentials or unauthorized key generation.
Prometheus Telemetry and Metrics Design
Our cost tracker exposes four primary Prometheus telemetry series over a standard /metrics HTTP endpoint:
anthropic_tokens_total(Counter):- Labels:
workspace_id,model,token_type(input,output,cache_creation,cache_read) - Tracks absolute volume over time.
- Labels:
anthropic_estimated_cost_dollars(Gauge):- Labels:
workspace_id,model - Represents cumulative dollar expenditure computed per workspace.
- Labels:
anthropic_cache_read_ratio(Gauge):- Labels:
workspace_id,model - Tracks real-time caching performance between 0.0 and 1.0.
- Labels:
anthropic_audit_events_total(Counter):- Labels:
event_type,actor_type - Ingests organization audit occurrences for security monitoring.
- Labels:
Python Implementation: Cost Engine and Metrics Exporter
Here is the complete Python production service using pydantic, httpx, and prometheus_client.
import osimport sysimport timeimport loggingfrom typing import Any, Dict, List, Optionalimport httpxfrom pydantic import BaseModel, Fieldfrom prometheus_client import start_http_server, Counter, Gaugelogging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")logger = logging.getLogger("claude-cost-tracker")MODEL_PRICING: Dict[str, Dict[str, float]] = { "claude-3-7-sonnet": { "input_per_m": 3.00, "output_per_m": 15.00, "cache_write_per_m": 3.75, "cache_read_per_m": 0.30, }, "claude-3-5-sonnet": { "input_per_m": 3.00, "output_per_m": 15.00, "cache_write_per_m": 3.75, "cache_read_per_m": 0.30, }, "claude-3-5-haiku": { "input_per_m": 0.80, "output_per_m": 4.00, "cache_write_per_m": 1.00, "cache_read_per_m": 0.08, }, "claude-3-opus": { "input_per_m": 15.00, "output_per_m": 75.00, "cache_write_per_m": 18.75, "cache_read_per_m": 1.50, },}PROMETHEUS_TOKEN_USAGE = Counter( "anthropic_tokens_total", "Total token consumption across workspaces and models", ["workspace_id", "model", "token_type"],)PROMETHEUS_ESTIMATED_COST = Gauge( "anthropic_estimated_cost_dollars", "Estimated cumulative cost in USD by workspace and model", ["workspace_id", "model"],)PROMETHEUS_CACHE_READ_RATIO = Gauge( "anthropic_cache_read_ratio", "Ratio of cache read tokens to total prompt tokens", ["workspace_id", "model"],)PROMETHEUS_AUDIT_EVENTS = Counter( "anthropic_audit_events_total", "Total audit events ingested from Admin API", ["event_type", "actor_type"],)class UsageRecord(BaseModel): workspace_id: str = Field(default="default") model: str input_tokens: int = 0 output_tokens: int = 0 cache_creation_input_tokens: int = 0 cache_read_input_tokens: int = 0class CostSummary(BaseModel): model: str input_cost: float output_cost: float cache_write_cost: float cache_read_cost: float total_cost: float cache_read_ratio: floatdef get_pricing_key(model_name: str) -> str: m = model_name.lower() if "claude-3-7-sonnet" in m: return "claude-3-7-sonnet" if "claude-3-5-sonnet" in m: return "claude-3-5-sonnet" if "claude-3-5-haiku" in m: return "claude-3-5-haiku" if "claude-3-opus" in m: return "claude-3-opus" return "claude-3-7-sonnet"def calculate_cost(record: UsageRecord) -> CostSummary: rates = MODEL_PRICING.get(get_pricing_key(record.model), MODEL_PRICING["claude-3-7-sonnet"]) input_cost = (record.input_tokens / 1_000_000.0) * rates["input_per_m"] output_cost = (record.output_tokens / 1_000_000.0) * rates["output_per_m"] cache_write_cost = (record.cache_creation_input_tokens / 1_000_000.0) * rates["cache_write_per_m"] cache_read_cost = (record.cache_read_input_tokens / 1_000_000.0) * rates["cache_read_per_m"] total_cost = input_cost + output_cost + cache_write_cost + cache_read_cost total_input = record.input_tokens + record.cache_creation_input_tokens + record.cache_read_input_tokens cache_read_ratio = (record.cache_read_input_tokens / total_input) if total_input > 0 else 0.0 return CostSummary( model=record.model, input_cost=round(input_cost, 6), output_cost=round(output_cost, 6), cache_write_cost=round(cache_write_cost, 6), cache_read_cost=round(cache_read_cost, 6), total_cost=round(total_cost, 6), cache_read_ratio=round(cache_read_ratio, 4), )class AdminAPIClient: def __init__(self, admin_key: str, base_url: str = "https://api.anthropic.com"): self.base_url = base_url.rstrip("/") self.headers = { "x-api-key": admin_key, "anthropic-version": "2023-06-01", "Content-Type": "application/json", } def fetch_audit_logs(self, limit: int = 50) -> List[Dict[str, Any]]: url = f"{self.base_url}/v1/organizations/audit_logs" with httpx.Client(timeout=10.0) as client: try: resp = client.get(url, headers=self.headers, params={"limit": limit}) if resp.status_code == 200: return resp.json().get("data", []) return [] except Exception as e: logger.error("Failed to query audit logs: %s", e) return []def update_prometheus_metrics(record: UsageRecord, summary: CostSummary) -> None: w = record.workspace_id m = record.model PROMETHEUS_TOKEN_USAGE.labels(workspace_id=w, model=m, token_type="input").inc(record.input_tokens) PROMETHEUS_TOKEN_USAGE.labels(workspace_id=w, model=m, token_type="output").inc(record.output_tokens) PROMETHEUS_TOKEN_USAGE.labels(workspace_id=w, model=m, token_type="cache_creation").inc( record.cache_creation_input_tokens ) PROMETHEUS_TOKEN_USAGE.labels(workspace_id=w, model=m, token_type="cache_read").inc( record.cache_read_input_tokens ) PROMETHEUS_ESTIMATED_COST.labels(workspace_id=w, model=m).set(summary.total_cost) PROMETHEUS_CACHE_READ_RATIO.labels(workspace_id=w, model=m).set(summary.cache_read_ratio)TypeScript Implementation: Edge-Ready Telemetry Exporter
For Node.js backends and serverless gateways, we implement the equivalent architecture using prom-client and zod:
import http from "node:http";import { Registry, Counter, Gauge, collectDefaultMetrics } from "prom-client";import { z } from "zod";export interface ModelRates { inputPerM: number; outputPerM: number; cacheWritePerM: number; cacheReadPerM: number;}export const MODEL_PRICING: Record = { "claude-3-7-sonnet": { inputPerM: 3.0, outputPerM: 15.0, cacheWritePerM: 3.75, cacheReadPerM: 0.3 }, "claude-3-5-sonnet": { inputPerM: 3.0, outputPerM: 15.0, cacheWritePerM: 3.75, cacheReadPerM: 0.3 }, "claude-3-5-haiku": { inputPerM: 0.8, outputPerM: 4.0, cacheWritePerM: 1.0, cacheReadPerM: 0.08 }, "claude-3-opus": { inputPerM: 15.0, outputPerM: 75.0, cacheWritePerM: 18.75, cacheReadPerM: 1.5 },};export const UsageRecordSchema = z.object({ workspaceId: z.string().default("default"), model: z.string(), inputTokens: z.number().int().nonnegative().default(0), outputTokens: z.number().int().nonnegative().default(0), cacheCreationInputTokens: z.number().int().nonnegative().default(0), cacheReadInputTokens: z.number().int().nonnegative().default(0),});export type UsageRecord = z.infer;export function calculateCost(record: UsageRecord) { const modelKey = record.model.includes("haiku") ? "claude-3-5-haiku" : record.model.includes("opus") ? "claude-3-opus" : "claude-3-7-sonnet"; const rates = MODEL_PRICING[modelKey]; const inCost = (record.inputTokens / 1_000_000.0) * rates.inputPerM; const outCost = (record.outputTokens / 1_000_000.0) * rates.outputPerM; const cwCost = (record.cacheCreationInputTokens / 1_000_000.0) * rates.cacheWritePerM; const crCost = (record.cacheReadInputTokens / 1_000_000.0) * rates.cacheReadPerM; const totalCost = inCost + outCost + cwCost + crCost; const totalIn = record.inputTokens + record.cacheCreationInputTokens + record.cacheReadInputTokens; const cacheRatio = totalIn > 0 ? record.cacheReadInputTokens / totalIn : 0.0; return { totalCost: Number(totalCost.toFixed(6)), cacheReadRatio: Number(cacheRatio.toFixed(4)), };}Governance Edge Cases and Anomaly Detection
Deploying organization-wide cost governance introduces several subtle operational anomalies:
1. Thinking Token Overflow
When using Claude 3.7 Sonnet with extended thinking enabled, reasoning tokens are billed at output rates ($15.00 per million tokens). If a task loops on ambiguous instructions, reasoning traces can consume thousands of tokens per turn. Set deterministic thinking budgets using budget_tokens and alert when output_tokens exceeds 4,000 in non-code tasks.
2. Admin API Rate Throttling
The Anthropic Admin API enforces strict rate limits distinct from standard Messages API limits. Do not poll /v1/organizations/audit_logs on a high-frequency loop per incoming user request. Instead, run a centralized collector daemon that polls audit events every 60 seconds with exponential backoff.
3. Asynchronous Batch Cost Reconciliations
When utilizing the Message Batches API, token pricing is cut by 50%. Ensure your billing reconciler accounts for this discount so batch workloads are not billed at standard on-demand pricing tiers. Review our guide on Asynchronous Batch Processing for batch discount mechanics.
FAQ
- How do thinking tokens factor into the billable calculation?
Thinking tokens generated during extended reasoning are included directly within the
usage.output_tokenscounter emitted by the Messages API. As a result, they are billed at the standard output rate for the chosen model ($15.00 per million tokens on Sonnet).
- Can standard API keys query the Admin API endpoints?
No. Endpoints under
/v1/organizationsrequire an Anthropic Admin API Key created by an Organization Administrator. Standard workspace-level API keys will receive a 403 Forbidden response.
- What is the most effective way to lower Claude token expenditure?
Implement prompt caching on system prompts and multi-turn tool catalogs, and schedule non-urgent background tasks through the Message Batches API. Combining prompt caching with batch processing reduces total compute expenditure by up to 80%.
Verifying Your Pipeline
The companion repository contains an automated verification probe that validates calculations across all four pricing tiers, verifies TypeScript syntax, and runs key safety scans:
./test_probe.shExpected output:
=== Running Usage Cost Tracker & Admin API Probe ===[1/3] Testing Python Cost Calculation & Pricing Matrix...Python calculation unit checks passed![2/3] Validating TypeScript file syntax and imports...TypeScript source structure verified![3/3] Scanning for leaked keys...Secret scan clean: No API keys found.=== All Probes Succeeded (<5s) ===For complete code examples, Docker deployment templates, and Grafana dashboard manifests, inspect the ZeroLabs Recipes GitHub Repository. For related optimization techniques, see our guides on Configuring Prompt Caching Breakpoints and Optimizing Token Costs with Ephemeral Cache.