Metadata-Version: 2.5
Name: superpenguin
Version: 0.20.0
Summary: SuperPenguin Python SDK: AI cost management, attribution, and spend tracking
Project-URL: Homepage, https://superpenguin.ai
Project-URL: Documentation, https://superpenguin.ai/docs
License-Expression: MIT
License-File: LICENSE
Keywords: ai,anthropic,attribution,cost-management,deepgram,elevenlabs,gemini,google,litellm,livekit,llm,monitoring,observability,openai,spend-tracking,stt,tts,voice-ai
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Provides-Extra: anthropic
Requires-Dist: anthropic; extra == 'anthropic'
Provides-Extra: cohere
Requires-Dist: cohere<8,>=7.2; extra == 'cohere'
Provides-Extra: deepgram
Requires-Dist: deepgram-sdk>=3.0.0; extra == 'deepgram'
Provides-Extra: dev
Requires-Dist: anthropic; extra == 'dev'
Requires-Dist: openai; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Provides-Extra: elevenlabs
Requires-Dist: elevenlabs>=1.0.0; extra == 'elevenlabs'
Provides-Extra: google
Requires-Dist: google-genai; extra == 'google'
Provides-Extra: livekit
Requires-Dist: livekit-agents>=1.0.0; extra == 'livekit'
Provides-Extra: openai
Requires-Dist: openai; extra == 'openai'
Provides-Extra: otel
Requires-Dist: opentelemetry-api; extra == 'otel'
Provides-Extra: voice
Requires-Dist: deepgram-sdk>=3.0.0; extra == 'voice'
Requires-Dist: elevenlabs>=1.0.0; extra == 'voice'
Requires-Dist: livekit-agents>=1.0.0; extra == 'voice'
Description-Content-Type: text/markdown

# 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`](../../docs/vs-native-attribution.md) for the full breakdown of what OpenAI / Anthropic / Deepgram / ElevenLabs offer natively vs. what SuperPenguin adds on top.

## Installation

```bash
pip install superpenguin
```

For OpenTelemetry trace correlation:

```bash
pip install "superpenguin[otel]"
```

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

```bash
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.

```python
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:

```python
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)

```python
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

```python
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`](https://github.com/googleapis/python-genai) client, which targets either the Gemini API (AI Studio) or Vertex AI:

```python
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:

```python
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.

`invoke_model` and `invoke_model_with_response_stream` are wrapped too (the JSON body is read for content capture). Bedrock calls take no per-call `extra_body`, so attribute with wrap-level `metadata`, `sp.metadata()` or `@sp.trace`.

### 4. Streaming works transparently

```python
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](https://github.com/BerriAI/litellm) to call 100+ LLM providers through a single interface, one call patches everything:

```python
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!"}],
)
```

Calling a **LiteLLM proxy** instead? Wrap an OpenAI client pointed at it. Pass
`provider="litellm"` unless the proxy's host name contains "litellm" (then it is
detected automatically). The call is recorded under the `litellm` route with
LiteLLM's own per-request cost (the `x-litellm-response-cost` header, or
`usage.cost` on the last chunk of a stream):

```python
from openai import OpenAI

client = sp.wrap(
    OpenAI(api_key="sk-litellm-...", base_url="http://localhost:4000/v1"),
    provider="litellm",
)
```

### 6. Add attribution metadata

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

```python
# 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.

```python
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, 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.

```python
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](https://docs.livekit.io/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:

```python
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()` | n/a |
| `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 catalogued Deepgram / ElevenLabs SKUs. Override per session via the constructor if you're on a different SKU:

```python
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

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

```python
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

```python
@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:

```python
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` | Ingest API endpoint |
| `spend_base_url` | `str` | `https://spend.superpenguin.ai` | Optional spend-read endpoint override |
| `flush_interval` | `float` | `5.0` | Seconds between background batch flushes |
| `batch_size` | `int` | `50` | Max events per batch POST |
| `spend_request_timeout` | `float` | `3.0` | Timeout in seconds for each spend read |

### Environment variables

| Variable | Description |
|----------|-------------|
| `SP_API_KEY` | API key (used if not passed to `init()`) |
| `SP_BASE_URL` | Ingest API base URL override |
| `SP_SPEND_BASE_URL` | Spend-read-only 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 pending usage events and content samples. Call it before exit in short-lived scripts, jobs and serverless handlers:

```python
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 (1 USD = 1,000,000 micros). Client may send 0; the server prices from the live catalog. |
| `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. |

## Manual usage reporting (custom agent harnesses)

If your agent calls a model through its own HTTP code or a framework that
SuperPenguin does not wrap, report each completed model call yourself with
`sp.track_usage(...)` (Python SDK 0.20.0 or later). Pass the request and
response bodies too, and the call can be sampled for content capture exactly
like a wrapped call:

```python
import superpenguin as sp

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

def on_model_call(body: dict, reply: dict, latency_ms: int) -> None:
    """Called by your harness after each model call returns."""
    sp.track_usage(
        provider="anthropic",               # the provider that bills the call
        model=reply["model"],
        usage=reply["usage"],               # input_tokens / output_tokens (or prompt_tokens / completion_tokens)
        latency_ms=latency_ms,
        idempotency_key=f"{run_id}:{step}", # optional: a retried report is stored once
        metadata={
            "session_id": run_id,           # one agent run or conversation
            "turn_id": turn_id,             # one turn; reuse it across that turn's tool round-trips
            "customer_id": customer_id,
            "feature": "research_agent",
            "prompt_key": "planner",
        },
        request=body,                       # the exact JSON body sent to the provider
        response=reply,                     # the full response body
    )
```

- The usage event is submitted first and is identical with or without
  `request` / `response`; spend never depends on capture.
- `request` and `response` are optional. They are read only when your
  organization has content capture on and the call is sampled (same rules as
  `sp.wrap()`: org setting, `prompt_key` and `session_id` sampling); an
  unsampled call is never serialized. A sampled call goes through the same
  pipeline as a wrapped one: credentials dropped, binary stripped, redaction,
  nothing truncated.
- `metadata` outranks an active `sp.metadata()` / `@sp.trace` scope, which
  fills any field you leave out.
- **Do not both wrap a client and report the same call manually.** That
  counts its usage twice. Use `sp.track_usage` only for calls no wrapped
  client or `sp.patch_litellm()` already tracks.
- Capture runs on a background thread. In a short-lived process (a script,
  a job, a serverless handler) call `sp.flush()` before it exits.

## 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`:

```python
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.

### Batch and flex results with prompt samples

`submit()` and `track_*_batch` record usage only, so batch work produces no
prompt samples. If you already loop over results yourself (for example Sail
batch or flex jobs), call `sp.track_batch_result` instead of `submit()`. It
submits the usage dict you already build, unchanged, and also samples the
request and response the way `sp.wrap()` does (when your organization has
content capture on):

```python
import superpenguin as sp

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

for line in output_lines:                     # the job's output JSONL, parsed
    custom_id = line["custom_id"]
    event = build_usage_event(line)           # the dict you pass to submit() today
    sp.track_batch_result(
        request=request_lines[custom_id]["body"],  # the matching input line's body
        response=line["response"]["body"],
        usage_event=event,
    )
sp.flush()
```

The whole request body and the whole response body are captured, in any
provider shape (Chat `messages`, Responses `instructions` / `input` /
`output`, and others). The sample is stamped with the usage
event's `created_at` (or `completed_at=` when the event has none), carries its
feature and other attribution, and reporting the same result twice stores one
sample. Without `usage_event`, pass `provider=`, `batch_id=` and the line's
`custom_id` (taken from whole input or output lines automatically) and the SDK
builds the same event `track_openai_batch` would.

### Batch prompt samples without storing prompts (0.19.0)

Batch output files hold only responses, so the call above means keeping every
prompt until its result returns. Instead, send each half when you already have
it, and the server joins them on the line's idempotency key:

```python
import superpenguin as sp

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

# At submit time, once the batch id is known: the prompt half.
for line in input_lines:
    sp.track_batch_request(
        f"{batch.id}:{line['custom_id']}",    # the key the usage event will carry
        line,                                  # the input line (its body is read)
        {"feature": "ai_extractor"},
        provider="sail_research",
    )

# On completion: usage, plus the output half. No request needed.
for line in output_lines:
    event = build_usage_event(line)            # idempotency_key = f"{batch.id}:{custom_id}"
    sp.track_batch_result(usage_event=event, response=line["response"]["body"])
sp.flush()
```

Only sampled lines send anything. Both calls hash the key against your
organization's stable batch rate, so they agree without shared state, and an
unsampled line reads no text and makes no network call. The usage event is
submitted unchanged, exactly as with `submit()`. The same opt-in, redaction
and encryption apply as for every sample. A prompt half whose result never
arrives is discarded after 14 days.

## 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 calls **automatically**, there is nothing to call to turn it on. Samples are encrypted at rest and deletable by the org owner from Settings.

**What a sample holds (0.20.0).** A sampled call is captured in full and nothing is truncated: the complete request as you passed it (every message in order with its role, system and developer instructions, tool definitions, tool calls and tool results, thinking blocks and every model parameter) and the complete response (every choice, candidate or output item, text, tool calls, reasoning summaries and thinking text, stop reasons, usage and ids). Streamed calls are captured as the full response rebuilt from the stream. Transport settings and credentials (`api_key`, `extra_headers`, `timeout`, `http_client` and similar) are left out. Images, audio, files and other binary data are replaced by `[STRIPPED_BINARY:<bytes>]`, and opaque provider blobs (Anthropic thinking signatures, redacted thinking, Responses encrypted reasoning) by `[STRIPPED_OPAQUE:<chars>]`. The only size limit is a 64 MiB safety ceiling per sample; a call above it sends no content.

Before a sample is sent, the SDK replaces secrets (provider API keys, tokens, private keys, Azure and Google Cloud keys and signed-link signatures, passwords in URLs and `password=`-style assignments, secret-named tool arguments) with `[REDACTED_SECRET]`, and emails, phone numbers, payment cards, US SSNs, IBANs and public IP addresses with `[REDACTED_EMAIL]`, `[REDACTED_PHONE]`, `[REDACTED_CARD]`, `[REDACTED_SSN]`, `[REDACTED_IBAN]` and `[REDACTED_IP]`. This runs on every string in the request and the response, of any length, with nothing cut. Names, addresses and metadata values are not redacted. The exact list is in the [content capture docs](https://superpenguin.ai/docs/concepts/content-capture#what-is-redacted).

### Upgrading to full capture (0.20.0)

Upgrade the package (`pip install -U "superpenguin>=0.20.0"`). No code change
is needed for wrapped clients; check your integration style below.

| Integration | What to do |
|-------------|------------|
| `sp.wrap(client)` (OpenAI, Anthropic, Gemini, Bedrock, OpenAI-compatible gateways) | Nothing. Full requests and responses are captured, streams included. Set `session_id` / `turn_id` (below) to group calls. |
| `sp.patch_litellm()` | Nothing. Put `session_id` / `turn_id` in `sp.metadata()` or in LiteLLM's per-call `metadata={...}`. |
| Batch and flex jobs | `track_*_batch` helpers record usage only. For samples use `track_batch_request` / `track_batch_result` ([above](#batch-apis)); `session_id` / `turn_id` come from their `metadata` or the usage event. |
| Custom agent harness reporting usage itself | Use `sp.track_usage(..., request=body, response=reply)` ([above](#manual-usage-reporting-custom-agent-harnesses)). Do not also wrap that client. |

Each stored document is `{"request": ...}` / `{"response": ...}` followed by a
`blocks` index: one entry per system prompt, tool list, message and output
with its path and a SHA-256 fingerprint, computed after redaction. It needs
nothing from you.

Capture runs in the background. Long-running services need nothing more; a
short-lived process should call `sp.flush()` before exit (an `atexit` handler
also flushes on normal interpreter shutdown, but not on `os._exit`, a killed
container or a frozen serverless runtime).

### 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:

```python
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 request holds the whole conversation you sent (including earlier tool calls and their results) and the response holds the model's reply and any new tool calls. If you omit `turn_id`, each row gets a fresh random one and will not group.

`session_id` and `turn_id` are read from every metadata source, with the same precedence as other attribution (most specific wins): `sp.wrap(client, metadata={...})` defaults, an `sp.metadata()` block or `@sp.trace(metadata=...)`, then per-call `extra_body={"sp_metadata": {...}}` (or LiteLLM's `metadata={...}`), and `metadata=` on `track_usage` and the batch helpers. Python SDK versions before 0.20.0 dropped them from samples when they were set per call; upgrade if your samples show no `session_id`.

### 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):

```python
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,        # extra str -> str rules, run on every string after the built-in redaction
    max_queue_bytes=32 * 1024 * 1024,  # memory for samples waiting to upload (gzip bytes)
)
```

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

### Cost to your application

Capture never slows down or changes your calls, and usage and spend tracking are unaffected:

- Unsampled calls are never serialized. The sampling decision runs first, on your thread.
- A sampled call costs your thread a shallow copy of the request. Reading, redacting, serializing and compressing happen on a background thread.
- Memory is bounded: at most 32 sampled calls wait for processing, and at most `max_queue_bytes` (default 32 MiB, gzip) and 200 samples wait to upload. When either is full the sample is dropped, never blocking your call. (The batch helpers above, which run in your reporting loop rather than on a request, wait up to 5 seconds for room first.)
- Uploads are gzip-compressed; a sample too large for one request (over 3.5 MiB compressed) is sent in parts.
- `sp.content_capture_stats()` returns the counts (`queued`, `sent`, `dropped_queue_full`, `dropped_sample_ceiling`, `serialize_errors`, and more), and drops are reported to SuperPenguin with the next upload.

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

```python
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](https://superpenguin.ai/docs/sdk/python/providers#cohere)
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](https://github.com/carrotlabsai/cost-management/pull/815), and package
publication is recorded in [PR #819](https://github.com/carrotlabsai/cost-management/pull/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:

```python
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.
