superpenguin 0.9.0

SuperPenguin Python SDK — AI cost management, attribution, and spend tracking

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 .

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

Method	Sync	Async
client.listen.rest.v("1").transcribe_url	✓	✓ (via client.listen.asyncrest)
client.listen.rest.v("1").transcribe_file	✓	✓ (via client.listen.asyncrest)
Live WebSocket STT (listen.live.v(...))	⏳ planned (v2)	⏳ planned (v2)
Callback-mode async transcribe (callback= URL)	✗ — response is empty	✗
Deepgram TTS (speak.rest.v(...))	⏳ pricing seed pending	⏳
Voice Agent API (agent.v(...))	⏳ pricing seed pending	⏳

SKU routing

Both methods we wrap hit Deepgram's HTTP REST endpoint, which bills against the pre-recorded SKU — ~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

Method	Sync	Async
client.text_to_speech.convert	✓	✓ (AsyncElevenLabs)
client.text_to_speech.convert_as_stream	✓ (stream proxy)	✓ (async stream proxy)
client.text_to_sound_effects.convert	✓	✓
client.text_to_sound_effects.convert_as_stream	✓ (stream proxy)	✓
Real-time WebSocket TTS (text_to_speech.stream)	⏳ planned	⏳
Speech-to-Speech (speech_to_speech.convert)	⏳ pricing seed pending	⏳
Voice changer	⏳ pricing seed pending	⏳

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

LiveKit event	Provider row	Billing unit
STTMetrics (per turn)	deepgram / nova-3-monolingual	audio_seconds
TTSMetrics (per turn)	elevenlabs / eleven_flash_v2_5	characters (+ audio_seconds for analytics)
LLMMetrics (per turn)	skipped — already captured by sp.wrap()	—
on_session_end()	livekit / agent-session-minute	audio_seconds (= session duration)
on_session_end()	livekit / observability-event	events (count of MetricsCollectedEvent fan-out)

Defaults match the pricing entries seeded in pricing/models.json. 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 — 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

Provider	Server-pull (Connect Provider)	SDK standalone wrap	SDK LiveKit shim
Deepgram	✓ daily usage breakdown	✓ sp.wrap(DeepgramClient(...))	✓ per-turn STT
ElevenLabs	✓ daily usage analytics	✓ sp.wrap(ElevenLabs(...))	✓ per-turn TTS
LiveKit	credentials only — no usage API	✗ (no standalone surface to wrap)	✓ per-session billing rows (the only source)

Three integration paths, three different shapes:

• Server-pull is daily aggregates pulled from the provider's billing API on a cron — 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 — 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()

Parameter	Type	Default	Description
api_key	str	SP_API_KEY env var	Your SuperPenguin API key
base_url	str	https://app.superpenguin.ai	API endpoint
flush_interval	float	5.0	Seconds between background batch flushes
batch_size	int	50	Max events per batch POST

Environment variables

Variable	Description
SP_API_KEY	API key (used if not passed to init())
SP_BASE_URL	API base URL override

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

sp.wrap()

Parameter	Type	Default	Description
client	OpenAI | Anthropic | genai.Client | DeepgramClient | ElevenLabs	required	The client to wrap. Provider is auto-detected from type(client).__module__.
name	str	None	Override the default event name (LLM wrappers only today)
metadata	dict	None	Default metadata for every call (customer_id, feature, team, etc.)
tags	list[str]	None	Tags added to every event
tier	str	None	Deepgram only. "growth" routes rows to the discounted Growth-plan SKU; omit / None for PAYG. Raises TypeError for non-Deepgram providers.

sp.patch_litellm()

Parameter	Type	Default	Description
name	str	None	Override the default event name
metadata	dict	None	Default metadata for every litellm call
tags	list[str]	None	Tags added to every event

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

Field	Type	Purpose
customer_id	string	End-customer or account consuming the AI call
feature	string	Product feature name (e.g., search, support_agent)
team	string	Internal team owning the feature
environment	string	production, staging, dev, etc.
prompt_key	string	Identifier for the prompt template
prompt_version	string	Version of the prompt template
session_id	string	Groups one conversation. Used by content capture (below) to stitch a multi-turn exchange together; on request_logs it lands in custom tags.
turn_id	string	Groups one turn (which may span multiple model calls, e.g. a tool round-trip). For content capture, reuse the same value across those calls; if omitted, each captured row gets a fresh random one and will not group.
Any other key	string	Stored as custom tags, queryable in the dashboard

What Gets Tracked

Each event includes:

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

Field	Description
provider	"openai", "anthropic", "google", or "litellm"
model	Model name used
input_tokens	Prompt token count
output_tokens	Completion token count
cached_tokens	Cached prompt tokens (if applicable)
cost_usd_micros	Estimated cost in USD micros (1 USD = 1,000,000 micros)
latency_ms	End-to-end call duration
streaming	Whether the call was streamed
has_tools	Whether tool calls were used
has_vision	Whether image inputs were included
base_url	Provider endpoint host the call was sent to (captured automatically from the wrapped client; useful for gateways/proxies).
batch	true for rows recorded from a provider Batch API job (see Batch APIs), false/unset for live calls.

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

Field	Description
provider	"deepgram", "elevenlabs", or "livekit"
model	e.g. deepgram/nova-3-monolingual-prerecorded, elevenlabs/eleven_flash_v2_5, livekit/agent-session-minute, livekit/observability-event
modality	"audio_in" (STT), "audio_out" (TTS), "session" (LiveKit minute), or "event" (LiveKit observability)
audio_seconds	Billable audio duration for STT and LiveKit-session rows; recorded for analytics on TTS rows
characters	Synthesized character count on TTS rows (the ElevenLabs billing unit)
events	Count of MetricsCollectedEvent fan-out on the LiveKit observability-event row
cost_usd_micros	Estimated cost in USD micros, computed against the bundled pricing/models.json
session_id	Cross-provider correlation key (typically the LiveKit room name). NULL on standalone Deepgram/ElevenLabs rows.
idempotency_key	f"{session_id}:{kind}:{request_id_or_seq}" — server-side dedup via idx_request_logs_idempotency. Only set on LiveKit-shim rows; standalone calls are atomic.
routing_path	"direct" (default) or "livekit_inference" when using LiveKit's bundled inference. NULL on standalone rows.
latency_ms	Standalone wrap: full call duration (or stream completion). LiveKit shim: best-effort TTS ttft / STT duration in ms.

Batch APIs

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 retention-bounded.

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).