Carrot AI Python SDK

Automatic tracing for LLM calls. Wrap your OpenAI or Anthropic client — or patch litellm — and every call is captured as a structured trace in the Carrot platform with zero changes to your application code.

Installation

pip install carrot-ai

Or install from source (in the sdk/python/ directory):

pip install -e .

Quick Start

1. Wrap your client (one line)

import carrot_ai
from openai import OpenAI

carrot_ai.init(api_key="sk-...")  # your Carrot API key

client = carrot_ai.wrap(OpenAI())

# Use the client exactly as normal — traces are captured automatically
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

That's it. Every create() call through the wrapped client is captured as a trace with the full request, response, token counts, and latency.

2. Works with Anthropic too

import carrot_ai
from anthropic import Anthropic

carrot_ai.init(api_key="sk-...")

client = carrot_ai.wrap(Anthropic())

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

3. Streaming works transparently

client = carrot_ai.wrap(OpenAI())

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

# Trace is submitted automatically when the stream finishes

4. LiteLLM support

If you use litellm to call 100+ LLM providers through a single interface, one call patches everything:

import carrot_ai
import litellm

carrot_ai.init(api_key="sk-...")
carrot_ai.patch_litellm()

# Every litellm.completion() / litellm.acompletion() is now traced
response = litellm.completion(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

Streaming, token counts, and @trace nesting all work the same way as with wrap().

5. Custom trace names

By default, auto-traced calls use the API resource name (e.g. "chat.completions"). You can set a custom name to identify traces from a specific service or client:

client = carrot_ai.wrap(OpenAI(), name="my-chatbot")

carrot_ai.patch_litellm(name="summarizer-service")

This is useful when you have multiple services sending traces and want to distinguish them in the dashboard.

@trace Decorator

For multi-step pipelines (RAG, agents, chains), use the @trace decorator to capture the full workflow as a single trace. Any wrapped LLM calls inside the function are automatically linked as children.

import carrot_ai
from openai import OpenAI

carrot_ai.init(api_key="sk-...")
client = carrot_ai.wrap(OpenAI())


@carrot_ai.trace
def answer_question(question: str) -> str:
    # Step 1: retrieve context
    docs = search_knowledge_base(question)

    # Step 2: LLM call (auto-traced as a child)
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context:\n{docs}"},
            {"role": "user", "content": question},
        ],
    )

    return response.choices[0].message.content


# The decorator creates a parent trace for the full function,
# and the LLM call inside creates a child trace linked via parent_trace_id
result = answer_question("How do I reset my password?")

Decorator variants

# Bare decorator
@carrot_ai.trace
def my_function():
    ...

# With a custom name
@carrot_ai.trace("my-pipeline")
def my_function():
    ...

# With tags and metadata
@carrot_ai.trace(name="my-pipeline", tags=["production"], metadata={"version": "2"})
def my_function():
    ...

Async support

Both wrap() and @trace work with async clients and async functions:

from openai import AsyncOpenAI

client = carrot_ai.wrap(AsyncOpenAI())


@carrot_ai.trace
async def answer_question(question: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
    )
    return response.choices[0].message.content

Configuration

carrot_ai.init()

Parameter	Type	Default	Description
api_key	str	CARROT_API_KEY env var	Your Carrot API key
base_url	str	https://api.carrotlabs.ai	API endpoint (override for self-hosted)
flush_interval	float	5.0	Seconds between background batch flushes
batch_size	int	50	Max traces per batch POST

Environment variables

Variable	Description
CARROT_API_KEY	API key (used if not passed to init())
CARROT_BASE_URL	API base URL override

If CARROT_API_KEY is set, init() is called automatically on first use — you can skip the explicit init() call.

carrot_ai.wrap()

Parameter	Type	Default	Description
client	OpenAI | Anthropic	required	The LLM client to wrap
name	str	None	Override the default trace name (derived from the API resource, e.g. "chat.completions")
tags	list[str]	None	Default tags added to every trace from this client

carrot_ai.patch_litellm()

Parameter	Type	Default	Description
name	str	None	Override the default trace name ("chat.completions")
tags	list[str]	None	Default tags added to every trace from litellm calls

Patches litellm.completion and litellm.acompletion in-place. Safe to call multiple times (only patches once).

carrot_ai.flush()

Force-flush any pending traces to the server. Useful before process exit in short-lived scripts:

carrot_ai.flush()

An atexit handler also flushes automatically on normal interpreter shutdown.

Proxy Header (Alternative)

If you already route inference through the Carrot proxy, you can capture traces without the SDK by adding a header:

from openai import OpenAI

client = OpenAI(
    base_url="https://api.carrotlabs.ai/v1",
    api_key="sk-...",  # your Carrot API key
    default_headers={"X-Carrot-Trace": "true"},
)

# Traces are captured automatically by the proxy
response = client.chat.completions.create(
    model="my-model",
    messages=[{"role": "user", "content": "Hello!"}],
)

This only works for requests routed through the Carrot proxy. Use the SDK wrap() approach for calls sent directly to OpenAI, Anthropic, or any other provider.

Publishing to PyPI

1. Bump the version in both pyproject.toml and src/carrot_ai/__init__.py.
2. Update CHANGELOG.md with the new entry.
3. Run tests: python3 -m pytest tests/ -v
4. Build: rm -rf dist/ && python3 -m build
5. Publish:

source .env
uv publish --token "$UV_PUBLISH_TOKEN"

Store your PyPI API token in sdk/python/.env (gitignored):

UV_PUBLISH_TOKEN=pypi-...

What Gets Captured

Each trace includes:

Field	Description
input	Messages, model, temperature, and other request params
output.message	Assistant response (content + tool calls)
metadata.provider	"openai", "anthropic", or "litellm"
metadata.model	Model name used
metadata.input_tokens	Prompt token count
metadata.output_tokens	Completion token count
metadata.latency_ms	End-to-end call duration
metadata.source	"sdk", "decorator", or "proxy"
metadata.parent_trace_id	Links child LLM calls to @trace parent
status	"success" or "error"
started_at / ended_at	ISO 8601 timestamps