indratrace 0.2.0

OpenTelemetry-native observability SDK for the IndraTrace platform — one-line traces, logs, metrics, and model-call token usage for web apps and AI agents.

IndraTrace SDK

OpenTelemetry-native observability SDK for the IndraTrace platform — one-line instrumentation for web apps and AI agents: traces, logs, metrics, and model-call token usage.

pip install indratrace

Two words to know up front:

• A span is one timed step — a web request, a database call, one model call.
• A trace is the full story of one request — its spans stacked on a timeline, so you can see where the time went and what called what.

init_observability() ships traces, logs, and metrics; trace_agent / trace_tool wrap your agents and tools; and model spans carry exact, provider-reported token counts when the GenAI extras are installed. Token counts are recorded raw — the SDK never computes cost; the platform derives it at query time.

import logging

from fastapi import FastAPI

from indratrace import init_observability, trace_agent, trace_tool

# Once, at app startup.
init_observability(product="my-app", env="prod", ingest_key="...")

app = FastAPI()  # every HTTP request becomes a span, automatically


@trace_tool  # a span per tool call
async def risk_score(vendor: str) -> int:
    logging.getLogger(__name__).info("scoring %s", vendor)  # ships with trace context
    return len(vendor)


@trace_agent("compliance-checker")  # a span wrapping the whole agent request
async def run(query: str) -> int:
    return await risk_score(query)

The import logging above is just Python's built-in logging — nothing IndraTrace-specific, and not required. Whatever your app already logs ships automatically (see Configuration); the line is there to show a log call picking up its span's trace context.

Both decorators work on sync and async functions. They are transparent: a tool that raises gets its span marked ERROR with the exception recorded, and the exception then propagates to your code unchanged.

Tracing a regular REST API

You don't need to be building an AI agent. For a plain FastAPI service, the one init line is the whole setup — every endpoint is reported as a span, with its status and duration, no per-route code:

from fastapi import FastAPI

from indratrace import init_observability

init_observability(product="orders-api", env="prod", ingest_key="...")

app = FastAPI()


@app.get("/orders/{order_id}")            # this endpoint is now a span automatically
def get_order(order_id: str) -> dict:
    return {"id": order_id}

When one endpoint is slow and you want to see inside it — which query, which parser ate the time — wrap that piece in @trace_step. It adds a child span under the request so the timeline shows the breakdown:

from indratrace import trace_step


@trace_step                               # a span named "step load_order"
def load_order(order_id: str) -> dict:
    ...                                   # e.g. a database query
    return {"id": order_id}

@trace_step is the neutral sibling of @trace_tool: same behavior, but for timing ordinary functions (database queries, parsers, validation) where calling them a "tool" would be misleading. Bare or called (@trace_step()), sync or async, exceptions recorded and re-raised unchanged.

Install the extras you need — FastAPI for HTTP auto-instrumentation, and anthropic / openai / gemini / bedrock for model spans with token usage:

pip install "indratrace[fastapi,anthropic,openai,gemini,bedrock]"

Token usage from model calls

With the anthropic, openai, gemini, or bedrock extra installed, every provider call made after init_observability() produces a model span carrying the exact, provider-reported token counts (gen_ai.usage.input_tokens / gen_ai.usage.output_tokens), nested under whatever agent/tool span is active. No wrapper, no config — just call the provider as you already do:

import anthropic

from indratrace import init_observability, trace_agent, trace_tool

init_observability(product="my-app", ingest_key="...")
client = anthropic.Anthropic()


@trace_tool
def summarize(doc: str) -> str:
    msg = client.messages.create(               # model span with token counts,
        model="claude-haiku-4-5",               # a child of this tool span
        max_tokens=256,
        messages=[{"role": "user", "content": doc}],
    )
    return msg.content[0].text


@trace_agent("summarizer")
def run(doc: str) -> str:
    return summarize(doc)

Streaming calls are captured too — usage lands on the span from the final streamed event.

For a provider the SDK does not auto-instrument, stamp the counts yourself from inside a span with record_llm_usage:

from indratrace import record_llm_usage

record_llm_usage(
    model="some-model-v2",
    input_tokens=resp.usage.input,
    output_tokens=resp.usage.output,
    system="acme-ai",
)

Token counts are stored raw — the SDK never computes cost; the platform derives it at query time from a price table.

Capturing prompt & completion text

By default, model spans carry token counts but not the prompt or completion text — because prompts often contain customer data. Turn the text on when you want to see exactly what was sent and returned (the usual case in dev and staging, off in production):

init_observability(product="my-app", ingest_key="...", capture_content=True)

Or set INDRATRACE_CAPTURE_CONTENT=true in the environment (an explicit capture_content= argument wins over it). When on, the prompt lands on the model span under gen_ai.input.messages and the completion under gen_ai.output.messages. This flag only gates the text — token counts are captured either way.

Session & user context

Wrap a conversation in session(...) and every span started inside it — your agent/tool spans, the FastAPI HTTP span, and the GenAI model spans — carries session.id and/or user.id. No per-call wiring: the ids ride OTel baggage and a span processor stamps them at span start.

from indratrace import session

with session(session_id="conversation-42", user_id="u-1001"):
    answer = run(query)          # every span here is tagged with both ids

It works across async/await and threads, and nests — an inner session(user_id=...) overrides only user.id and keeps the outer session.id. For middleware that can't bracket a with (it tags on request-in and untags on request-out, in separate callbacks), call it imperatively and keep the handle:

handle = session(session_id=request.headers["x-session-id"])
try:
    ...                          # dispatch the request
finally:
    handle.detach()             # or handle.close(); restores the prior context

Feedback (👍 / 👎)

Tie a user's thumbs-up/down back to the trace that produced the answer. Capture the trace id at answer time with current_trace_id(), hand it back to the caller, and record the score whenever the user reacts — often minutes later, out of band:

from indratrace import current_trace_id, record_feedback

@trace_agent("assistant")
def answer(query: str) -> dict:
    text = run(query)
    return {"answer": text, "trace_id": current_trace_id()}  # store this id

# later, when the user clicks 👍
record_feedback(1, comment="spot on", trace_id=stored_trace_id)

score is any number — the convention is 1 for positive, 0/-1 for negative, but any scale (e.g. 1–5) works. If you omit trace_id, the current trace's id is used when you're inside one. record_feedback emits a short feedback span carrying feedback.score, the optional feedback.comment, and feedback.trace_id, which the platform joins back to the original trace. Called inside session(...), the feedback span carries the session/user ids too.

Bring your own backend

The SDK emits standard OTLP over HTTP — nothing IndraTrace-specific on the wire. Point endpoint= (or INDRATRACE_ENDPOINT) at any OTLP receiver and the telemetry flows there, no ingest key required outside the IndraTrace platform:

# Your own OpenTelemetry Collector, Jaeger, Grafana (Tempo/Alloy), SigNoz, …
init_observability(product="my-app", endpoint="http://otel-collector:4318")

export INDRATRACE_ENDPOINT="http://localhost:4318"   # e.g. a local Jaeger all-in-one

The x-indratrace-key header is only sent when you set a key (ingest_key= / INDRATRACE_KEY), which the hosted IndraTrace platform uses to authenticate ingest. Your own collector doesn't need it — leave it unset.

Configuration

Resolution order is explicit arg > env var > default:

Parameter (init_observability(...))	Env var	Default
product	INDRATRACE_PRODUCT	required — raises/warns if unset
env	INDRATRACE_ENV	dev
ingest_key	INDRATRACE_KEY	none (no auth header sent)
endpoint	INDRATRACE_ENDPOINT	http://localhost:4318
capture_content	INDRATRACE_CAPTURE_CONTENT	false (token counts only, no prompt/completion text)

Your existing logging calls ship automatically once your app is at INFO — the usual case under basicConfig(level=INFO), uvicorn, or gunicorn. The SDK does not change your root logger's level on its own; if your app never configured logging (so it sits at the stdlib default of WARNING), pass log_level="INFO" to opt in:

init_observability(product="my-app", ingest_key="...", log_level="INFO")

The SDK never raises into your app: if the collector is unreachable or the config is wrong, it logs one warning and runs un-instrumented. The decorators hold to that too — they run your function even when init_observability() was never called.

Built by Indrasol.