Python providers

Wrap OpenAI, Azure OpenAI, Anthropic, Gemini, Bedrock, LiteLLM, Deepgram, ElevenLabs, and OpenAI Realtime with the Python SDK.

Pass your native client to sp.wrap(). The provider is auto-detected.

For OpenRouter, you can instead configure Broadcast to hosted OTLP. Do not wrap and Broadcast the same calls unless both paths carry the same generation ID.

OpenAI

from openai import OpenAI, AsyncOpenAI

client = sp.wrap(OpenAI())
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

async_client = sp.wrap(AsyncOpenAI())

Streaming works transparently. The cost event is submitted when the stream completes.

Wrap AzureOpenAI (or an OpenAI client pointed at an Azure endpoint). SuperPenguin detects Azure from the host (*.openai.azure.com, *.services.ai.azure.com, and related Azure AI hosts). There is no separate Azure wrapper.

Azure region is not auto-captured. Typical Azure hosts use a resource name, not a region, so pass wrap-level region and deployment_type if you want SDK estimates to use the matching rate card. Omit both for Global Standard (the default).

from openai import AzureOpenAI

client = sp.wrap(
    AzureOpenAI(
        azure_endpoint="https://my-resource.openai.azure.com",
        api_key="...",
        api_version="2024-10-21",
    ),
    region="eastus2",
    deployment_type="regional",
)

Deployment	What to pass
Global Standard	omit `region` and `deployment_type`
Data Zone	`deployment_type="data-zone"`
Regional	`region` (location code, e.g. `eastus2`) and `deployment_type="regional"`

region must be the Azure location code from the portal / ARM (eastus2, westeurope, swedencentral). Case does not matter (EastUS2 works). Display names (East US 2) and AWS-style ids (us-east-2) will not match, so the estimate falls back to Global. For deployment_type, use exactly "regional" or "data-zone" (keep the hyphen). Passing only region is not enough for regional rates. Connecting Azure for billed spend already uses invoice amounts; these fields only affect SDK estimates. See Batch and service-tier pricing.

Doubleword

Wrap an OpenAI client pointed at https://api.doubleword.ai/v1. SuperPenguin attributes the billed route as doubleword (the model maker stays in model_vendor). Scope custom prices to doubleword.

Anthropic

from anthropic import Anthropic

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

Google Gemini (AI Studio or Vertex AI)

from google import genai

client = sp.wrap(genai.Client(api_key="..."))
# Or Vertex: 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 (sync and async). Tiered Gemini pricing is applied server-side from input token count.

AWS Bedrock

import boto3

client = sp.wrap(
    boto3.client("bedrock-runtime", region_name="us-west-2"),
    metadata={"customer_id": "cust_acme_123", "feature": "doc_summary"},
    deployment_type="global",  # optional cross-region / global profiles
)

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

Wrapped methods: converse and converse_stream (boto3 sync / aioboto3 async). The wrapper sends usage only; the server prices rows with the aws_bedrock rate card. Region is auto-captured from meta.region_name or the endpoint host. invoke_model is not wrapped in v1.

LiteLLM

import litellm

sp.patch_litellm(metadata={"environment": "production"})
response = litellm.completion(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

Every litellm.completion() / litellm.acompletion() call is tracked after patching. SuperPenguin captures billed cost when LiteLLM includes it on the response. You do not pass region.

Deepgram

from deepgram import DeepgramClient, PrerecordedOptions

dg = sp.wrap(
    DeepgramClient(api_key="..."),
    metadata={"feature": "podcasts"},
    tier="growth",  # optional Growth-plan SKU
)

result = dg.listen.rest.v("1").transcribe_url(
    {"url": "https://example.com/episode.mp3"},
    PrerecordedOptions(model="nova-3", multilingual=True),
)

Rows include audio_seconds and estimated cost.

ElevenLabs

from elevenlabs.client import ElevenLabs

el = sp.wrap(ElevenLabs(api_key="..."), metadata={"feature": "ivr-greeting"})
audio = el.text_to_speech.convert(
    voice_id="21m00Tcm4TlvDq8ikWAM",
    text="Welcome to SuperPenguin",
    model_id="eleven_flash_v2_5",
)

OpenAI Realtime API

Realtime is a WebSocket event stream, not a create() call. Wrap the connection:

client = OpenAI()

with client.realtime.connect(model="gpt-realtime") as conn:
    conn = sp.wrap_realtime(conn, metadata={"feature": "voice-agent"})
    conn.response.create()
    for event in conn:
        ...  # usage tracked on response.done

Each completed turn emits input/output tokens with audio portions as separate input_audio_tokens / output_audio_tokens legs. For WebRTC or raw sockets, use sp.track_realtime_event(event, model="gpt-realtime").