rudderstack-ai

Product analytics for the AI features of a Python backend. rudderstack-ai observes your OpenAI, OpenAI Agents SDK, Anthropic, Google Gen AI, LangChain / LangGraph or Claude Agent SDK calls and sends the standard AI events (spec v0.2) through the RudderStack client your app already uses. The Node.js package is @rudderstack/ai; both send the same events, pinned by the shared fixtures in spec/fixtures.

How it fits

your request handler ── ai_interaction ── wrapped client / LangChain handler / tracked_query
                              │
                              └─ your rudder-sdk-python client ─▶ RudderStack ─▶ warehouse, destinations, audiences

• One event per answer. However many model calls, agents and tool steps an answer takes, you get one ai_response_received with its totals: tokens, cost, latency, models, tool calls, status. The prompt, the answer and later feedback share an interaction_id; every event carries conversation_id.
• Product analytics, not debugging. The events land next to your other product events, keyed to the same userId, so you can build funnels, retention and audiences ("users whose refund questions went unresolved"). It complements tracing tools (Langfuse, LangSmith, Braintrust, APM): set trace_id to join the two, or use the optional tracing module to send OpenTelemetry spans through the same pipe.
• Private by default. Events carry IDs and metrics. Prompt, answer and feedback text are sent only when you turn on content capture, and are then masked.
• Never in the way. Clients are wrapped, not patched; the SDK never throws into your LLM call or flush() (its own errors are logged), and provider errors reach your code unchanged. Delivery errors from the RudderStack client are handled by that client: pass its on_error callback to observe them (an exception raised from the client's own consumer thread is not the SDK's to catch). Server-side only.

Install

Python 3.12 or later. Install the extra for each library you use: openai, anthropic, google-genai (Gemini API or Vertex AI), bedrock (boto3 Bedrock Runtime), langchain (LangChain, LangGraph, Deep Agents), openai-agents (OpenAI Agents SDK), claude-agent-sdk, litellm (LiteLLM SDK), and tracing for OpenTelemetry spans (rudderstack.ai.tracing).

pip install "rudderstack-ai[openai]"  # or: uv add "rudderstack-ai[openai]"

The package has no dependencies of its own. It sends events through your rudder-sdk-python client (rudderstack.analytics) and never creates one.

Quick start

import rudderstack.analytics as rudder_analytics
from openai import OpenAI

import rudderstack.ai
from rudderstack.ai import ai_interaction, track_action
from rudderstack.ai.openai import wrap_openai

rudder_analytics.write_key = "<WRITE_KEY>"
rudder_analytics.dataPlaneUrl = "<DATA_PLANE_URL>"
rudderstack.ai.configure(analytics=rudder_analytics)  # once, at startup

client = wrap_openai(OpenAI())


def answer(user_id: str, chat_id: str, question: str) -> dict[str, str | None]:
    with ai_interaction(user_id=user_id, conversation_id=chat_id, feature="support_bot") as interaction:
        reply = client.chat.completions.create(
            model="gpt-4.1-mini", messages=[{"role": "user", "content": question}]
        )
        interaction.set_outcome("resolved")
    return {"answer": reply.choices[0].message.content, "interaction_id": interaction.id}


def feedback(user_id: str, chat_id: str, interaction_id: str, thumbs_up: bool) -> None:
    track_action(
        "feedback_given",
        interaction_id=interaction_id,
        conversation_id=chat_id,
        user_id=user_id,
        action_details={"feedback_type": "thumbs", "feedback_value": thumbs_up},
    )

This sends ai_user_prompt_created and ai_response_received for the answer and ai_user_action for the feedback. In a script, worker or serverless handler, call rudderstack.ai.flush() before exiting (Serverless and workers). examples/ has a runnable app per integration, a background agent, three tracing apps and a full-stack chat app (FastAPI, Anthropic and the RudderStack JavaScript SDK). Each runs against the real APIs with your keys, or offline against the bundled mock server.

Choose your integration

Use one integration per call path. Everything inside one ai_interaction becomes one answer, and a call reported by two integrations (a wrapped client used by LangChain, a wrapper plus a tracing span) is counted once.

Rules that apply to every chat app:

• Cost precedence: use the provider or gateway's reported cost first, then a registered model price; omit cost when neither is known.
• One ai_interaction per user turn, not per conversation. Pass your chat thread's ID as conversation_id. Resending the history on every turn is fine: usage is counted per model call, not from the history.
• Streaming responses: a route that returns a stream must keep the interaction open until the stream ends. Pass the generator through interaction.bind() (example).
• Thread pools: ThreadPoolExecutor and loop.run_in_executor do not copy context; wrap the function with rudderstack.ai.propagate(fn) (asyncio.to_thread copies it).
• A call outside any interaction still sends its own ai_response_received (trigger: user when identity is in context, else system), but no prompt event and no classification.

What each integration tracks and its edge cases: docs/integrations.md.

OpenAI

from openai import AsyncOpenAI
from rudderstack.ai.openai import wrap_openai

# AzureOpenAI or a compatible endpoint: provider="azure.ai.openai", "openrouter"...
client = wrap_openai(AsyncOpenAI())

• Streamed chat calls report usage only when the response includes it. The wrapper asks for it (stream_options.include_usage) on every host, including compatible endpoints (Azure, Groq, OpenRouter, Ollama, vLLM), and hides the extra chunk unless you asked for it. A request the host rejects (HTTP 400 or 422 naming the parameter, or with no error code) is sent once more as you wrote it, so a working call never breaks; once a host names the parameter, that client and its with_options() copies stop adding it there. Other rejections (a content filter, a prompt too long) are not resent. stream_usage=True always adds it (no resend), False never does.
• finish_reason uses Chat Completions' terms on both APIs: a Responses output with only tool calls is tool_calls, incomplete at max_output_tokens is length, a refusal (message.refusal, a Responses refusal part) is refusal; a streamed Chat Completions refusal keeps the wire's stop. Refusals and content filters are visible through finish_reason, not through status: a completed call is response_status: success even when the model refused, was filtered, or was cut off at length (structured output too), and it keeps its tokens and cost. Only an exception your code gets fails the answer: chat.completions.parse() raising LengthFinishReasonError / ContentFilterFinishReasonError gives response_status: error with error_type length / content_filter; responses.parse() raising on truncated structured output is a failed call (see Compatibility).
• The wrapper is a proxy: isinstance(client, OpenAI) is False. Pass rudderstack.ai.unwrap(client) to code that checks the class.

OpenAI Agents SDK

from agents import SQLiteSession
from rudderstack.ai.openai_agents import tracked_run, tracked_run_streamed

result = await tracked_run(
    triage_agent,
    question,
    tracking={"user_id": user_id, "feature": "support_agent"},
    session=SQLiteSession(chat_id),
)

stream = tracked_run_streamed(triage_agent, question, tracking={"user_id": user_id})
async for event in stream.stream_events():
    render(event)

tracked_run(agent, input, *, tracking=None, **run_kwargs) runs Runner.run (tracked_run_sync: Runner.run_sync) and reports the whole run, across handoffs, as one interaction; inside ai_interaction (or with tracking["interaction"]) it reports into that one instead. run_kwargs go to the Runner unchanged, and your hooks still run. tracking takes the interaction options plus analytics and provider (the label for every call, default openai, for OpenAI-compatible endpoints).

• Model calls: each model response is one call, attributed to the agent that made it (agent_name in usage_breakdown), with the agent's configured model name, input, output, reasoning (when the provider reports them) and cached tokens, and finish_reason: stop for a completed response. A response that wrap_openai already recorded (the same provider response ID) is counted once and gains the agent name.
• Tools: every tool the model requested is a tool call: function tools, hosted tools (web_search) and handoffs (transfer_to_<agent>). A local function tool also reports its duration and outcome; the Agents SDK's default failure output ("An error occurred while running the tool.") counts as a failure. Hosted tools and handoffs have no known outcome.
• Agents used as tools: the nested run's model calls are recorded from Agents SDK trace spans and attributed to the nested agent, with the model name the provider reported. This needs Agents SDK tracing (on by default; RunConfig(tracing_disabled=True) or OPENAI_AGENTS_DISABLE_TRACING turn it off, and then only the top-level calls are counted). The nested run's own tools are not tracked. The helpers register a process-wide trace processor once and leave exporters alone; if you replace processors with set_trace_processors, include register_openai_agents_tracing()'s processor.
• Failures: a guardrail tripwire ends the run with response_status: error and error_type: guardrail_trip. A provider error fails the call in flight (attributed to the running agent) and the run; calls before it keep their usage.
• Conversation: tracking conversation_id, then the Runner's conversation_id, then the session's ID, then a generated ID.
• Streaming: the interaction ends when stream_events() finishes or fails, or when you stop consuming it or call cancel() (cancelled).
• Classification: interactions these helpers open are not classified unless you pass classify=True in tracking.

Gateways and routers (OpenRouter, a LiteLLM proxy) go through the same wrapper: wrap_openai(OpenAI(base_url=...), provider="openrouter"). Their own cost wins over a registered price (usage.cost, LiteLLM's x-litellm-response-cost); a vendor-prefixed model (openai/gpt-4o) is priced by a price registered for gpt-4o and reported under its vendor (openai), with the full name in model; requested_model names the model you asked for when a router or fallback served another one. Details: docs/integrations.md. See the OpenRouter example.

LiteLLM

import litellm
from rudderstack.ai.litellm import instrument_litellm

instrument_litellm()  # once at startup
reply = litellm.completion(
    model="anthropic/claude-haiku-4-5", messages=messages, fallbacks=["gemini/gemini-3.5-flash-lite"]
)

Tracks litellm.completion / acompletion (sync, async, streamed), and so the frameworks built on LiteLLM (CrewAI, DSPy, smolagents). A call with fallbacks= is one call, from the model that answered. Cost is LiteLLM's response_cost, then a registered price. LiteLLM 1.83.1 and later require openai<3, so the litellm extra (>=1.88.6) cannot be installed alongside openai 3; that constraint is LiteLLM's. See docs/integrations.md and the fallback example. The Node.js SDK tracks LiteLLM through its proxy (wrapOpenAI), not an SDK integration; the Vercel AI Gateway is JavaScript-only (Node.js SDK).

Anthropic

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

# AnthropicBedrock: provider="aws.bedrock"; AnthropicVertex: "gcp.vertex_ai"
client = wrap_anthropic(Anthropic())

Google Gen AI (Gemini and Vertex AI)

from google import genai
from rudderstack.ai.google_genai import wrap_google_genai

client = wrap_google_genai(genai.Client(api_key=os.environ["GEMINI_API_KEY"]))
vertex = wrap_google_genai(genai.Client(vertexai=True, project="my-project", location="us-central1"))

Amazon Bedrock (Converse)

import boto3
from rudderstack.ai.bedrock import wrap_bedrock

client = wrap_bedrock(boto3.client("bedrock-runtime", region_name="us-east-1"))

LangChain and LangGraph

One callback handler covers LangChain, LangGraph and Deep Agents. Pass it in the run's config, not to a model's constructor (a handler that sees only model runs makes each model call its own answer, and logs a warning).

from rudderstack.ai.langchain import RudderStackCallbackHandler

handler = RudderStackCallbackHandler()

result = graph.invoke(
    {"messages": [("user", question)]},
    config={
        "callbacks": [handler],
        "configurable": {"thread_id": chat_id},
        "metadata": {"user_id": user_id, "feature": "support_agent"},
    },
)

• Outside ai_interaction, the handler opens one interaction per root run (invoke / stream) with identity from metadata (user_id, anonymous_id, conversation_id, interaction_id, feature, intent, trace_id). Without conversation_id, LangGraph's thread_id is the conversation.
• With a checkpointer, each invoke is one turn; resuming an interrupted graph (Command(resume=...)) opens a new interaction without a prompt event.
• A graph that stops at interrupt() ends its answer with response_status: paused (inside ai_interaction too, unless a later call answers it), so nothing is left open while it waits, even across processes. To report the resume as the same turn, pass the same interaction_id in the metadata of both runs, for example one derived from the thread_id; Send branches and subgraphs stay in that interaction. See examples/langgraph-checkpoint.
• A wrapped OpenAI or Anthropic client called directly from a graph node cannot see the handler's interaction. Wrap the run in ai_interaction to join them (that also turns on classification, which handler-opened interactions skip).
• usage_breakdown[].agent_name comes from the agent_name / lc_agent_name metadata (LangChain agents and Deep Agents subagents set it) or an agent:<name> tag. Providers are reported with their OpenTelemetry names (aws.bedrock, gcp.vertex_ai, azure.ai.openai...). See Deep Agents across Claude and Gemini.
• A model run's usage is summed over its prompts. A stream the consumer stops or a cancelled run is a cancelled call that keeps the tokens streamed so far (summed over chunks, as LangChain merges them), unpriced.

Claude Agent SDK

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

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage
from rudderstack.ai.claude_agent_sdk import tracked_query, usage_snapshot

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

Resumed and continued sessions. On resume / continue_conversation, on later turns of a ClaudeSDKClient and on later results of a streaming-input query(), the SDK reports usage and cost for the whole session so far. The SDK subtracts the previous result it saw in this process; across processes, store usage_snapshot() and pass it as the next turn's usage_baseline. Without either, a resumed turn counts its own tokens and prices them with your registered prices (no cost when none is registered). Turns it opens are classified only with classify=True.

Tracing only

If your model calls are already instrumented with OpenTelemetry (OpenLLMetry, OpenInference, the official GenAI instrumentations), register_tracing(product_events=True) builds the same product events from the spans, without a provider wrapper. You still open the interaction. See Tracing.

Cloud model IDs and pricing

No prices ship with the SDK, and cloud providers name models their own way. A registered base-model price also covers:

• Bedrock model IDs (provider aws.bedrock only): a leading cross-region inference profile (us., us-gov., eu., apac., jp., au., global.) and then the vendor prefix (anthropic., amazon., meta....) are stripped for the lookup, so claude-sonnet-4-5 prices us.anthropic.claude-sonnet-4-5-20250929-v1:0 and nova-pro prices amazon.nova-pro-v1:0.
• Vertex AI revisions: claude-sonnet-4-5@20250929 matches claude-sonnet-4-5 through the usual version-suffix rule.

Azure OpenAI deployment names are yours to choose, so they are never guessed. Map them:

from rudderstack.ai import configure, register_price

register_price("gpt-4o", input_per_mtok=2.5, output_per_mtok=10)
configure(azure_deployment_models={"support-prod": "gpt-4o"})

The mapping applies to calls with provider azure.ai.openai and only to the price lookup: events keep the deployment name. Each call replaces the previous mapping (None or {} clears it); clear_prices() leaves it in place. The suffix rule still never prices another model of the family: claude-sonnet-4-5 does not price anthropic.claude-sonnet-4-5-mini-v1:0.

Other frameworks

For a client without an integration, report each call:

from rudderstack.ai import ai_interaction

with ai_interaction(user_id=user_id, conversation_id=chat_id) as interaction:
    result = my_llm.generate(prompt)
    interaction.record_call(
        provider="acme",
        model="acme-large",
        input_tokens=result.input_tokens,  # including cached tokens
        output_tokens=result.output_tokens,
        response_id=result.id,  # de-duplicates a call reported twice
    )

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

Identity

Events need a userId or an anonymousId. Pass them explicitly, set them once per request with with ai_context(user_id=..., conversation_id=..., feature=...): (every interaction, wrapped call and track_action inside uses them), or let the request middleware read them from the RudderStack JavaScript SDK's cookies. ai_context uses contextvars, so concurrent requests and asyncio tasks never mix. Nested contexts inherit what they do not set.

Request middleware

When the page runs the RudderStack JavaScript SDK (v3), its cookies arrive with every same-origin request. The middleware wraps each request in ai_context(request=...), so events carry the browser's anonymousId and session without any identity plumbing. When the page does not run the JavaScript SDK (or for API-only clients), the cookies are absent and the middleware does nothing.

All three are in rudderstack.ai.middleware, which imports no framework. Without middleware, pass the request (Starlette, Django, Flask/Werkzeug, an ASGI scope, a WSGI environ or a header mapping) or the cookies (a Cookie header or a name-to-value map, e.g. from a queue message) yourself: ai_interaction(request=request) or ai_interaction(cookies={"rl_anonymous_id": anonymous_id}).

Precedence: arguments of the interaction (or track_action), then ai_context values, then cookies. A context={"sessionId": ...} you set wins too. Only these three cookies are read; unreadable cookies are skipped. A WebSocket keeps its handshake's cookies for its whole life.

Stale rl_user_id after logout. The JavaScript SDK keeps rl_user_id until the page calls rudderanalytics.reset(), so on a shared browser it can name the previous user. Pass the server's own user_id (or an anonymous_id for a logged-out visitor) whenever you know it, call reset() on logout, or turn the fallback off with configure(identity_from_cookies=False) (per interaction: identity_from_cookies=False).

Legacy v1.1 cookies (RudderEncrypt:) are not supported; upgrade the browser SDK to v3.

Background agents

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

from rudderstack.ai import ai_context, ai_task

with ai_context(account_id=workspace.id):
    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 is the system: userId stays unset (anonymousId: ai_system), and the request's identity and cookies are never used, so machine activity never counts as a person's. Put the customer in account_id and the person the work is for in on_behalf_of. For multi-tenant bots and cron jobs, also pass a stable workspace-specific anonymous_id and explicit conversation_id; otherwise the shared ai_system ID merges unrelated machine traffic in identity-based analytics. Spans inside a task never carry the person's identity either.
• Chaining. Inside another interaction or task, a task is chained to it: parent_interaction_id and conversation_id are the current interaction's, and trigger_source defaults to agent. Pass parent_interaction_id="..." to link a run started elsewhere (e.g. from a queue message), or None to start a new chain.
• Flush. A task that is not chained to a running interaction flushes on exit (async with awaits it), so a cron job or Lambda delivers its events before it stops. Pass flush=False in a long-lived worker and flush at shutdown; flush=True makes a chained task flush too.
• Classification is off for tasks; ai_task(..., classify=True) turns it on, and the task waits for the label before flushing (inside an event loop, use async with for that).

Product signals

Feedback

Return interaction.id with the answer, then report what the user did with it through track_action() (quick start). Action types: feedback_given, feedback_retracted, copied_response, shared, reported, regenerated. feedback_retracted clears the interaction's feedback, so the latest feedback_given / feedback_retracted per interaction_id is the final state. conversation_id is required (argument or ai_context); without it the event is dropped with a warning. feedback_text is sent only with content capture on.

Intent classification

Opt-in. After each answer, your own model client labels the interaction with one of your intents and an outcome, sent as one ai_interaction_classified. Nothing waits for it.

import anthropic

rudderstack.ai.configure(
    classifier={
        "client": anthropic.Anthropic(),  # or AsyncAnthropic, OpenAI, AsyncOpenAI: your key, your provider
        "model": "claude-haiku-4-5",
        "intents": ["refund_request", "order_status", "bug_report"],
        "outcomes": ["resolved", "partial", "unresolved", "unclear"],  # optional; this is the default
        "intent_summary": False,  # optional: a one-sentence masked summary
        "timeout_ms": 5000,  # optional
    }
)
# Or your own function (sync or async; it receives a rudderstack.ai.Turn): {"classify": fn, ...}

One label per interaction you open (ai_interaction, start_interaction) that succeeded; errors and cancelled answers are skipped. Interactions an integration opens on its own are not classified: a wrapped call outside any interaction, a LangChain run outside one, a tracked_query turn outside one. To label them, wrap the request in ai_interaction, or pass classify=True to tracked_query / tracked_messages / record_claude_result. Tasks: ai_task(..., classify=True). Opt one interaction out with ai_interaction(classify=False). flush() / aflush() wait for pending classifications before flushing. Labels, input, cost, async clients, failures and limits: docs/events.md.

Declared tool events

Rules turn tool calls into your own product events (save_memory → memory_captured), next to the standard events:

rudderstack.ai.configure(
    event_rules={
        "tools": {
            "save_memory": "memory_captured",
            "create_ticket": rudderstack.ai.ToolRule(
                event="ticket_created",
                properties=lambda args, result: {"priority": args.get("priority")},
                when=lambda args, result: result is not None,
            ),
        }
    }
)

A rule is an event name or a ToolRule (or a mapping with event, properties, when). args is what the model passed (decoded JSON), result what the tool returned or your app sent back to the model, None when unseen. Each call sends its event once, when its result is known or when the interaction ends (tool_status: unknown). Wrap the whole tool loop in one ai_interaction to get results. Names starting with ai_ are reserved; a malformed rule raises at configure().

Tool breakdown

configure(tool_breakdown=True) (or per run: ai_interaction(tool_breakdown=True)) adds one entry per agent and tool identity to ai_response_received. Entries include measured-only duration totals, maxima and coverage, fixed error categories, MCP server/tool splits and nested parent_tool where integrations expose them. Tool names only, never inputs or outputs. Details: docs/events.md.

Multi-agent

usage_breakdown has one entry per agent, provider, served model and differing requested model, and tool_breakdown one per agent, MCP server, tool and parent tool, so the same model or tool used by several agents stays apart. Agents are named by:

• LangChain / LangGraph: agent_name / lc_agent_name run metadata (Deep Agents subagents set it), an agent:<name> tag, the parent run's agent, the handler's agent_name, else the root run's name (a LangGraph graph or Deep Agents agent).
• Claude Agent SDK: a subagent's calls and tools take the subagent_type of the Task / Agent call that launched it, and its tools carry that call's name as parent_tool; main-loop calls have no agent_name. Each model's tokens and cost still add up to the SDK's modelUsage: a subagent's answer, which is not streamed, counts from its tool result's usage; when the streamed messages still fall short, the main loop keeps the result's own usage and the subagents share the rest, and a model's cost is split between its agents by Claude's price ratios.
• record_call and record_tool_call: the agent_name you pass.

Every usage_breakdown entry has cache_write_tokens (summed in total_cache_write_tokens) and error_count (failed calls, retries included). Nested ai_tasks report separately; roll a run tree up by parent_interaction_id (query).

Privacy

Sent by default: IDs (userId, anonymousId, conversation_id, interaction_id), the fields you set (feature, intent, outcome, run fields), models, providers, token counts, cost, latency, status, error categories, tool names and counts. Never sent: prompt, answer and feedback text (unless content capture is on), tool arguments and results, request headers, any cookie other than the three above, traits or emails.

Content capture. configure(capture_content=True) adds prompt_text, response_text and action_details.feedback_text; ai_interaction(capture_content=True) turns it on for one interaction. prompt_text comes from ai_interaction(prompt_text=...) or the tracked_query prompt; response_text from interaction.end(response_text=...) or the Claude Agent SDK result. Treat captured text as personal data.

Masking is on by default for all captured text, the classifier's intent_summary and captured span content:

Limits: patterns catch fixed shapes only. Names, postal addresses and free-form IDs are not caught, some phone shapes are missed, and an order number shaped like a phone number is masked. Text over 32,000 characters is not scanned: the field is withheld.

The redact hook runs after the built-in patterns, for what they cannot catch:

rudderstack.ai.configure(redact=lambda text: my_ner.mask_names(text))
rudderstack.ai.configure(redact=False)  # opt out of masking entirely

A hook that raises or returns a non-string drops the field, so unmasked text is never sent. redact=None restores the default. The patterns are shared with the Node.js SDK (spec/fixtures/redaction-v0.1.json).

Tracing

rudderstack.ai.tracing (the tracing extra) is an OpenTelemetry span processor and exporter: each finished AI span becomes one ai_otel_span event with the complete OTLP span, sent through your RudderStack client. Use it to analyse spans in the warehouse or forward them to an OTLP endpoint through a RudderStack destination; skip it if product events are all you need. It also joins the two: spans started inside an interaction carry its interaction_id and identity, and the interaction's trace_id is the spans' trace ID.

from opentelemetry.instrumentation.openai_v2 import OpenAIInstrumentor
from rudderstack.ai.tracing import register_tracing

tracing = register_tracing()  # client and capture_content from configure()
OpenAIInstrumentor().instrument(tracer_provider=tracing.tracer_provider)

• Existing OpenTelemetry setup (Langfuse, Traceloop, the distro): add create_span_processor() to your TracerProvider next to your vendor's processor. register_tracing() adds it to an existing global SDK provider and leaves that provider running on shutdown().
• What is exported: AI spans only (gen_ai.*, OpenInference or Vercel AI SDK attributes); HTTP and database spans are not, and get no identity attributes. Content attributes and events are removed unless content capture is on, then masked. The exporter truncates events over max_event_bytes (default 32,000) and drops a span that still does not fit, with a warning.
• Product events from spans: register_tracing(product_events=True) records each LLM and tool span that ends inside an open interaction as a call on it, so OpenTelemetry alone produces ai_response_received.

Options, the event shape, identity resolution, privacy lists and size limits: docs/tracing.md.

Serverless and workers

The RudderStack client sends events from a background queue. Before a short-lived process or handler exits, flush:

import rudderstack.ai


def handler(event: object, context: object) -> object:  # AWS Lambda
    try:
        return run_agent(event)
    finally:
        rudderstack.ai.flush()  # inside an event loop: await rudderstack.ai.aflush()

flush() waits for pending classifications (up to the classifier timeout plus a second), then flushes every RudderStack client the SDK has used: the configured one and those passed to contexts, interactions, wrappers and tracing. It returns False when a classification was still running. A non-chained ai_task flushes for you. For a long-lived worker or web app, flush once at shutdown (FastAPI lifespan, atexit.register(rudderstack.ai.flush)). With tracing, also call tracing.force_flush() (or use processor="simple"), since spans wait in the span processor's batch.

Events

Every event carries conversation_id, interaction_id, feature when set, and context.wrapperLibrary with the integration and specVersion. Property tables, sample payloads, cost rules and warehouse column types: docs/events.md. Load spec/rudderstack-catalog/ as a tracking plan before the first events arrive, so cost columns are created as decimals.

Configuration

configure() changes only the options you pass; None clears analytics, event_rules and classifier. Prices match a model exactly or followed by a version suffix (-2025-04-14, -20251001, -0613, -v2, @20241022, -latest): gpt-4o prices gpt-4o-2024-08-06, never gpt-4o-mini. The full API is in docs/api.md; every public function has a docstring and type hints.

Troubleshooting

Turn on the package's logs with logging.basicConfig() and logging.getLogger("rudderstack.ai").setLevel(logging.DEBUG) (rudderstack.ai.tracing for spans).

Compatibility

Upgrading from 0.2.0

No API was removed. Behaviour changes:

• Masking is on. Captured prompt_text, response_text and feedback_text (and captured span content in ai_otel_span) arrive masked. configure(redact=False) restores the 0.2.0 output.
• Price matching is stricter. A registered name prices only that model and its dated or versioned variants, no longer any model it is a prefix of. Register each model family you use (gpt-4o-mini separately from gpt-4o).
• Cost rounding is half-up at 8 decimals, like the Node.js SDK (0.2.0 rounded half-to-even): about 1% of values differ by 1e-8.
• LangChain providers use OpenTelemetry names: amazon_bedrock → aws.bedrock, google_vertexai → gcp.vertex_ai, google_genai → gcp.gemini, azure → azure.ai.openai. Update prices registered with provider= and dashboards grouped by provider.
• Tracing identity (user.id, rudderstack.anonymous_id) is added only to AI spans, and never to spans inside ai_task.
• ai_task chained inside another interaction no longer flushes by default (flush=True restores it).
• A missing RudderStack client logs a warning (once) instead of a debug message; end(status=...) with an invalid status derives the status and warns instead of dropping the event.
• Deprecated (still working, removed in 1.0): Interaction.adopt_analytics, note_integration, flush_prompt, record_span_call, note_outcome, mark_first_token, track_stream, untrack_stream; start_interaction(integration=...); and from rudderstack.ai.tracing: AnalyticsClient, __version__, the CONTENT_* constants, is_content_attribute, is_content_event.

New since 0.2.0: identity from cookies and rudderstack.ai.middleware, intent classification, declared tool events, tool breakdown, masking, flush() / aflush(), propagate(), estimate_cost(), register_prices(), error helpers. See the changelog.

Known limitations

• Not tracked: OpenAI with_streaming_response, a streamed with_raw_response call and a non-streamed responses.create(background=True) (only queued; polling is not intercepted). Use record_call for those.
• A structured responses.parse() call cut off at the token limit raises while parsing, before the SDK returns the response: it is recorded as a failed call without its tokens.
• A streamed LangChain ChatOpenAI call has no response_id (langchain-openai replaces it with a run ID).
• With claude-agent-sdk versions whose ResultMessage lacks terminal_reason, an aborted run is reported as a success.
• The .stream() helpers are observed through a private attribute of the SDK; if a future SDK renames it, calls still work but go untracked (CI's latest-versions job catches it).
• Platform differences from the Node.js SDK: no Vercel AI SDK (JavaScript only); flush() is sync with an async aflush(); the Claude Agent SDK's Python parser drops mcp_tool_use blocks before the SDK sees them.

Contributing

See CONTRIBUTING.md and RELEASING.md. Report security issues as described in SECURITY.md.

License

MIT