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.contentWorks 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"})| Platform | Helper | Required state |
|---|---|---|
| OpenAI / Azure OpenAI | track_openai_batch | status == "completed" |
| Anthropic | track_anthropic_batch | processing_status == "ended" |
| Gemini Developer API | track_google_batch | JOB_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()
| Parameter | Default | Description |
|---|---|---|
api_key | SP_API_KEY | SuperPenguin key |
base_url | https://app.superpenguin.ai | API endpoint |
flush_interval | 5.0 | Seconds between flushes |
batch_size | 50 | Max events per POST |
sp.wrap() highlights
| Parameter | Notes |
|---|---|
metadata | Default attribution bag |
tier | Deepgram Growth SKU when set to "growth" |
region / deployment_type / service_tier | Pricing axes; Bedrock auto-captures region |
Call sp.flush() before exit in short-lived scripts. An atexit handler also flushes on normal shutdown.