Send Claude Platform traces to Dynatrace

import os
import sys
from datetime import datetime, timedelta, timezone

import requests

ANTHROPIC = "https://api.anthropic.com/v1/organizations/analytics"
HEADERS = {
    "x-api-key": os.environ["ANTHROPIC_ADMIN_KEY"],
    "anthropic-version": "2023-06-01",
    "anthropic-beta": "analytics-usage-cost-2026-08-24",
    "User-Agent": "Dynatrace/1.0.0 (https://www.dynatrace.com)",
}
DT_INGEST = os.environ["DT_ENV_URL"] + "/api/v2/bizevents/ingest"
DT_HEADERS = {"Authorization": "Bearer " + os.environ["DT_TOKEN"]}
GROUP_BY = ["model", "workspace_id", "service_tier", "product"]


def fetch(path, params):
    """Return all rows for a query, restarting if a page cursor expires."""
    for _ in range(3):  # restart if a page cursor expires
        rows, query = [], dict(params, limit=1000)
        while True:
            resp = requests.get(f"{ANTHROPIC}/{path}", headers=HEADERS, params=query, timeout=60)
            if resp.status_code == 410:  # cursor expired after a data refresh
                break
            resp.raise_for_status()
            body = resp.json()
            refreshed = body["data_refreshed_at"]
            if refreshed is None or datetime.fromisoformat(refreshed.replace("Z", "+00:00")) < end:
                sys.exit(f"{path}: data not complete yet (refreshed at {refreshed}), try later")
            rows += body["data"]
            if not body["has_more"]:
                return rows
            query["page"] = body["next_page"]
    sys.exit(f"{path}: page cursor kept expiring, try again later")


day = os.environ.get("IMPORT_DAY")  # YYYY-MM-DD (UTC); defaults to the previous UTC day
today = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
start = datetime.strptime(day, "%Y-%m-%d").replace(tzinfo=timezone.utc) if day else today - timedelta(days=1)
end = start + timedelta(days=1)
window = {
    "starting_at": start.strftime("%Y-%m-%dT%H:%M:%SZ"),
    "ending_at": end.strftime("%Y-%m-%dT%H:%M:%SZ"),
}
events = []

for row in fetch("usage", {**window, "time_bucket_width": "1h", "group_by[]": GROUP_BY}):
    g, u = row["group"], row["usage"]
    events.append({
        "event.type": "anthropic.usage",
        "event.provider": "claude-platform",
        "bucket.start": row["starting_at"],
        "bucket.end": row["ending_at"],
        "gen_ai.provider.name": "anthropic",
        "gen_ai.request.model": g["model"],
        "workspace_id": g["workspace_id"],
        "service_tier": g["service_tier"],
        "product": g["product"],
        "requests": u["requests"],
        "uncached_input_tokens": u["uncached_input_tokens"],
        "cache_read_input_tokens": u["cache_read_input_tokens"],
        "cache_creation_5m_input_tokens": u["cache_creation"]["ephemeral_5m_input_tokens"],
        "cache_creation_1h_input_tokens": u["cache_creation"]["ephemeral_1h_input_tokens"],
        "output_tokens": u["output_tokens"],
        "total_tokens": u["total_tokens"],
        "web_search_requests": u["server_tool_use"]["web_search_requests"],
    })

for row in fetch("cost", {**window, "time_bucket_width": "1d",
                          "group_by[]": GROUP_BY + ["cost_type", "token_type"]}):
    g, c = row["group"], row["cost"]
    events.append({
        "event.type": "anthropic.cost",
        "event.provider": "claude-platform",
        "bucket.start": row["starting_at"],
        "bucket.end": row["ending_at"],
        "gen_ai.provider.name": "anthropic",
        "gen_ai.request.model": g["model"],
        "workspace_id": g["workspace_id"],
        "service_tier": g["service_tier"],
        "product": g["product"],
        "cost_type": g["cost_type"],
        "token_type": g["token_type"],
        "cost.usd": float(c["amount"]) / 100,
        "cost.list_usd": float(c["list_amount"]) / 100,
    })

for i in range(0, len(events), 500):
    requests.post(DT_INGEST, headers=DT_HEADERS, json=events[i:i + 500], timeout=30).raise_for_status()
print(f"Sent {len(events)} events")