SuperPenguin Docs
SDKsPython

Python advanced usage

Tracing, batch jobs, content capture controls, LiveKit voice, and Python configuration reference.

Tracing with @sp.trace

Group multi-step pipelines under one parent trace:

@sp.trace(name="answer-question", metadata={"feature": "rag"})
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

Works with async functions. Nested wrapped calls inherit trace metadata unless overridden.

Per-call metadata overrides

Recommended: sp.metadata() (any provider, never touches the HTTP body):

with sp.metadata(customer_id=customer_id, prompt_key="summarize"):
    client.chat.completions.create(...)

OpenAI / Anthropic also accept extra_body={"sp_metadata": {...}} (stripped before send). LiteLLM accepts a metadata= kwarg. Precedence: per-call > @sp.trace > wrap defaults.

Batch API jobs

Native Batch APIs are async and file-based. sp.wrap() does not auto-capture them. Call a helper once the job succeeds:

batch = client.batches.retrieve("batch_abc")  # status == "completed"
count = sp.track_openai_batch(client, batch, metadata={"feature": "nightly-eval"})
PlatformHelperRequired state
OpenAI / Azure OpenAItrack_openai_batchstatus == "completed"
Anthropictrack_anthropic_batchprocessing_status == "ended"
Gemini Developer APItrack_google_batchJOB_STATE_SUCCEEDED

Rows are marked batch=true with deterministic idempotency keys. See Batch and service-tier pricing.

Content capture narrowing

Capture stays off until the org opts in from Settings. From the SDK you can only reduce capture:

sp.configure_content_capture(
    allow=True,
    record_prompt=True,
    record_outcome=True,
    sample_rate_multiplier=1.0,
)

Details: Content capture and privacy.

LiveKit voice agents

Use LiveKitObservability to stitch STT / TTS / session rows on one session_id. Keep the LLM on sp.wrap() for full token detail, and forward MetricsCollectedEvents to the shim. Call on_session_end() when the room ends.

Configuration

sp.init()

ParameterDefaultDescription
api_keySP_API_KEYSuperPenguin key
base_urlhttps://app.superpenguin.aiAPI endpoint
flush_interval5.0Seconds between flushes
batch_size50Max events per POST

sp.wrap() highlights

ParameterNotes
metadataDefault attribution bag
tierDeepgram Growth SKU when set to "growth"
region / deployment_type / service_tierPricing axes; Bedrock auto-captures region

Call sp.flush() before exit in short-lived scripts. An atexit handler also flushes on normal shutdown.