/docs/sdks/python

Python SDK

Python SDK for tracing AI agents, model calls, tool calls, metadata, media, metrics, and trace reads.

Installation

pip install theta-obsrv

Configuration

API keys are project-scoped on the server. THETA_PROJECTis optional and only needed when you want explicit project filtering.

export THETA_API_KEY="tk_live_…"
export THETA_BASE_URL="https://api.obsrv.tech"  # optional
export THETA_PROJECT="proj_…"                   # optional

Tracing

with client.trace(name="checkout", run_type="prod") as t:
    t.set_metadata(user_id="u_a72c", release="v3.2.1")

    with t.step(name="plan", type="llm", model="claude-sonnet-4-6") as s:
        s.log_message(role="user", text="Buy milk")
        s.log_message(role="assistant", text="On it.")
        s.set_token_usage(input=900, output=120)
        s.set_metadata(provider_cost_usd=0.014)

    with t.step(name="inventory.lookup", type="tool") as s:
        s.log_tool_call(
            name="inventory.lookup",
            arguments={"sku": "milk"},
            result={"in_stock": True},
            latency_ms=83,
        )

    t.attach_image("screenshot.png")

client.flush()

Agent wrapping

@client.wrap_agent("support-agent")
def agent(ctx, query):
    result = call_llm(query)
    ctx.on_complete(result)
    return result

result, run_id = agent("Help me cancel my order")
client.record_metric("task_adherence", run_id, passed=True)

Multimodal attachments

t.attach_image("./screenshot.png")
t.attach_audio(audio_bytes, mime="audio/wav")
t.attach_video("./recording.mp4")
t.attach_file("./report.pdf")
t.attach_sensor(depth_frame_bytes, modality="depth-camera")

Provider integrations

from openai import OpenAI
from theta_observability.integrations import wrap_openai

oai = wrap_openai(OpenAI(), client)
oai.chat.completions.create(model="gpt-4o-mini", messages=[...])

Batching and shutdown

The SDK queues traces and flushes either when the batch fills or after the configured flush interval. Short-lived scripts should flush before exit; long-lived workers can call close() during shutdown.

client.flush()
client.close()

Reading data back

result = client.list_traces(
    status=["error"],
    metadata_filters=[{"key": "release", "value": "v3.2.1"}],
    limit=50,
)

for trace in result.data:
    print(trace.trace_id, trace.status, trace.latency_ms)

detail = client.get_trace("tr_…")
print(detail.name, len(detail.steps))

Reference

TraceClient(api_key?, project?, base_url?, flush_interval?, max_batch?)
client.trace(name, run_type=..., user_id=..., metadata=...)
client.wrap_agent(name)
t.step(name, type=..., model=..., metadata=...)
s.log_message(role=..., text=..., images=..., attachments=...)
s.log_tool_call(name, arguments, result=..., error=..., latency_ms=...)
s.set_token_usage(input, output, total=None)
t.set_metadata(**kwargs) and t.set_metadata_path(dot_path, value)
client.record_metric(name, trace_id, passed=... | score=... | label=...)
client.list_traces(...) and client.get_trace(trace_id)
client.flush() and client.close()

← Orgs & projects

Node →