SuperPenguin Python SDK

Track AI costs automatically across LLMs and voice. Wrap your OpenAI, Anthropic, Google Gemini, or AWS Bedrock client (or patch litellm) and every LLM call is captured with token counts, estimated cost, latency, and attribution metadata. For voice, the same sp.wrap() works on Deepgram (STT) and ElevenLabs (TTS) clients standalone, and a one-line shim captures LiveKit Agents per-turn metrics + per-session billing rows so a single voice call's STT + LLM + TTS + agent-session all stitch back together on one session_id. No proxy required.

How is this different from native provider attribution? See docs/vs-native-attribution.md for the full breakdown of what OpenAI / Anthropic / Deepgram / ElevenLabs offer natively vs. what SuperPenguin adds on top.

Installation

pip install superpenguin

For OpenTelemetry trace correlation:

pip install "superpenguin[otel]"

Or install from source (in the sdk/python/ directory):

pip install -e .

Live spend reads

Read the latest spend observed for your organization before starting another AI request. The same SDK key authorizes ingestion and spend reads.

import os

import superpenguin as sp

sp.init(api_key=os.environ["SP_API_KEY"])

try:
    current = sp.spend.get(
        window="1d",
    )
    print(current["spendUsd"], current["asOf"], current["stalenessMs"])
except sp.SuperPenguinSpendError as error:
    if error.retryable:
        # Decide whether your application should retry, queue, or continue.
        pass

Supported windows are 1d through 7d, 14d, and month. These are UTC calendar-day windows: 1d is today, numbered windows include today plus the preceding N - 1 days, and month is the current calendar month through today. They are not rolling-hour windows. Filter by feature, model, or both:

filtered = sp.spend.get(
    window="7d",
    feature="document-processing",
    model="gpt-5",
)

spendMicros is the exact USD-micro total as an integer string; spendUsd is a convenience float. window contains the resolved UTC start, exclusive end, type, and timeZone. asOf is when the reader loaded the newest snapshot used by the response, and stalenessMs conservatively estimates how far it may trail usage already accepted by SuperPenguin. There is no public freshThrough field. The result is advisory, not a budget reservation.

The SDK flushes its local queue before reading by default, but ingestion and materialization remain asynchronous. Pass flush=False to skip that flush. Identical concurrent reads in the same process share one flush and HTTP request.

Production spend reads use https://spend.superpenguin.ai automatically. spend_base_url or SP_SPEND_BASE_URL is an optional advanced override that affects only spend reads; usage flushes continue to use base_url or SP_BASE_URL. An explicit custom base_url is the fallback for both paths unless a spend-only override is set.

SuperPenguinSpendError exposes code, retryable, and optional HTTP status. The SDK makes one bounded spend request and does not retry automatically. Codes are invalid_request, invalid_api_key, rate_limited, spend_unavailable, and request_failed.

Quick Start

1. Wrap your client (one line)

import superpenguin as sp
from openai import OpenAI

sp.init(api_key="sp-...")  # your SuperPenguin API key

client = sp.wrap(OpenAI())

# Use the client exactly as normal; cost events are captured automatically
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

That's it. Every create() call through the wrapped client is captured with provider, model, token counts, estimated cost (USD), and latency.

2. Works with Anthropic too

import superpenguin as sp
from anthropic import Anthropic

sp.init(api_key="sp-...")

client = sp.wrap(Anthropic())

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

3. Google Gemini (AI Studio or Vertex AI)

The same sp.wrap() works on the unified google-genai client, which targets either the Gemini API (AI Studio) or Vertex AI:

import superpenguin as sp
from google import genai

sp.init(api_key="sp-...")

# AI Studio
client = sp.wrap(genai.Client(api_key="..."))

# Or Vertex AI
client = sp.wrap(genai.Client(vertexai=True, project="my-gcp", location="us-central1"))

response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents="Hello!",
)

Both generate_content and generate_content_stream are tracked, on client.models and client.aio.models (async). Tiered pricing for gemini-2.5-pro and gemini-3.1-pro-preview is applied automatically based on the input token count.

3.5 AWS Bedrock (boto3 / aioboto3)

The same sp.wrap() instruments a Bedrock runtime client. Wrap the client and every converse / converse_stream call auto-submits a row with the raw modelId, token counts, latency, and metadata:

import superpenguin as sp
import boto3

sp.init(api_key="sp-...")

client = sp.wrap(boto3.client("bedrock-runtime"), metadata={"customer_id": "acme_corp"})

response = client.converse(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
)

async works too via aioboto3 (session.client("bedrock-runtime")); both converse and converse_stream are tracked, with streaming usage accumulated across the event stream into a single row.

Bedrock is a reseller: a model's price on Bedrock differs from its origin vendor (Claude on Bedrock vs. Anthropic direct). To keep this accurate and current, the Bedrock wrapper sends no client-side cost. It emits token facts plus the raw modelId, and the server prices the row with its provider-scoped Bedrock rate card.

v1 scope is the Converse API only. InvokeModel (vendor-specific bodies, no uniform usage) and per-call metadata overrides are not yet supported; pass defaults via sp.wrap(..., metadata={...}).

4. Streaming works transparently

client = sp.wrap(OpenAI())

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

# Cost event is submitted automatically when the stream finishes

5. LiteLLM support

If you use litellm to call 100+ LLM providers through a single interface, one call patches everything:

import superpenguin as sp
import litellm

sp.init(api_key="sp-...")
sp.patch_litellm()

# Every litellm.completion() / litellm.acompletion() is now tracked
response = litellm.completion(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

6. Add attribution metadata

Attach metadata to attribute costs to customers, features, teams, or environments:

# Set defaults for all calls from this client
client = sp.wrap(OpenAI(), metadata={
    "customer_id": "cust_acme_123",
    "feature": "doc_summary",
    "team": "product",
    "environment": "production",
})

# Or override per-call via extra_body
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this document"}],
    extra_body={
        "sp_metadata": {
            "customer_id": "cust_other_456",
            "prompt_key": "summarize_v2",
            "prompt_version": "3",
        }
    },
)

7. Standalone Deepgram (STT): sp.wrap(deepgram_client)

If you call Deepgram directly from a backend service (no LiveKit Agents in the loop), wrap the client and every transcribe_url / transcribe_file call auto-submits a row with audio_seconds, the canonical model SKU, and the cost in USD micros.

import superpenguin as sp
from deepgram import DeepgramClient, PrerecordedOptions

sp.init(api_key="sp-...")

dg = sp.wrap(
    DeepgramClient(api_key="..."),
    metadata={"customer_id": "cust_acme_123", "feature": "podcasts"},
    tier="growth",  # optional; drop or set "growth" if you're on the Growth plan
)

result = dg.listen.rest.v("1").transcribe_url(
    {"url": "https://example.com/episode.mp3"},
    PrerecordedOptions(model="nova-3", multilingual=True),
)
# A single request_logs row is submitted automatically with
#   provider="deepgram"
#   model="deepgram/nova-3-multilingual-prerecorded[-growth]"
#   audio_seconds = result.metadata.duration

What's wrapped

SKU routing

Both methods we wrap hit Deepgram's HTTP REST endpoint, which bills against the pre-recorded SKU, about 44% cheaper than the streaming WebSocket SKU on every Nova model. The wrapper composes the right slug from four signals:

• model engine (Nova-3 vs Nova-2, mono- vs multi-lingual, pulled from response.metadata.model_info)
• API endpoint (always -prerecorded for the methods we wrap today)
• commitment tier (-growth when you pass tier="growth"; defaults to PAYG)

Example slugs the dashboard will see: deepgram/nova-3-monolingual-prerecorded, deepgram/nova-3-multilingual-prerecorded-growth, deepgram/nova-2-monolingual-prerecorded.

Per-call metadata

Deepgram's request models don't expose an extra_body hook, so per-call metadata overrides aren't supported in this iteration. Pass defaults via sp.wrap(..., metadata={...}). A sp.context() context manager that works across all wrappers is planned.

Don't double-wrap with LiveKit

If you're already using LiveKitObservability (Section 9), do NOT additionally wrap the Deepgram client used inside the LiveKit Agents plugin. Both paths would emit a row for the same audio. In practice LiveKit Agents constructs its own internal Deepgram client that you don't hold a reference to, so natural separation is the norm.

8. Standalone ElevenLabs (TTS): sp.wrap(elevenlabs_client)

Same sp.wrap() pattern for ElevenLabs. Every text_to_speech.convert / convert_as_stream / text_to_sound_effects.convert call auto-submits a row with characters = len(text), the canonical model SKU, and the cost.

import superpenguin as sp
from elevenlabs.client import ElevenLabs

sp.init(api_key="sp-...")

el = sp.wrap(
    ElevenLabs(api_key="..."),
    metadata={"customer_id": "cust_acme_123", "feature": "ivr-greeting"},
)

audio = el.text_to_speech.convert(
    voice_id="21m00Tcm4TlvDq8ikWAM",
    text="Welcome to SuperPenguin",  # 23 characters
    model_id="eleven_flash_v2_5",
)
# A single request_logs row is submitted automatically with
#   provider="elevenlabs"
#   model="elevenlabs/eleven_flash_v2_5"
#   characters=23

What's wrapped

Stream wrapping

convert_as_stream returns an iterator of audio bytes. The wrapper proxies the iterator and submits the row when you finish consuming it (or call .close()/exit a with block). The cost is fixed at len(text) regardless of how you consume the stream; wrapping the iterator just means latency_ms reflects time-to-completion instead of time-to-iterator-construction.

Failed requests

ElevenLabs doesn't bill failed synthesis. If a call raises, the wrapper still emits a row (so error-rate dashboards stay accurate) with status_code != 200 and cost_usd_micros = 0.

9. Voice agents (LiveKit + Deepgram + ElevenLabs)

If you run a voice agent on top of LiveKit Agents, typically Deepgram for STT, an OpenAI/Anthropic LLM for reasoning, and ElevenLabs for TTS, wire the LiveKitObservability shim into your AgentSession and three of those four billing surfaces are captured per-turn straight from LiveKit's own MetricsCollectedEvent stream:

import asyncio
import superpenguin as sp
from superpenguin.voice import LiveKitObservability
from openai import AsyncOpenAI

sp.init(api_key="sp-...")

llm_client = sp.wrap(AsyncOpenAI())  # ← LLM still goes through sp.wrap()

obs = LiveKitObservability(
    session_id=ctx.room.name,  # any stable per-call id; LiveKit room name works
    metadata={"customer_id": "cust_acme_123", "feature": "drive-thru"},
)

@session.on("metrics_collected")
def _on_metrics(ev):
    asyncio.create_task(obs.on_metrics(ev))

# When the call ends, emit the LiveKit session-minute + observability-event rows:
await obs.on_session_end(duration_seconds=elapsed_seconds)

What gets emitted

Defaults match the catalogued Deepgram / ElevenLabs SKUs. Override per session via the constructor if you're on a different SKU:

obs = LiveKitObservability(
    session_id=ctx.room.name,
    stt_provider="deepgram", stt_model="deepgram/nova-3-multilingual",
    tts_provider="elevenlabs", tts_model="elevenlabs/eleven_v3",
    routing_path="livekit_inference",  # ← if you're using LiveKit's bundled inference
)

Why no LLM row from the shim?

LLMMetrics is intentionally a no-op. The LLM call is already captured by the sp.wrap() client wrapper with full token / cache / tool detail; emitting a second row from the voice shim would double the LLM line on the dashboard.

Cross-provider correlation

Every row the shim emits carries the same session_id, so the dashboard can join Deepgram + LLM + ElevenLabs + LiveKit rows back into a single voice call. Each row also carries a deterministic idempotency_key = f"{session_id}:{kind}:{request_id}", so an SDK-side retry is deduplicated server-side via the idx_request_logs_idempotency index.

Subscription quotas

Quotas (LiveKit Build/Ship/Scale, ElevenLabs Pro/etc., Deepgram Growth) are not applied per-row because per-call costing has no view of monthly aggregation. The dashboard subtracts subscription_tiers[tier].included_units from the monthly sum before applying the per-unit overage rate, using the tier auto-detected from your Connect-Provider integration.

Voice provider integration model

Three integration paths, three different shapes:

• Server-pull is daily aggregates pulled from the provider's billing API on a cron. It is fast to set up but coarse (no per-call attribution, no latency).
• Standalone wrap is per-call rows from the SDK with full latency and metadata; pick this when you call Deepgram or ElevenLabs directly from a backend service.
• LiveKit shim is per-turn rows from LiveKit's own MetricsCollectedEvent stream; pick this when you run inside livekit-agents. Don't combine with standalone wrap on the same client because that double-counts.

LiveKit itself doesn't expose a billing or usage API at all, so the SDK shim is the only path that produces LiveKit cost rows. The LiveKit Connect-Provider entry exists purely so the dashboard can show "LiveKit · connected" and store the credential triple for future control-plane operations (force-disconnect a runaway session, list active rooms, etc.).

@sp.trace Decorator

For multi-step pipelines (RAG, agents, chains), use the @sp.trace decorator. Any wrapped LLM calls inside the function are automatically linked as children.

import superpenguin as sp
from openai import OpenAI

sp.init(api_key="sp-...")
client = sp.wrap(OpenAI())


@sp.trace
def answer_question(question: str) -> str:
    docs = search_knowledge_base(question)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context:\n{docs}"},
            {"role": "user", "content": question},
        ],
    )

    return response.choices[0].message.content


result = answer_question("How do I reset my password?")

Decorator variants

@sp.trace
def my_function(): ...

@sp.trace("my-pipeline")
def my_function(): ...

@sp.trace(name="my-pipeline", tags=["production"], metadata={"customer_id": "acme"})
def my_function(): ...

Async support

Both wrap() and @sp.trace work with async clients and functions:

from openai import AsyncOpenAI

client = sp.wrap(AsyncOpenAI())


@sp.trace
async def answer_question(question: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
    )
    return response.choices[0].message.content

Configuration

sp.init()

Environment variables

If SP_API_KEY is set, init() is called automatically on first use.

sp.wrap()

sp.patch_litellm()

sp.flush()

Force-flush any pending events. Useful before process exit in short-lived scripts:

sp.flush()

An atexit handler also flushes automatically on normal interpreter shutdown.

Metadata Fields

What Gets Tracked

Each event includes:

LLM rows (from sp.wrap() / sp.patch_litellm())

Voice rows (from standalone sp.wrap() or LiveKitObservability)

Batch APIs

Keep sp.wrap() on a shared OpenAI-compatible client used for both live and deferred requests. The wrapper skips successful non-streaming job acknowledgements that have missing or null usage, or a queued / in_progress status. It still records terminal responses with an explicit zero-valued usage object and all failed requests.

When deferred work finishes, submit each completed item with sp.get_client().submit(...) or a track_*_batch helper and a stable idempotency_key.

Provider Batch APIs (OpenAI / Anthropic / Google) return results asynchronously, so there is no live call to wrap. After a batch job completes, hand the retrieved batch to the matching tracker and the SDK records one request_logs row per request in the job with batch=true:

import superpenguin as sp
from openai import OpenAI

sp.init(api_key="sp-...")
client = OpenAI()

batch = client.batches.retrieve("batch_abc")  # status == "completed"
count = sp.track_openai_batch(client, batch, metadata={"feature": "nightly-eval"})
# Also: sp.track_anthropic_batch(client, batch, ...) and sp.track_google_batch(...)

Batch rows are billed at the provider's discounted batch rate where the pricing catalog defines one.

Content Capture (opt-in)

Off by default. The SDK sends only cost metadata, never prompt or completion text, unless your organization opts in from Settings (Pro or Enterprise). Once enabled, the SDK captures a sampled subset of prompts and completions automatically, there is nothing to call to turn it on. Samples are text-only (images/audio/binary stripped), redacted by default, encrypted at rest, and deletable by the org owner from Settings.

Grouping multi-turn conversations

Set session_id (one conversation) and turn_id (one turn) in metadata. Each model call is its own captured row, so a turn that makes a tool call (model -> tool -> model) spans two rows. Reuse the same turn_id across those calls to group them:

with sp.metadata({
    "session_id":  conversation_id,
    "turn_id":     turn_id,
    "customer_id": "cust_acme_123",
    "prompt_key":  "support-reply",
}):
    first = client.chat.completions.create(model="gpt-4o", messages=msgs)
    # ...the model requests a tool; you run it locally...
    final = client.chat.completions.create(model="gpt-4o", messages=msgs_with_tool_result)

Within a row, the model's tool calls are stored on the outcome side and the tool results you send back on the prompt side. If you omit turn_id, each row gets a fresh random one and will not group.

Narrowing capture from the SDK

sp.configure_content_capture() can only reduce capture (it can never enable it, the Settings opt-in is always required):

sp.configure_content_capture(
    allow=True,           # set False to veto ALL capture from this process
    record_prompt=True,   # set False to drop inputs, keep outputs
    record_outcome=True,  # set False to drop outputs, keep inputs
    sample_rate_multiplier=1.0,   # scale the server sample rate down (0.0-1.0)
    redactor=None,        # custom str -> str scrubber; default redacts emails/keys
)

To exclude a sensitive section entirely, call sp.configure_content_capture(allow=False) around it (the control is process-global, not per call).

Cohere

Available in the published package since Python SDK 0.15.0.

Wrap cohere.ClientV2 or cohere.AsyncClientV2 from the optional cohere dependency (version 7.2 or later in the 7.x series). Both synchronous and asynchronous calls are supported. The wrapper captures usage for text Embed, Rerank, Chat and streaming Chat.

import cohere
import superpenguin as sp

sp.init(api_key="YOUR_SUPERPENGUIN_KEY")
client = sp.wrap(cohere.ClientV2(api_key="YOUR_COHERE_KEY"))
response = client.embed(
    model="embed-v5.0-fast",
    texts=["A small penguin."],
    input_type="search_document",
    embedding_types=["float"],
)
sp.flush()

You can also wrap an OpenAI client pointed at https://api.cohere.ai/compatibility/v1. Use a Cohere API key and omit input_type for Embed on that endpoint. Use the native Cohere client when you need to specify the input type or call Rerank.

For AWS Bedrock, wrap a boto3 or aioboto3 client. Cohere BedrockClient Embed and BedrockClientV2 Rerank are also supported with synchronous clients. Read or close AWS response bodies after each request. AWS calls are attributed to Bedrock and retain their region.

Cost requires provider-reported billing quantities and a matching model price. Bedrock Rerank responses without billed search units leave per-request cost unknown. Unknown cost does not mean free usage.

Usage tracking for batch jobs, image or multimodal Embed, Parse, Transcribe and SageMaker is not supported. The provider guide lists endpoint restrictions and retired Bedrock Chat models separately from requests with unknown cost.

See the Cohere provider guide for verified models, both API examples and supported operations.

Decision APIs (available since 0.16.0)

This release is published. Production readiness is recorded in PR #815, and package publication is recorded in PR #819.

wrap(client) detects OpenAI and TypeSafe clients. For OpenAI, the same wrapped client captures generation through responses.create / chat.completions.create and decisions through decisions.create; no separate decision wrapper or per-call tracking helper is needed. The explicit wrap_typesafe(client) helper is also available. Each decision request emits one event. OpenAI decisions and TypeSafe System One preserve returned model identity and reported usage. Locally qualified with OpenAI 3.26.0 and typesafe-sdk 0.7.4 using mocked HTTP. Older OpenAI clients without a decisions resource still work. Decision output usage is retained even when the reviewed rate is zero; missing usage remains unknown. Decision content capture is disabled.

Microsoft Foundry decisions can be captured automatically on an HTTPX client:

import os
import httpx
import superpenguin as sp

# Initialize SuperPenguin normally, then configure once.
with sp.wrap_microsoft_decision(
    httpx.Client(timeout=30), region="eastus", deployment_type="global",
    metadata={"feature": "routing"},
) as client:
    response = client.post(
        os.environ["AZURE_FOUNDRY_ENDPOINT"] + "/providers/microsoft/v1/systemone",
        headers={"api-key": os.environ["AZURE_FOUNDRY_API_KEY"]},
        json={
            "model": os.environ["AZURE_DECISION_DEPLOYMENT"],
            "state": "The parcel arrived intact.",
            "questions": {"intact": {"type": "noul", "instructions": "Is the parcel intact?"}},
        },
    )
    response.raise_for_status()
    decision = response.json()

httpx.Client, httpx.AsyncClient and their HTTPX2 equivalents are supported. For async use, wrap the AsyncClient once and await its normal requests. Only POSTs to the exact Microsoft decision path on *.services.ai.azure.com are captured. Model/usage are read automatically from a buffered JSON response; malformed, missing or unread streaming usage stays unknown. Request streams are never consumed for instrumentation. Original responses and exceptions are preserved. Other calls pass through untouched, including Azure generation requests; keep using sp.wrap on the generation provider client. Region and deployment type must match the reviewed serving route. No retries or global patches are introduced. Do not also call track_decision for the same request. Explicit reporting remains available for other HTTP stacks. No prompt, answers, headers or key are captured.

These APIs are not yet published. Production pricing/schema and provider qualification must pass the repository release runbook before publication.