Metadata-Version: 2.4
Name: rudderstack-ai
Version: 0.3.0
Summary: Automatic RudderStack AI events (spec v0.2) for OpenAI, Anthropic, LangChain and the Claude Agent SDK, plus OpenTelemetry AI tracing.
Keywords: rudderstack,analytics,ai,llm,openai,anthropic,langchain,claude,opentelemetry,tracing
Author: RudderStack
Author-email: RudderStack <sdk@rudderstack.com>
License-Expression: MIT
License-File: LICENSE.md
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: rudderstack-ai[openai,anthropic,langchain,claude-agent-sdk,tracing] ; extra == 'all'
Requires-Dist: anthropic>=1.0.0,<2 ; extra == 'anthropic'
Requires-Dist: claude-agent-sdk>=0.2.82,<0.3 ; extra == 'claude-agent-sdk'
Requires-Dist: langchain-core>=1.2.0,<2 ; extra == 'langchain'
Requires-Dist: openai>=2.0.0,<4 ; extra == 'openai'
Requires-Dist: opentelemetry-api>=1.30.0,<2 ; extra == 'tracing'
Requires-Dist: opentelemetry-sdk>=1.30.0,<2 ; extra == 'tracing'
Requires-Dist: opentelemetry-exporter-otlp-proto-common>=1.30.0,<2 ; extra == 'tracing'
Maintainer: RudderStack
Maintainer-email: RudderStack <sdk@rudderstack.com>
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/rudderlabs/ai-sdk-python
Project-URL: Repository, https://github.com/rudderlabs/ai-sdk-python
Project-URL: Issues, https://github.com/rudderlabs/ai-sdk-python/issues
Project-URL: Changelog, https://github.com/rudderlabs/ai-sdk-python/blob/main/CHANGELOG.md
Provides-Extra: all
Provides-Extra: anthropic
Provides-Extra: claude-agent-sdk
Provides-Extra: langchain
Provides-Extra: openai
Provides-Extra: tracing
Description-Content-Type: text/markdown

# rudderstack-ai

[![CI](https://github.com/rudderlabs/ai-sdk-python/actions/workflows/ci.yml/badge.svg?branch=main)](https://github.com/rudderlabs/ai-sdk-python/actions/workflows/ci.yml)

Product analytics for the AI features of a Python backend. `rudderstack-ai` observes your OpenAI, Anthropic, LangChain / LangGraph or Claude Agent SDK calls and sends the standard [AI events (spec v0.2)](https://github.com/rudderlabs/ai-sdk-python/blob/main/spec/ai-events-v0.2.schema.json) through the RudderStack client your app already uses. The Node.js package is [`@rudderstack/ai`](https://github.com/rudderlabs/ai-sdk-node); both send the same events, pinned by the shared fixtures in [`spec/fixtures`](https://github.com/rudderlabs/ai-sdk-python/blob/main/spec/fixtures).

## Where it fits

```text
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](#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:

```bash
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)
```

The package has no dependencies of its own. It sends events through your [`rudder-sdk-python`](https://github.com/rudderlabs/rudder-sdk-python) client (`rudderstack.analytics`) and never creates one.

## Quick start

```python
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:
    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](#serverless-and-workers)).

## 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:

- **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](#streaming-from-a-route-handler)).
- **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.

### OpenAI

Tracked: `chat.completions.create` / `parse` / `stream` (also under `beta.chat`), `responses.create` / `parse` / `stream`, `beta.responses.create`, their non-streamed `with_raw_response` forms; sync and async, streaming or not.

```python
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 api.openai.com and hides the extra chunk unless you asked for it. For a compatible endpoint that supports it, pass `stream_usage=True`; otherwise streamed calls there have no tokens or cost.
- The wrapper is a proxy: `isinstance(client, OpenAI)` is `False`. Pass `rudderstack.ai.unwrap(client)` to code that checks the class.

### Anthropic

Tracked: `messages.create` (including `stream=True`), `messages.parse`, the `messages.stream()` helper, the same under `beta.messages`, and every model call of `beta.messages.tool_runner()`; sync and async.

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

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

`input_tokens` includes cache reads and writes, so totals mean the same for every provider; cache reads are also in `cached_input_tokens`.

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

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

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

```python
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).
- **One interaction per turn.** Outside `ai_interaction`, each result is its own answer, all in one conversation (a resumed session keeps the conversation its first query used). A turn no user message started (a background task notification) is `trigger: system` with no prompt event. These turns are classified only with `classify=True`.
- **`ClaudeSDKClient`**: `async for message in tracked_messages(client.receive_response()): ...`. For a single result you map yourself: `record_claude_result(result_message)`.
- `tracked_query` and `tracked_messages` take the interaction fields (`conversation_id`, `feature`, `account_id`, `classify`...) as keyword arguments.

### OpenTelemetry spans 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](#tracing).

### Your own calls

For a client without an integration, report each call:

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

### Streaming from a route handler

```python
from fastapi.responses import StreamingResponse


@app.post("/chat")
async def chat(body: ChatRequest) -> StreamingResponse:
    with ai_interaction(user_id=body.user_id, conversation_id=body.chat_id) as interaction:

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

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

`bind()` works with sync and async generators, so Django's `StreamingHttpResponse` and Flask's `Response(stream_with_context(...))` take the bound generator the same way. The interaction ends when the stream ends; a client that disconnects ends it as `cancelled`. A bound stream that is never consumed never sends its answer.

## Identity

Events need a `userId` or an `anonymousId`. Pass them explicitly, set them once per request with `ai_context`, or let the request middleware read them from the RudderStack JavaScript SDK's cookies.

```python
from rudderstack.ai import ai_context

with ai_context(user_id=user.id, conversation_id=chat.id, feature="support_bot"):
    ...  # every interaction, wrapped call and track_action in here uses these
```

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

| Framework | Setup |
| --- | --- |
| FastAPI / Starlette (any ASGI app, HTTP and WebSocket) | `app.add_middleware(AIContextMiddleware)`; call `configure()` in the app's lifespan or at import |
| Django (sync and async) | `"rudderstack.ai.middleware.DjangoAIContextMiddleware"` in `MIDDLEWARE`; call `configure()` in an `AppConfig.ready()` |
| Flask | `init_flask(app)` in the app factory, after `configure()` |

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:

```python
with ai_interaction(request=request, user_id=current_user_id):
    ...
with ai_interaction(cookies={"rl_anonymous_id": anonymous_id, "rl_session": session_cookie}):
    ...
```

| Cookie | Becomes | Rule |
| --- | --- | --- |
| `rl_anonymous_id` | `anonymousId` | unless set explicitly |
| `rl_session` | `context.sessionId` | while the session is active; never `sessionStart` |
| `rl_user_id` | `userId` | fallback only: ignored when a `user_id` or `anonymous_id` is set explicitly or in `ai_context` |

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

```python
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`. 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 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:

```python
track_action(
    "feedback_given",
    interaction_id=interaction_id,
    conversation_id=chat_id,
    user_id=user_id,
    action_details={"feedback_type": "thumbs", "feedback_value": False, "feedback_text": comment},
)
```

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.

```python
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.
rudderstack.ai.configure(
    classifier={
        "classify": lambda turn: {"intent": "order_status", "outcome": "resolved"},
        "intents": ["order_status"],
    }
)
```

- **Which interactions.** 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)`.
- **Labels.** `intent` is one of `intents`, else `other`; `outcome` one of `outcomes`, else `unclear`. They share names with the app-set `intent` / `outcome`, so queries can coalesce them.
- **Input.** The prompt, the final answer, the tool names (never arguments or results) and the most recent earlier messages, within a fixed 12,000-character budget. The text goes only to your own provider.
- **Isolation and cost.** One extra small model call per answer, on your key, never tracked as an AI event or counted in the interaction. `classifier_cost_usd` is sent when the classifier model is priced.
- **Async clients** run on the event loop that ends the interaction. Where no loop runs (a sync route in a thread pool, a worker), the classification is skipped with a warning: use a sync client there.
- **No text.** An interaction with neither prompt nor answer text is skipped; the first one logs a warning (classification needs prompt or answer text).
- **Failures and load.** A timeout, unparseable reply or a `classify` that raises sends nothing; the first of each kind logs a warning. 16 classifications run at once and 1,000 wait; beyond that, work is dropped with a warning (at most once a minute). An invalid `classifier` logs a warning and leaves classification off.

`flush()` / `aflush()` wait for pending classifications before flushing.

### Declared tool events

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

```python
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 tool name to `ai_response_received`: `call_count`, and `error_count` / `duration_ms` where the integration observes them. Tool names only, never inputs or outputs. Details: [docs/events.md](https://github.com/rudderlabs/ai-sdk-python/blob/main/docs/events.md#tool_breakdown).

## 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:

| Pattern | Replaced by |
| --- | --- |
| Email addresses | `[email]` |
| Phone numbers (international `+` form, US shapes, `98765 43210`) | `[phone]` |
| Payment card numbers (Luhn-valid) | `[card]` |
| IPv4 and IPv6 addresses | `[ip]` |
| US Social Security numbers | `[ssn]` |
| IBANs (checksum-valid) | `[iban]` |
| API keys and tokens (`sk-`, `sk_live_`, `rk_live_`, `AIza`, `xox?-`, `gh?_`, `AKIA`/`ASIA`, JWTs, `Bearer`), passwords in URLs, `api_key=` / `token:` / `password=` values | `[secret]` |

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:

```python
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`](https://github.com/rudderlabs/ai-sdk-python/blob/main/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.

```python
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](https://github.com/rudderlabs/ai-sdk-python/blob/main/docs/tracing.md).

## Serverless and workers

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

```python
import rudderstack.ai


def handler(event, context):  # 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

| Event | When |
| --- | --- |
| `ai_user_prompt_created` | A user-triggered interaction starts (sent with its first model call) |
| `ai_response_received` | The interaction ends: totals, status, models, cost, `usage_breakdown` |
| `ai_user_action` | `track_action()`: feedback, copy, share, report, regenerate |
| `ai_interaction_classified` | The classifier labelled the interaction (opt-in) |
| `ai_llm_response_received` | One per model call (opt-in, `emit_llm_calls=True`; do not sum it with the totals) |
| your declared tool events | One per call of a tool named in `event_rules` |
| `ai_otel_span` | One per exported span (tracing module) |

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](https://github.com/rudderlabs/ai-sdk-python/blob/main/docs/events.md). Load [`spec/rudderstack-catalog/`](https://github.com/rudderlabs/ai-sdk-python/blob/main/spec/rudderstack-catalog) as a tracking plan before the first events arrive, so cost columns are created as decimals.

## Configuration

```python
rudderstack.ai.configure(
    analytics=rudder_analytics,
    prices={"gpt-4.1-mini": {"input_per_mtok": 0.40, "output_per_mtok": 1.60, "cached_input_per_mtok": 0.10}},
    capture_content=False,
    emit_llm_calls=False,
    tool_breakdown=False,
    identity_from_cookies=True,
)
```

| Option | Default | |
| --- | --- | --- |
| `analytics` | none | Your RudderStack client (the `rudderstack.analytics` module or a `Client`). A client passed to an interaction, `ai_context` or integration wins. |
| `prices` | none | USD per 1M tokens, merged into the registry. Your rates; none ship with the package. |
| `capture_content` | `False` | Send prompt, response and feedback text (masked). |
| `emit_llm_calls` | `False` | Also send `ai_llm_response_received` per model call. |
| `tool_breakdown` | `False` | Add `tool_breakdown` to `ai_response_received`. |
| `identity_from_cookies` | `True` | Read the JavaScript SDK's cookies from a request. |
| `redact` | built-in masking | A `(text) -> str` hook run after masking, or `False` to turn masking off. |
| `event_rules` | none | `{"tools": {name: rule}}`: declared tool events. |
| `classifier` | none | Intent classification: `{client, model, intents, ...}` or `{classify, ...}`. |

`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](https://github.com/rudderlabs/ai-sdk-python/blob/main/docs/api.md); every public function has a docstring and type hints.

## Troubleshooting

Turn on the package's logs (`rudderstack.ai`, and `rudderstack.ai.tracing` for spans):

```python
import logging

logging.basicConfig()
logging.getLogger("rudderstack.ai").setLevel(logging.DEBUG)
```

| Symptom | Cause |
| --- | --- |
| No events, warning "no RudderStack client" | `configure(analytics=...)` was not called before the first event. |
| Events missing in a script, worker or Lambda | The process exited before the client's queue was sent: `flush()`. |
| Every model call is its own `ai_response_received` | Calls run outside `ai_interaction`: in a thread pool (use `propagate`), in a stream that outlives the interaction (use `bind()`; a warning names the interaction), or with the LangChain handler on a model instead of the run config. |
| No cost | No price registered for the model, or the name does not match ([Configuration](#configuration)). Streamed calls to a compatible endpoint need `wrap_openai(..., stream_usage=True)`. |
| Cost column holds whole numbers | The warehouse created it as an integer: [docs/events.md](https://github.com/rudderlabs/ai-sdk-python/blob/main/docs/events.md#warehouse-column-types). |
| No `ai_interaction_classified` | The interaction was opened by an integration, the answer did not succeed, or an async client ran without an event loop (warning). |
| Prompt event arrives late | It is held until the first model call starts, so it carries that call's integration tag. |
| `register_tracing()` raises `RuntimeError` | The global provider is not an SDK `TracerProvider`, or already has a RudderStack processor. Use `create_span_processor()`. |

## Compatibility

| | Supported | Tested in CI |
| --- | --- | --- |
| Python | 3.12+ | 3.12, 3.13, 3.14 |
| `openai` | `>=2.0.0,<4` | 2.0.0 and latest |
| `anthropic` | `>=1.0.0,<2` | 1.0.0 and latest |
| `langchain-core` | `>=1.2.0,<2` | 1.2.0 and latest, with LangGraph 1.x and Deep Agents 0.7 |
| `claude-agent-sdk` | `>=0.2.82,<0.3` | 0.2.82 and latest |
| `opentelemetry-api` / `-sdk` / `-exporter-otlp-proto-common` | `>=1.30.0,<2` | 1.30.0 and latest |
| `rudder-sdk-python` | 2.x, or any object with its `track()` signature | 2.1.9 |
| Web frameworks (middleware) | Starlette/FastAPI, Django, Flask | Starlette 1.7, Django 6.1, Flask 3.1 |

Known limitations:

- Not tracked: OpenAI `with_streaming_response`, a streamed `with_raw_response` call and a non-streamed `responses.create(background=True)` (no outcome yet). Use `record_call` for those.
- 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.

## 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`.
- **OpenAI streamed chat** asks for usage only on api.openai.com; pass `stream_usage=True` for a compatible endpoint that supports it.
- **`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](https://github.com/rudderlabs/ai-sdk-python/blob/main/CHANGELOG.md).

## Examples

[`examples/`](https://github.com/rudderlabs/ai-sdk-python/blob/main/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.

## Contributing, security, license

[CONTRIBUTING.md](https://github.com/rudderlabs/ai-sdk-python/blob/main/CONTRIBUTING.md) · [SECURITY.md](https://github.com/rudderlabs/ai-sdk-python/blob/main/SECURITY.md) · [RELEASING.md](https://github.com/rudderlabs/ai-sdk-python/blob/main/RELEASING.md) · [MIT](https://github.com/rudderlabs/ai-sdk-python/blob/main/LICENSE.md)
