Anthropic Observability
Overview
Instrument Claude API calls with structured logging, Prometheus metrics, and cost tracking. Every API response includes usage data and rate limit headers — capture these for dashboards and alerting.
Structured Logging
import anthropic
import logging
import time
import json
logger = logging.getLogger("claude")
def create_with_logging(client: anthropic.Anthropic, **kwargs) -> anthropic.types.Message:
start = time.monotonic()
request_meta = {
"model": kwargs.get("model"),
"max_tokens": kwargs.get("max_tokens"),
"tool_count": len(kwargs.get("tools", [])),
"stream": kwargs.get("stream", False),
}
try:
response = client.messages.create(**kwargs)
duration_ms = int((time.monotonic() - start) * 1000)
logger.info(json.dumps({
"event": "claude.request",
"request_id": response._request_id,
"model": response.model,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"cache_read_tokens": getattr(response.usage, "cache_read_input_tokens", 0),
"stop_reason": response.stop_reason,
"duration_ms": duration_ms,
"content_blocks": len(response.content),
}))
return response
except anthropic.APIStatusError as e:
duration_ms = int((time.monotonic() - start) * 1000)
logger.error(json.dumps({
"event": "claude.error",
"status": e.status_code,
"error_type": getattr(e, "type", "unknown"),
"duration_ms": duration_ms,
"request_id": e.response.headers.get("request-id", "unknown"),
}))
raise
Prometheus Metrics
from prometheus_client import Counter, Histogram, Gauge
claude_requests = Counter(
"claude_requests_total", "Total Claude API requests",
["model", "stop_reason", "status"]
)
claude_latency = Histogram(
"claude_latency_seconds", "Claude API latency",
["model"], buckets=[0.5, 1, 2, 5, 10, 30, 60]
)
claude_tokens = Counter(
"claude_tokens_total", "Token usage",
["model", "direction"] # direction: input|output|cache_read
)
claude_cost = Counter(
"claude_cost_usd", "Estimated cost in USD",
["model"]
)
claude_rate_limit_remaining = Gauge(
"claude_rate_limit_remaining", "Remaining rate limit",
["dimension"] # dimension: requests|tokens
)
def track_metrics(response, duration: float):
model = response.model
claude_requests.labels(model=model, stop_reason=response.stop_reason, status="ok").inc()
claude_latency.labels(model=model).observe(duration)
claude_tokens.labels(model=model, direction="input").inc(response.usage.input_tokens)
claude_tokens.labels(model=model, direction="output").inc(response.usage.output_tokens)
# Cost estimation
pricing = {"claude-haiku-4-20250514": (0.80, 4.0), "claude-sonnet-4-20250514": (3.0, 15.0)}
rates = pricing.get(model, (3.0, 15.0))
cost = (response.usage.input_tokens * rates[0] + response.usage.output_tokens * rates[1]) / 1e6
claude_cost.labels(model=model).inc(cost)
Key Metrics Dashboard
| Metric | Description | Alert Threshold |
|--------|-------------|-----------------|
| claude_requests_total{status="error"} | Error count | > 5% of total |
| claude_latency_seconds p99 | Tail latency | > 10s |
| claude_cost_usd daily | Daily spend | > 80% budget |
| claude_rate_limit_remaining{dimension="requests"} | RPM headroom | < 10% remaining |
| claude_tokens_total{direction="output"} rate | Output throughput | Spike detection |
Usage API (Server-Side)
# Anthropic's Usage & Cost API for billing reconciliation
# GET https://api.anthropic.com/v1/usage
# Returns daily token usage and cost per model
Error Handling
| Observability Gap | Risk | Fix |
|-------------------|------|-----|
| No request_id logged | Can't debug with support | Capture response._request_id |
| Missing cost tracking | Budget surprise | Track per-request cost |
| No latency histogram | Can't spot slow queries | Add Prometheus/Datadog histograms |
Prerequisites
- Define SLOs, alert owners, budget and rate-limit thresholds, approved metric labels, and retention rules for telemetry.
- Configure authenticated server-side access through a secret manager and use a sandbox workspace with synthetic requests to verify instrumentation.
- Establish a redaction/filter policy before enabling logs, traces, dashboards, or usage reconciliation; prompts, responses, secrets, and personal data are never telemetry fields.
Instructions
- Instrument the request boundary with request ID, model, status, stop reason, token aggregates, cache counters, and duration while excluding content and high-cardinality identifiers.
- Emit success and failure metrics for authentication, 4xx/5xx, 429, timeout, latency, spend, and remaining rate-limit headroom. Validate labels against an allowlist and cap cardinality.
- Test dashboards and alerts with synthetic success, timeout, rate-limit, permission, and malformed-response fixtures. Verify the alert path without sending live customer data.
- Reconcile usage through the approved authenticated server-side API on a bounded schedule, compare aggregate totals, and alert on unexplained divergence or budget breach.
- Canary telemetry changes, then promote with owner approval. If redaction, cardinality, or retention checks fail, disable the new sink, restore the prior configuration, and preserve only a redacted receipt.
Output
Produce an observability receipt containing instrumentation version, metric/label allowlist, synthetic test results, alert thresholds, aggregate usage/cost/latency/error outcomes, retention policy, canary scope, approval, and rollback reference. Exclude prompts, responses, API keys, user IDs, and raw exception bodies.
Examples
Send synthetic fixture-request-001 through a staging client and assert request_id_present=1; content_fields=0; labels_allowlisted=1; inject a synthetic 429 and verify the alert fires. Record telemetry=pass; retention=24h; rollback=metrics-v1 without recording the fixture text.
Resources
Next Steps
For incident response, see anth-incident-runbook.