Python providers

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

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.

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

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