superpenguin 0.2.2

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

SuperPenguin Python SDK

Track AI costs automatically. Wrap your OpenAI, Anthropic, or Google Gemini client — or patch litellm — and every LLM call is captured with token counts, estimated cost, latency, and attribution metadata. No proxy required.

Installation

pip install superpenguin

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.

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",
        }
    },
)

@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://api.carrotlabs.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	required	The LLM client to wrap
name	str	None	Override the default event name
metadata	dict	None	Default metadata for every call (customer_id, feature, team, etc.)
tags	list[str]	None	Tags added to every event

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
Any other key	string	Stored as custom tags, queryable in the dashboard

What Gets Tracked

Each event includes:

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

Never captured: Prompt content, response content, images, audio, tool arguments, or function results. The SDK only captures cost-relevant metadata.