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.

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

Contents

• Install
• Quick start
• How it works
• Configuration
• Events
• Privacy and content capture
• Cost
• Adoption tag: context.wrapperLibrary
• 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[all]"

The package has no dependencies of its own. 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)

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).

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, copied_response, shared, reported, regenerated. 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.

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().

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.

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.

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=...).

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.
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. Failed calls count as $0. A cancelled stream's usage is at best partial, so that call is left unpriced, which leaves total_cost_usd out rather than understating it.

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.

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 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.

Contributing and security

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

License

MIT