Python SDK

pip install opexia-trace

import opexia.trace

opexia.trace.init(
    org_id="acme",
    workspace_id="3f2b1c8e-9a4d-4f11-b7c2-1e5a8d3f6b90",
    project_id="checkout-agent",
    backend_url="https://ingest.opexia.dev",
    api_key="opx_live_xxxxxxxxxxxxxxxxxxxx",
    transport="direct",
)

export OPEXIA_TRANSPORT=direct

import os, opexia.trace

opexia.trace.init(
    org_id=os.environ["OPEXIA_ORG_ID"],
    workspace_id=os.environ["OPEXIA_WORKSPACE_ID"],
    project_id=os.environ.get("OPEXIA_PROJECT_ID", "default"),
    backend_url=os.environ.get("OPEXIA_INGEST_URL", "https://ingest.opexia.dev"),
    api_key=os.environ["OPEXIA_API_KEY"],
    transport=os.environ.get("OPEXIA_TRANSPORT", "direct"),
    fail_open=os.environ.get("ENV") == "production",
)

# .env
OPEXIA_ORG_ID=acme
OPEXIA_WORKSPACE_ID=3f2b1c8e-9a4d-4f11-b7c2-1e5a8d3f6b90
OPEXIA_PROJECT_ID=checkout-agent
OPEXIA_INGEST_URL=https://ingest.opexia.dev
OPEXIA_API_KEY=opx_live_xxxxxxxxxxxxxxxxxxxx
OPEXIA_TRANSPORT=direct

from opexia.trace import observe

@observe(reasoning_role="retrieval", node_type="retrieval")
def fetch_order(order_id: str) -> dict:
    return db.get(order_id)

@observe(reasoning_role="synthesis", node_type="agent", name="compose_answer")
async def compose(ctx: dict) -> str:
    return await llm.complete(ctx)

from opexia.trace import ReasoningTrace

with ReasoningTrace("answer_customer", reasoning_role="synthesis",
                    node_type="agent", end_user="user_8412") as node:

    node.record_decision(
        selected="lookup_order",
        scores={"lookup_order": 0.88, "search_kb": 0.31},
        alternatives=[{"name": "search_kb", "why_not": "no order id in message"}],
        rules_fired=["order_id_present"],
    )

    node.record_sources(
        consulted=["https://docs.internal/returns", "https://docs.internal/shipping"],
        used=["https://docs.internal/returns"],
        dropped=["https://docs.internal/shipping"],
        scores={"https://docs.internal/returns": 0.93},
    )

    node.record_cost(
        model="claude-opus-4-6",
        input_tokens=2_140,
        output_tokens=310,
    )

from opexia.trace import estimate_cost_usd, pricing_version

usd = estimate_cost_usd(model="claude-opus-4-6", input_tokens=2140, output_tokens=310)
print(f"${usd:.4f} (pricing {pricing_version()})")

opexia.trace.init(..., sampler_rate=0.1)   # 10% of traces

"""Minimal end-to-end instrumented agent."""
import os
import opexia.trace
from opexia.trace import ReasoningTrace, observe

opexia.trace.init(
    org_id=os.environ["OPEXIA_ORG_ID"],
    workspace_id=os.environ["OPEXIA_WORKSPACE_ID"],
    project_id="checkout-agent",
    backend_url="https://ingest.opexia.dev",
    api_key=os.environ["OPEXIA_API_KEY"],
    transport="direct",
    fail_open=os.environ.get("ENV") == "production",
)


@observe(reasoning_role="retrieval", node_type="retrieval")
def retrieve(query: str) -> list[str]:
    return ["https://docs.internal/returns", "https://docs.internal/shipping"]


def handle(query: str, user_id: str) -> str:
    # One ReasoningTrace per logical request = one trace.
    with ReasoningTrace("handle_request", reasoning_role="synthesis",
                        node_type="agent", end_user=user_id) as node:

        # Put the query on the EARLIEST span of the trace.
        node._span.set_attribute("opexia.query_text", query)

        docs = retrieve(query)
        node.record_decision(
            selected="answer_from_kb",
            scores={"answer_from_kb": 0.91, "escalate": 0.12},
            rules_fired=["kb_hit"],
        )
        node.record_sources(consulted=docs, used=docs[:1], dropped=docs[1:])

        answer = f"Returns are accepted within 30 days. ({len(docs)} sources)"

        node.record_cost(model="claude-opus-4-6", input_tokens=2140, output_tokens=310)
        # Put the outcome on the LATEST span of the trace.
        node._span.set_attribute("opexia.outcome_text", answer)
        return answer


if __name__ == "__main__":
    print(handle("Can I return this?", user_id="user_8412"))

"""verify_span.py — emit one span and confirm it landed."""
import os, time, uuid, httpx
import opexia.trace
from opexia.trace import ReasoningTrace

ORG   = os.environ["OPEXIA_ORG"]            # slug, for the read API
WS    = os.environ["OPEXIA_WORKSPACE"]      # slug, for the read API
WS_ID = os.environ["OPEXIA_WORKSPACE_ID"]   # UUID, for the SDK
KEY   = os.environ["OPEXIA_API_KEY"]

opexia.trace.init(
    org_id=ORG, workspace_id=WS_ID, project_id="verify",
    backend_url="https://ingest.opexia.dev", api_key=KEY, transport="direct",
)

marker = f"verify-{uuid.uuid4().hex[:8]}"
with ReasoningTrace("verify_span", reasoning_role="analysis", node_type="agent") as n:
    n._span.set_attribute("opexia.query_text", marker)
    n.record_cost(model="claude-opus-4-6", input_tokens=10, output_tokens=5)

# Force the batch out, then let ingestion settle.
from opentelemetry import trace as ot
ot.get_tracer_provider().force_flush()
print(f"emitted {marker}; waiting for ingestion…")
time.sleep(20)

# 1. Did anything dead-letter?
health = httpx.get(
    f"https://api.opexia.dev/v1/observ/orgs/{ORG}/workspaces/{WS}/health/ingestion",
    headers={"x-opexia-api-key": KEY}, timeout=30,
).json()
print("dead-lettered last hour:", health.get("dead_letter_count_last_hour"))

# 2. Is the trace queryable?
traces = httpx.get(
    f"https://api.opexia.dev/v1/observ/orgs/{ORG}/workspaces/{WS}/traces",
    headers={"x-opexia-api-key": KEY},
    params={"page_size": 20, "include_empty": True}, timeout=30,
).json()
# Payload is under "data", not "items".
print("recent traces:", len(traces["data"]))
assert traces["data"], "No traces — check key, workspace UUID, and ingest URL."
print("✅ span landed")