revefi-usage-collection-sdk

Collect your AI provider usage and cost data inside your own environment and deliver it to Revefi — or to your own Snowflake account, so nothing has to leave your side except what you choose.

Currently supported provider: Anthropic (Usage & Cost Admin API). The record model is provider-neutral; OpenAI and Amazon Bedrock are planned.

How it works

1. You create an Anthropic Admin API key (sk-ant-admin01-..., Console → Settings → Organization → Admin keys). The key never leaves your environment — this SDK calls Anthropic directly from wherever you run it.
2. The SDK fetches two datasets:
   • Cost — daily billed line items (/v1/organizations/cost_report): exact dollars by model, workspace, token type, cost type, context window, service tier.
   • Usage — token counts at minute grain (/v1/organizations/usage_report/messages): uncached/cached input, cache writes, output tokens, web-search requests, by workspace/API key/model/tier/context window.
3. Data goes to one or both sinks:
   • Revefi — the raw API response pages, verbatim, as gzipped NDJSON over HTTPS (one JSON line per response page, each carrying its request window/params alongside). Revefi normalises server-side with the same code path it uses when pulling directly, so nothing is dropped client-side and the two paths can't drift.
   • Snowflake — locally normalised rows into LLM_PROVIDER_COST / LLM_PROVIDER_USAGE tables in your account, mirroring Revefi's internal genai.llm_provider_* tables (same columns, metrics_json shape, and update_hash recipe), MERGEd on UPDATE_HASH so re-running any window upserts instead of duplicating. Revefi then reads them through your existing Snowflake connection.

Only aggregate usage/cost rows are collected — no prompts, completions, or message content.

Install

pip install revefi-usage-collection-sdk            # Revefi sink + local export
pip install 'revefi-usage-collection-sdk[snowflake]'  # + Snowflake sink

Quick start (CLI)

export ANTHROPIC_ADMIN_KEY=sk-ant-admin01-...

# See exactly what would be sent — writes local NDJSON, sends nothing
revefi-usage export --days 7 --out ./audit

# Check credentials/connectivity
export REVEFI_INGEST_URL=https://ingest.revefi.com
export REVEFI_TOKEN=...
revefi-usage test-connection --sink revefi

# Collect the trailing 30 days and send to Revefi
revefi-usage collect --sink revefi --days 30

# ...or upload to your own Snowflake instead (or both)
export SNOWFLAKE_ACCOUNT=myorg-myaccount SNOWFLAKE_USER=... SNOWFLAKE_PASSWORD=...
revefi-usage collect --sink snowflake --days 30

Configuration precedence: CLI flags > environment variables > --config revefi-usage.yaml (scaffold one with revefi-usage init). Snowflake supports password or key-pair auth (SNOWFLAKE_PRIVATE_KEY_PATH / SNOWFLAKE_PRIVATE_KEY_PASSPHRASE).

Scheduling

collect is one-shot and idempotent — run it from cron (hourly or daily):

# hourly, re-fetching the trailing 30 days (restated days upsert in place)
0 * * * * . /etc/revefi-usage.env && /usr/local/bin/revefi-usage collect --sink revefi --days 30 >> /var/log/revefi-usage.log 2>&1

collect exits non-zero on any failure — alert on that. A silently dead collector is indistinguishable from zero spend.

Quick start (SDK)

from revefi_usage_sdk import AnthropicProvider, UsageCollector, RevefiSink

provider = AnthropicProvider(admin_key="sk-ant-admin01-...")
collector = UsageCollector(provider)

# Just get typed records and do your own thing
cost, usage = collector.fetch(days=30, bucket_width="1m")

# Or deliver to sinks (each sink is independent; failures are aggregated)
report = collector.run(sinks=[RevefiSink(ingest_url=..., token=...)], days=30)
assert report.ok, report.errors

Known limitations (Anthropic API)

• ~90 days of history at the source — backfill cannot go further.
• Priority Tier dollars are absent from the cost endpoint (the tokens do appear in usage with service_tier: priority).
• The admin endpoints do not exist for Claude on Amazon Bedrock / Google Vertex organisations, or for individual (non-organization) accounts.
• Data lands ~5 minutes after requests complete; keep polling to at most once per minute.

Development

pip install -e '.[dev,snowflake]'
pytest
python -m build

Mock ingest server

tools/mock_ingest_server.py stands in for the Revefi ingest endpoint and appends everything it receives (decompressed, one JSON line per record, with a delivery marker per request) to a file:

python tools/mock_ingest_server.py --port 8080 --out received.ndjson

export REVEFI_INGEST_URL=http://localhost:8080 REVEFI_TOKEN=dummy
revefi-usage collect --sink revefi --days 7
cat received.ndjson

Releasing

Publishing is tag-driven through GitHub Actions (.github/workflows/release.yml), following the same convention as revefi-ingestion-cli: pushing a release-<version> tag tests, builds, and uploads to PyPI. The workflow refuses to publish unless the tagged commit is contained in main and the tag suffix matches the pyproject.toml version, so a stale tag or a forgotten version bump fails loudly instead of shipping.

One-time setup: add a repo secret PYPI_API_TOKEN (Settings → Secrets and variables → Actions) holding a PyPI API token scoped to revefi-usage-collection-sdk.

Per release:

1. Bump version in pyproject.toml and __version__ in src/revefi_usage_sdk/__init__.py (keep them identical), merge to main.
2. Tag the merge commit and push the tag:

   git tag release-<version> <main-sha> && git push origin release-<version>
   

3. The release workflow publishes; verify at https://pypi.org/project/revefi-usage-collection-sdk/

Manual fallback: python -m build && twine check dist/* && twine upload dist/* (needs the PyPI API token). Note PyPI versions are immutable — a broken upload means bumping to the next version, not re-uploading.