rudderstack-ai

Product analytics for the AI features in your Python backend. rudderstack-ai wraps your OpenAI or Anthropic client, LangChain / LangGraph / Deep Agents runs, or Claude Agent SDK queries, and sends standard AI events (spec v0.2) through the RudderStack client you already use.

• One event per answer. However many model calls, agents or tool steps an answer takes, you get one ai_response_received with its totals: tokens, cost, latency, models, tool calls, status.
• Joinable. The prompt, the answer and later feedback share an interaction_id, and every event carries conversation_id. Your app adds intent and outcome, so you can build audiences such as "users whose refund questions went unresolved".
• Never in the way. The original client is not modified, tracking errors are swallowed, and provider errors reach your code unchanged.
• Private by default. Prompt, answer and feedback text are sent only if you turn on content capture.
• Traces too, if you want them. rudderstack.ai.tracing sends your OpenTelemetry AI spans as ai_otel_span events through the same client, joined to the product events by trace_id and interaction_id.

Server-side only: events are emitted where the model calls happen. The Node.js counterpart is @rudderstack/ai; both emit identical events.

Use cases

One package, three ways to use it. Each costs nothing for the others: OpenTelemetry is installed and imported only with the tracing extra.

Product events only. One ai_response_received per answer, from a wrapped client.

# pip install "rudderstack-ai[openai]"
rudderstack.ai.configure(analytics=rudder_analytics)
client = wrap_openai(OpenAI())
with ai_interaction(user_id=user.id, conversation_id=chat.id):
    client.chat.completions.create(model="gpt-4.1-mini", messages=messages)

Tracing only. Any OpenTelemetry instrumentation; each AI span becomes one ai_otel_span event (see Tracing).

# pip install "rudderstack-ai[tracing]" opentelemetry-instrumentation-openai-v2
from rudderstack.ai.tracing import register_tracing

tracing = register_tracing(rudder_analytics)
OpenAIInstrumentor().instrument(tracer_provider=tracing.tracer_provider)

Both, joined. Spans started inside an interaction carry its interaction_id and identity, and the interaction's ai_response_received.trace_id is the spans' trace id.

# pip install "rudderstack-ai[anthropic,tracing]" opentelemetry-instrumentation-anthropic
rudderstack.ai.configure(analytics=rudder_analytics)
tracing = register_tracing()  # client and capture_content from configure()
AnthropicInstrumentor().instrument(tracer_provider=tracing.tracer_provider)
client = wrap_anthropic(Anthropic())
with ai_interaction(user_id=user.id, conversation_id=chat.id):
    client.messages.create(model="claude-haiku-4-5", max_tokens=1024, messages=messages)

Contents

• Install
• Quick start
• How it works
• Background agents
• Configuration
• Events
• Privacy and content capture
• Cost
• Adoption tag: context.wrapperLibrary
• Tracing
• Compatibility
• Limitations
• Examples

Install

Python 3.12 or later. Install the extra for each library you use:

pip install "rudderstack-ai[openai]"            # or: uv add "rudderstack-ai[openai]"
pip install "rudderstack-ai[anthropic]"
pip install "rudderstack-ai[langchain]"         # LangChain, LangGraph, Deep Agents
pip install "rudderstack-ai[claude-agent-sdk]"
pip install "rudderstack-ai[tracing]"           # OpenTelemetry spans (rudderstack.ai.tracing)
pip install "rudderstack-ai[all]"

The package has no dependencies of its own; the tracing extra adds opentelemetry-api, opentelemetry-sdk and opentelemetry-exporter-otlp-proto-common. It sends events through your RudderStack Python SDK client (rudder-sdk-python) and never creates one. Both packages share the rudderstack namespace: rudderstack.analytics and rudderstack.ai.

Quick start

Configure once at startup with your existing client, either the module or a Client instance:

import rudderstack.analytics as rudder_analytics
import rudderstack.ai

rudder_analytics.write_key = "<WRITE_KEY>"
rudder_analytics.dataPlaneUrl = "<DATA_PLANE_URL>"

rudderstack.ai.configure(analytics=rudder_analytics)

If the events load into a warehouse, also set up the tracking plan and cost column types.

OpenAI

chat.completions.create / parse / stream and responses.create / parse / stream, sync and async, streaming or not.

from openai import OpenAI
from rudderstack.ai import ai_context, ai_interaction
from rudderstack.ai.openai import wrap_openai

client = wrap_openai(OpenAI())

with ai_context(user_id=user.id, conversation_id=chat.id, feature="support_bot"):
    with ai_interaction(intent="refund_request") as interaction:
        reply = client.chat.completions.create(model="gpt-4.1-mini", messages=messages)
        interaction.set_outcome("resolved")

return {"answer": reply.choices[0].message.content, "interaction_id": interaction.id}

AzureOpenAI and OpenAI-compatible endpoints work the same way; pass provider="azure.ai.openai" (or openrouter, vllm...) to label them.

Anthropic

messages.create (including stream=True), messages.parse and the messages.stream() helper, plus the same under beta.messages; sync and async.

from anthropic import Anthropic
from rudderstack.ai import ai_interaction
from rudderstack.ai.anthropic import wrap_anthropic

client = wrap_anthropic(Anthropic())

with ai_interaction(user_id=user.id, conversation_id=chat.id, feature="writer"):
    with client.messages.stream(model="claude-haiku-4-5", max_tokens=1024, messages=messages) as stream:
        for text in stream.text_stream:
            print(text, end="")

For AnthropicBedrock or AnthropicVertex, pass provider="aws.bedrock" or provider="gcp.vertex_ai".

LangChain, LangGraph and Deep Agents

One callback handler. Subagent usage is attributed by agent name.

from rudderstack.ai.langchain import RudderStackCallbackHandler

handler = RudderStackCallbackHandler()

result = agent.invoke(
    {"messages": messages},
    config={
        "callbacks": [handler],
        "metadata": {"user_id": user.id, "conversation_id": chat.id, "intent": "refund_request"},
    },
)

Without an active ai_interaction, the handler opens one per root run and reads identity from the run metadata (user_id, anonymous_id, conversation_id, interaction_id, feature, intent, trace_id). Inside an ai_interaction, it reports into that interaction instead.

Claude Agent SDK

tracked_query is a drop-in for claude_agent_sdk.query(). Cost and per-model usage come from the SDK's final ResultMessage.

from rudderstack.ai import ai_context
from rudderstack.ai.claude_agent_sdk import tracked_query

with ai_context(user_id=user.id, conversation_id=chat.id):
    async for message in tracked_query(prompt=question, options=options, feature="ops_agent"):
        ...

With ClaudeSDKClient, wrap its response stream: async for message in tracked_messages(client.receive_response()): ..., or map a result yourself with record_claude_result(result_message).

Multi-turn sessions. On a resumed session (options.resume) and on every later turn of a ClaudeSDKClient, the SDK reports model_usage and total_cost_usd for the whole session so far. Store usage_snapshot(result_message) with the session and pass it to the next turn, so each event covers that turn alone:

from rudderstack.ai.claude_agent_sdk import tracked_query, usage_snapshot

async for message in tracked_query(
    prompt=question,
    options=ClaudeAgentOptions(resume=session.id),
    usage_baseline=session.usage,  # None on the first turn
):
    if isinstance(message, ResultMessage):
        session.usage = usage_snapshot(message)  # JSON-serialisable

Without a baseline, a resumed tracked_query counts the turn's tokens from the turn itself and prices them with your registered prices; with no price registered it sends no cost rather than the session's. Subagents' output tokens can then be undercounted (the SDK streams their messages with the stream-start count), and models the SDK only restored from the transcript are left out.

Feedback

Return interaction.id with the answer, and send feedback from your server:

from rudderstack.ai import track_action

track_action(
    "feedback_given",
    interaction_id=body.interaction_id,
    conversation_id=body.conversation_id,
    user_id=user.id,
    action_details={"feedback_type": "thumbs", "feedback_value": False},
)

Action types: feedback_given, feedback_retracted, copied_response, shared, reported, regenerated. When a user takes their feedback back, send feedback_retracted with the same feedback_type (and the retracted feedback_value, for context): it clears the interaction's feedback, so the latest feedback_given or feedback_retracted per interaction_id is the final state. conversation_id is required (from the argument or the current ai_context); without it the event is dropped with a warning.

How it works

• ai_context sets identity once per request: user_id, anonymous_id, conversation_id, feature, and optionally a client and extra track context. It uses contextvars, so concurrent requests and asyncio tasks don't mix. Nested contexts inherit what they don't set.
• ai_interaction is one prompt and one answer. Every instrumented model call inside it is aggregated, and ai_response_received is sent when the block exits. An exception leaving the block sets the status (error, timeout or cancelled) and is re-raised unchanged.
• Adapters (wrap_openai, wrap_anthropic, the LangChain handler, tracked_query) only report calls into the active interaction. Calls are de-duplicated on the provider response ID, so a wrapped client used by LangChain with the handler attached is counted once.
• Calls outside an interaction still produce a valid event of their own: trigger: "user" when identity is in context, otherwise trigger: "system" with anonymousId: "ai_system" and a minted conversation_id. The SDK never invents an anonymousId when a userId is set: forward the browser SDK's anonymousId yourself to stitch profiles. No prompt event is sent for them.
• The prompt event is held until the first model call starts (or the interaction ends), so it goes through the same client and carries the same integration tag as the answer. It keeps its original timestamp.

Streaming to the browser

When a handler returns a stream, the interaction must stay open until the stream ends. Pass the generator through interaction.bind():

from fastapi.responses import StreamingResponse

with ai_interaction(user_id=user.id, conversation_id=chat.id) as interaction:

    async def tokens():
        stream = await client.chat.completions.create(model="gpt-4.1-mini", messages=messages, stream=True)
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

    body = interaction.bind(tokens())
return StreamingResponse(body, headers={"x-interaction-id": interaction.id})

A provider stream that is closed or abandoned before it completes counts as a cancelled call, so the answer is cancelled when that was the last call; a stream passed to bind() and closed early by the client ends the interaction as cancelled. For OpenAI chat streams, the wrapper asks for usage in the final chunk (stream_options.include_usage) and hides that extra chunk unless you requested it yourself.

Intent, outcome and trace ID

with ai_interaction(
    user_id=u, conversation_id=c, intent="refund_request", trace_id=langfuse_trace_id
) as interaction:
    ...
    interaction.set_intent("order_status")  # known only later: goes on the response event
    interaction.set_outcome("escalated")
    interaction.set_trace_id(trace_id)

intent given at the start goes on the prompt and response events; outcome and trace_id go on ai_response_received. All are free-form strings. trace_id links the interaction to Langfuse, LangSmith or an OpenTelemetry trace. With tracing registered, the first span started inside the interaction sets trace_id for you, unless you set one.

Manual instrumentation

For a client without an adapter, report calls yourself:

with ai_interaction(user_id=u, conversation_id=c) as interaction:
    result = my_llm.generate(prompt)
    interaction.record_call(
        provider="acme",
        model="acme-large",
        input_tokens=result.input_tokens,
        output_tokens=result.output_tokens,
        response_id=result.id,
        finish_reason=result.stop_reason,
    )
    interaction.end(response_text=result.text)  # optional: end early, with the answer (content capture only)

For frameworks with their own lifecycle hooks, start_interaction() creates an interaction without entering it; wrap work in interaction.activate() and call interaction.end().

Background agents

Use ai_task for LLM work no user is waiting on: a scheduled job, a webhook or queue worker, a Slack-triggered bot, or a sub-agent another agent hands work to. It is a system-triggered interaction: no ai_user_prompt_created, one ai_response_received per run.

from rudderstack.ai import ai_context, ai_task

with ai_context(account_id=workspace.id):  # set once for the whole worker
    with ai_task("alert-triage", on_behalf_of=alert.owner_id, trigger_source="schedule") as run:
        summary = summarise(alert)  # wrapped clients report into the run
        run.set_outcome("resolved")

It works as with, async with or a decorator (@ai_task("nightly-digest"), sync or async; each call is its own run).

• Actor vs beneficiary. The actor of a background run is the system: userId stays unset (anonymousId is ai_system) and an ai_context user_id, anonymous_id or conversation_id is not used, so machine activity never counts as a person's activity. Put the customer in account_id and the human the work is for in on_behalf_of. Pass user_id= only for a service identity of your own.
• Chaining. Inside another interaction or task, a task is chained to it: parent_interaction_id is the current interaction's ID, conversation_id its conversation and trigger_source defaults to agent. It also inherits account_id and on_behalf_of. Pass parent_interaction_id="..." to link a run started elsewhere (e.g. from a queue message), or parent_interaction_id=None to start a new chain.
• Flush. On exit, success or error, the task ends and then calls the RudderStack client's flush() (awaited under async with), so a cron job, worker or Lambda delivers its events before it stops. Pass flush=False for nested tasks or long-lived workers. A failing flush is logged, never raised; errors from your code propagate unchanged and mark the run error, as with ai_interaction.

ai_interaction and start_interaction accept the same fields (account_id, on_behalf_of, trigger_source, task_name, parent_interaction_id) without the defaults, and track_action takes account_id and on_behalf_of for late feedback on a background run. See examples/background-agent.

Configuration

import rudderstack.ai

rudderstack.ai.configure(
    analytics=rudder_analytics,  # global fallback client
    prices={
        "gpt-4.1-mini": {"input_per_mtok": 0.25, "output_per_mtok": 2.00, "cached_input_per_mtok": 0.025}
    },
    capture_content=False,  # prompt_text, response_text, feedback_text
    emit_llm_calls=False,  # also send one ai_llm_response_received per model call
)

configure() changes only the arguments you pass; prices are merged into the registered prices (call clear_prices() to start over). The prices above are placeholders: use your own rates.

Which client is used. ai_interaction(analytics=...) > ai_context(analytics=...) > the adapter's analytics= (wrap_openai(client, analytics=...), RudderStackCallbackHandler(analytics=...), tracked_query(analytics=...)) > configure(analytics=...). With no client anywhere, events are dropped with a debug log.

Logging. The package logs to the rudderstack.ai logger: warnings for dropped ai_user_action events, debug messages for swallowed tracking errors.

Serverless. The RudderStack client sends events from a background queue. In AWS Lambda and similar runtimes, call rudder_analytics.flush() before the handler returns (ai_task does this for you).

API reference

Every public function and class has a docstring with the details; the package ships type hints (py.typed).

Events

All four events are track calls. Every event has conversation_id and interaction_id, plus feature when set; the spec version travels in context.wrapperLibrary.specVersion. IDs the SDK mints (interaction_id, and conversation_id when you don't give one) are UUID v4 strings; IDs you pass are used as given. The JSON Schema is in spec/ai-events-v0.2.schema.json; use it in a tracking plan.

Run fields (optional, sent only when set, never null). The prompt, per-call and response events of an interaction carry all five; ai_user_action carries account_id and on_behalf_of.

ai_user_prompt_created

{
  "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f",
  "interaction_id": "0b6e2f4a-3c1d-4e7f-9a8b-5c6d7e8f9a0b",
  "feature": "support_chat",
  "input_method": "text",
  "intent": "refund_status"
}

Optional: input_method, intent, prompt_text (content capture only).

ai_response_received

One per answer: an OpenAI tool loop with two model calls, priced with a registered price:

{
  "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f",
  "interaction_id": "0b6e2f4a-3c1d-4e7f-9a8b-5c6d7e8f9a0b",
  "feature": "support_chat",
  "response_status": "success",
  "trigger": "user",
  "primary_model": "gpt-4.1-mini",
  "primary_provider": "openai",
  "latency_ms": 1840,
  "llm_call_count": 2,
  "tool_call_count": 1,
  "total_input_tokens": 60,
  "total_output_tokens": 24,
  "total_tokens": 84,
  "total_cached_input_tokens": 0,
  "total_cost_usd": 0.000063,
  "finish_reason": "stop",
  "intent": "refund_status",
  "outcome": "resolved",
  "usage_breakdown": [
    {
      "provider": "openai",
      "model": "gpt-4.1-mini",
      "call_count": 2,
      "input_tokens": 60,
      "output_tokens": 24,
      "cached_input_tokens": 0,
      "total_tokens": 84,
      "cost_usd": 0.000063
    }
  ],
  "trace_id": "lf-trace-91b2"
}

ai_user_action

{
  "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f",
  "interaction_id": "0b6e2f4a-3c1d-4e7f-9a8b-5c6d7e8f9a0b",
  "action_type": "feedback_given",
  "action_details": {"feedback_type": "thumbs", "feedback_value": true}
}

action_details.feedback_type and feedback_text must be strings; a value of another type is dropped with a warning.

ai_llm_response_received (optional)

Off by default (emit_llm_calls=True). One per model call, for debugging:

{
  "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f",
  "interaction_id": "0b6e2f4a-3c1d-4e7f-9a8b-5c6d7e8f9a0b",
  "feature": "support_chat",
  "provider": "openai",
  "model": "gpt-4.1-mini",
  "input_tokens": 25,
  "output_tokens": 12,
  "total_tokens": 37,
  "cached_input_tokens": 0,
  "cost_usd": 0.00003025,
  "response_id": "chatcmpl-9x2...",
  "finish_reason": "tool_calls"
}

The Claude Agent SDK reports usage per model, not per call, so there it sends one per model.

Privacy and content capture

By default no free text leaves your server: prompt_text, response_text and action_details.feedback_text are stripped. configure(capture_content=True) turns all three on; ai_interaction(capture_content=True) turns on prompt and response text for that interaction only (track_action follows the active interaction's setting when it runs inside one, else the global one). Treat captured text as personal data. (The examples turn it on so you can see the fields.)

Where the text comes from: prompt_text from ai_interaction(prompt_text=...) or the tracked_query prompt; response_text from interaction.end(response_text=...) or the Claude Agent SDK result (successful runs only); feedback_text from track_action(action_details=...).

AI analytics without personal data in events

AI events carry IDs and metrics only: conversation and interaction IDs, models, tokens, latency, cost and status. Prompt, answer and feedback text is opt-in (capture_content), and no traits or email are copied into events. Identify users with the same RudderStack client's identify call, so the users and identifies tables stay the one governed place for personal data and a deletion request only has to touch that profile. The userId on every AI event joins each prompt, answer, cost and feedback to that profile, ready for segments and activation through Audiences or Reverse ETL.

Cost

No price table ships with the package: list prices change and many teams pay negotiated rates. Cost is resolved in this order:

1. Framework-reported. The Claude Agent SDK's total_cost_usd / per-model costUSD (minus the usage_baseline on a multi-turn session).
2. Your prices. register_price() or configure(prices=...) (merged; clear_prices() empties the registry), in USD per 1M tokens. A name matches exactly or by longest prefix, so gpt-4.1-mini also prices gpt-4.1-mini-2025-04-14. A price registered with provider= wins over one for any provider.
3. None. Events carry tokens only.

register_price raises ValueError for an empty model name or a negative, infinite or NaN rate.

from rudderstack.ai import register_price

register_price(
    "claude-haiku-4-5",
    input_per_mtok=1.00,
    output_per_mtok=5.00,  # your rates
    cached_input_per_mtok=0.10,
    cache_write_per_mtok=1.25,
    provider="anthropic",
)

Cached input defaults to the input rate when cached_input_per_mtok is not given; the same goes for cache writes.

The SDK never sends a cost it doesn't know: an unpriced, failed or cancelled call keeps its tokens but has no cost_usd and adds nothing to totals. total_cost_usd is the sum of the priced calls, sent only when every successful call was priced, so a partial total never understates spend. A real priced 0 is sent as 0.

Warehouse column types

RudderStack warehouse destinations create a column with the type of the first value they load, and never change it: a whole number such as 0 (or 0.0, which arrives as the same number) makes an integer column, and later decimal costs in it are truncated to whole numbers. That is why the SDK never sends a placeholder 0 for an unknown cost, but a real cost of 0 (a free model) arriving first still makes an integer column. A tracking plan does not change this: it validates events but does not set column types.

1. Validate with the tracking plan. spec/rudderstack-catalog/ holds the AI events as a RudderStack data catalog and tracking plan, with costs typed as decimal numbers: rudder-cli validate -l spec/rudderstack-catalog, then rudder-cli apply -l spec/rudderstack-catalog, and connect the tracking plan to your source.
2. Fix a cost column that is already an integer (total_cost_usd, and cost_usd on ai_llm_response_received), then let RudderStack pick up the schema change (it caches the warehouse schema for up to 12 hours). Costs already loaded into it lost their decimals; replay them from the source if you need them.

Adoption tag: context.wrapperLibrary

Every event carries context.wrapperLibrary beside the core SDK's context.library, so RudderStack can see which integration produced it:

{
  "library": {"name": "analytics-python", "version": "2.1.9"},
  "wrapperLibrary": {"name": "rudderstack-ai", "version": "0.1.0", "integration": "openai", "specVersion": "0.2.0"}
}

integration is the first adapter that reported into the interaction: openai, anthropic, langchain, claude-agent-sdk, or manual (track_action, or an interaction with only hand-recorded calls). If a library built on top of this one sets its own tag with context={"wrapperLibrary": {...}} (on ai_context, ai_interaction or track_action), it is nested inside ours as wrapperLibrary.wrapperLibrary; other keys of that context are passed through. A context given to ai_interaction or track_action replaces the ai_context one for those events. Tracing events (ai_otel_span) carry {"name": "rudderstack-ai", "version": ..., "integration": "tracing"}, without specVersion: their schema version is properties.schema_version.

Tracing

rudderstack.ai.tracing (the tracing extra) is an OpenTelemetry span processor and exporter. Keep your normal OpenTelemetry instrumentation (the official GenAI instrumentations, OpenInference, OpenLLMetry); each finished AI span becomes one ai_otel_span track event, sent through your RudderStack client. Events land in your warehouse, and because each one embeds a complete OTLP payload, a RudderStack destination can forward it verbatim to Langfuse, Braintrust or any OTLP /v1/traces endpoint.

Importing rudderstack.ai.tracing without the extra raises an ImportError that names it.

Set up

Your app already uses OpenTelemetry (Langfuse, Braintrust, Arize, Traceloop, Honeycomb, the OpenTelemetry distro): add the RudderStack span processor to your TracerProvider, next to your vendor's. Both see every span; each exports what it wants. Use the processor, not just the exporter: the processor tags spans as they start (identity, rudderstack.interaction_id) and gives the interaction its trace_id, which is the join to the product events.

from rudderstack.ai.tracing import create_span_processor

tracer_provider.add_span_processor(create_span_processor())  # client and capture_content from configure()

With Langfuse, which adds its processor to the provider you pass:

from langfuse import Langfuse
from openinference.instrumentation.anthropic import AnthropicInstrumentor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer_provider = TracerProvider()
langfuse = Langfuse(tracer_provider=tracer_provider)  # keys from LANGFUSE_* environment variables
tracer_provider.add_span_processor(create_span_processor())
trace.set_tracer_provider(tracer_provider)
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)

Any instrumentation works the same way. OpenInference (Arize, Phoenix): openinference.instrumentation.<library>.<Library>Instrumentor().instrument(tracer_provider=tracer_provider). OpenLLMetry: the bare packages likewise (opentelemetry.instrumentation.anthropic.AnthropicInstrumentor), or with the Traceloop SDK, pass ours with your other processors:

from traceloop.sdk import Traceloop

Traceloop.init(processor=[create_span_processor(), *your_processors])

If your vendor or the distro already set the global SDK TracerProvider, register_tracing() adds the processor to it and returns a handle whose shutdown() leaves that provider running. It raises RuntimeError when the global provider is not an SDK TracerProvider (add create_span_processor() to the provider your instrumentation uses instead) or already has a RudderStack processor. This differs from the Node.js SDK, whose registerTracing() throws on an existing provider because OpenTelemetry JS providers take processors only at construction.

No OpenTelemetry yet: register_tracing() creates a TracerProvider, makes it global and adds the processor:

from opentelemetry.instrumentation.openai_v2 import OpenAIInstrumentor
from opentelemetry.sdk.resources import Resource
from rudderstack.analytics import Client

import rudderstack.ai
from rudderstack.ai.tracing import register_tracing

rudderstack.ai.configure(analytics=Client("<WRITE_KEY>", host="<DATA_PLANE_URL>"))
tracing = register_tracing(resource=Resource.create({"service.name": "chat-api"}))
OpenAIInstrumentor().instrument(tracer_provider=tracing.tracer_provider)

# ... use the OpenAI client as usual ...

tracing.shutdown()  # exports pending spans and flushes the RudderStack client

Exporter only (BatchSpanProcessor(RudderStackSpanExporter()), for example in a vendor's exporter list): raw spans without the join. Spans get no interaction_id, no identity from ai_context and no parent-to-child identity, and ai_response_received gets no trace_id. Identity comes only from the span's own attributes, else the trace id.

Serverless (AWS Lambda, Cloud Run jobs): use processor="simple" so spans are exported as they end, and call tracing.force_flush() before the handler returns.

Anything that emits OpenTelemetry spans works. Tested in examples/:

Joined to the product events

A span that starts inside ai_context or ai_interaction gets, unless it already has them:

The interaction also adopts the span's trace id as its trace_id when none was set, so ai_response_received.trace_id equals the spans' trace_id and every span carries interaction_id. A trace_id you set yourself is never overwritten. No extra API: it works with the ai_context / ai_interaction you already have.

Product events from spans

With product_events=True (default off), the spans build the interaction's product events, so you can instrument the model calls with OpenTelemetry only, without a provider wrapper:

tracing = register_tracing(product_events=True)
AnthropicInstrumentor().instrument(tracer_provider=tracing.tracer_provider)

with ai_interaction(prompt_text=question) as turn:  # explicit mode: the interaction is still required
    reply = anthropic_client.messages.create(...)  # a plain client: its span is the call

Each LLM-call span that ends while its interaction (named by rudderstack.interaction_id) is still open is recorded as one call on it, exactly as a wrapper would report it: ai_response_received gets its model, provider, tokens (cache reads and writes included), cost from your registered prices, finish reason, status and error_type, and time_to_first_token_ms when the span has gen_ai.response.time_to_first_chunk; emit_llm_calls adds one ai_llm_response_received per span. With content capture on and no response_text given to end(), response_text is the output text of the last successful LLM-call span. Prompt text never comes from spans, and ai_user_prompt_created / ai_user_action are unchanged. Calls from spans carry context.wrapperLibrary.integration: "tracing" (the first reporter of an interaction tags it, as with mixed wrappers).

The first convention a span carries decides. A failed span (status ERROR) is a failed call, with no cost; its error_type comes from http.response.status_code, error.type or the exception event (429 is rate_limit, 408 / 504 or a timeout error timeout, an abort or cancellation cancelled). Spans outside an interaction, spans that end after it ended and spans naming an interaction not open in this process produce no product events (implicit mode is not supported).

Double reporting. When a wrapper (or the LangChain handler) and an instrumentation both report the same call, it counts once: by the provider's response id when both have it, otherwise when the two reports come from the same provider and their time windows overlap (each report matches at most one other). Two calls that both have different response ids, or that don't overlap, are two calls. OpenInference spans carry no response id, so they rely on the time window.

Configuration

RudderStackSpanExporter(analytics=None, **options), create_span_processor(analytics=None, processor="batch", propagate_identity=True, product_events=False, **options) and register_tracing(analytics=None, tracer_provider=None, resource=None, set_global=True, processor="batch", propagate_identity=True, product_events=False, **options) take the same options:

processor: "batch" (background BatchSpanProcessor) or "simple" (SimpleSpanProcessor). propagate_identity (processor and register_tracing only, default True): a starting span inherits user.id / session.id from its parent when it lacks them. product_events (processor and register_tracing only, default False): see Product events from spans. register_tracing returns a TracingHandle with tracer_provider, span_processor, exporter, force_flush() and shutdown(); shutdown() shuts the provider down only if register_tracing created it. The processor's force_flush() also flushes the RudderStack client. Neither shutdown() closes your RudderStack client; call analytics.shutdown() yourself at exit.

Which spans are exported. By default (is_ai_span), spans with any gen_ai.* attribute (OpenTelemetry GenAI semantic conventions), an openinference.span.kind attribute (OpenInference), or an ai.operationId attribute (Vercel AI SDK). Parents such as HTTP server spans are not exported, but parent_span_id still points at them.

Identity. Track events need userId or anonymousId. The default resolver reads, each from the span, then from its resource, first match wins:

The trace-id fallback puts every span of a trace on one anonymous profile. Inside ai_context the span processor sets user.id and rudderstack.anonymous_id for you unless the span has them (see above); it also copies user.id and session.id from parent to child spans (propagate_identity), so setting them on the request's root span is enough. The bare exporter does neither. Outside ai_context, the vendors' own APIs work: OpenInference's using_attributes(user_id=..., session_id=...) and Langfuse's propagate_attributes(user_id=..., session_id=...) set user.id / session.id; Traceloop's set_association_properties({"user_id": ..., "session_id": ...}) sets the traceloop.association.properties.* keys. Or use baggage plus BaggageSpanProcessor (see tracing-fastapi-chat).

Errors. Nothing raises into your app. Failures are logged on the rudderstack.ai.tracing logger; a failing track call yields a failed export result.

ai_otel_span

One track event per span; schema: spec/ai-otel-span-v0.1.schema.json. Both SDKs emit identical events, pinned by spec/fixtures/tracing-conformance-v0.1.json. messageId is <trace_id>-<span_id>, so retries de-duplicate; timestamp is the span end time.

gen_ai summarises OpenTelemetry GenAI semantic-convention attributes; every field is optional:

OpenInference attributes (llm.model_name, llm.token_count.*) are not summarised; they are in otlp.

context.wrapperLibrary is {"name": "rudderstack-ai", "version": "<version>", "integration": "tracing"}; context.library stays the RudderStack client's own. The RudderStack client adds integrations, channel and sentAt, and writes "userId": null when there is no user id (the schema allows null as long as one id is a string).

Example (a GenAI chat span inside an interaction; otlp shortened):

{
  "type": "track",
  "event": "ai_otel_span",
  "userId": "user_123",
  "anonymousId": "3f2b8c1e-7a4d-4e9b-b6c5-0d1e2f3a4b5c",
  "messageId": "4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7",
  "timestamp": "2025-10-09T08:53:22.500Z",
  "properties": {
    "schema_version": "0.1",
    "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
    "span_id": "00f067aa0ba902b7",
    "parent_span_id": null,
    "span_name": "chat gpt-5",
    "span_kind": "client",
    "start_time": "2025-10-09T08:53:20.000Z",
    "end_time": "2025-10-09T08:53:22.500Z",
    "duration_ms": 2500.25,
    "status_code": "ok",
    "interaction_id": "0b6e2f4a-3c1d-4e7f-9a8b-5c6d7e8f9a0b",
    "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f",
    "gen_ai": {
      "provider_name": "openai",
      "operation_name": "chat",
      "request_model": "gpt-5",
      "input_tokens": 120,
      "output_tokens": 48,
      "finish_reasons": ["stop"],
      "conversation_id": "7d1f3c2a-9b4e-4c8d-a6f1-2e5b8c9d0a3f"
    },
    "truncated": false,
    "otlp": {"resourceSpans": [{"resource": {"attributes": []}, "scopeSpans": [{"scope": {}, "spans": [{"...": "..."}]}]}]}
  },
  "context": {
    "wrapperLibrary": {"name": "rudderstack-ai", "version": "0.1.1", "integration": "tracing"},
    "library": {"name": "analytics-python", "version": "2.1.9"}
  }
}

Tracing privacy

Content capture follows configure(capture_content=...) (off by default) unless you pass capture_content= to the tracing setup. With it off, the exporter removes these attributes from the span and its events, and drops these events entirely, before export. The lists are identical in the Node.js SDK and exported as constants.

The identity attributes user.id, rudderstack.anonymous_id, rudderstack.interaction_id and gen_ai.conversation.id are never removed. Everything else your instrumentation records (model names, token counts, URLs, custom attributes) is exported. Most instrumentations also only record content when you opt in (for example OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT for the official ones; OpenLLMetry records it unless TRACELOOP_TRACE_CONTENT=false). Use filter or a span processor to drop anything else sensitive.

Size limits

RudderStack rejects events over 32 KiB. The exporter measures each event as compact JSON in UTF-8 bytes (like the Node.js SDK) and, while it is over max_event_bytes (default 32,000, leaving room for the fields the RudderStack client adds), shortens the longest attribute string value in otlp (resource, span, events, links) to fit, appending ...[truncated], never below 256 bytes. The event then has truncated: true. A span that still does not fit is dropped with a warning. The Python RudderStack client serialises with spaces and \u escapes, so if your spans carry many attributes or much non-ASCII text, lower max_event_bytes.

Forwarding the OTLP payload

properties.otlp is a standard OTLP/JSON ExportTraceServiceRequest (hex ids, nanosecond timestamps as strings, integer kind and status.code), built with the official OTLP protobuf encoder. Posting it unchanged with Content-Type: application/json to any OTLP/HTTP /v1/traces endpoint (OpenTelemetry Collector, Langfuse, Braintrust, Honeycomb, ...) recreates the span; CI checks every payload against a real OpenTelemetry Collector. In the warehouse, query the summary columns, or unnest otlp for any attribute.

Compatibility

CI runs the test suite with the lowest and the highest versions each range allows.

Limitations

• Not tracked: OpenAI responses.create(background=True) (only queued at that point; no outcome or usage yet), OpenAI and Anthropic with_raw_response / with_streaming_response, and Anthropic's beta.messages.tool_runner() (it calls the API through the unwrapped client). Use ai_interaction + record_call for those.
• The .stream() helpers are observed by swapping the helper's private raw stream. If a future SDK renames it, the call still works but goes untracked; CI's latest-versions job catches this.
• An async stream abandoned with break outside any interaction is reported when it is closed or garbage-collected. Inside an interaction it is closed out as a cancelled call when the interaction ends. Close async streams (await stream.close() or async with) to release the connection.
• A stream passed to interaction.bind() must be consumed, or the response event is never sent.
• The LangChain handler's own interactions (opened from run metadata) are not visible to wrap_openai / wrap_anthropic clients called inside the run, since callbacks can't set the caller's context. When you combine a wrapped client with the handler, wrap the run in ai_interaction so both report into one interaction.
• A wrapped client is a proxy: attribute access, with_options() and type hints behave like the original, but isinstance(wrapped, OpenAI) is False. Pass unwrap(wrapped) to code that checks the class.
• MCP traffic is out of scope.

Examples

examples/ has a runnable app per integration, three tracing apps (tracing-*) and a full-stack chat app (FastAPI, Anthropic and the RudderStack JavaScript SDK; RUDDERSTACK_TRACING=1 adds joined spans). Each runs against the real APIs with your keys, or offline against the bundled mock server.

Contributing and security

See CONTRIBUTING.md and SECURITY.md. Releases: RELEASING.md.

License

MIT