agentracer

Agentracer Python SDK

Lightweight AI observability. Catch cost spikes, latency anomalies, and prompt bloat before they hit your users.

Installation

pip install agentracer

With provider-specific extras:

# OpenAI support
pip install agentracer[openai]

# Anthropic support
pip install agentracer[anthropic]

# Google Gemini support
pip install agentracer[gemini]

# All providers
pip install agentracer[all]

Quick Start

import agentracer

# Initialize once at app startup
agentracer.init(
    tracker_api_key="at_your_api_key",
    project_id="your_project_id",
    environment="production"
)

Usage

OpenAI

from agentracer.openai import openai

# Use exactly like the normal OpenAI client
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    feature_tag="chatbot"  # optional: tag this call
)

Anthropic

from agentracer.anthropic import anthropic

response = anthropic.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
    feature_tag="summarizer"  # optional: tag this call
)

Google Gemini

import google.generativeai as genai
from agentracer.gemini import gemini

# Configure your API key as usual
genai.configure(api_key="your-gemini-api-key")

model = gemini.GenerativeModel("gemini-1.5-pro")
response = model.generate_content(
    "Hello!",
    feature_tag="content-gen"  # optional: tag this call
)

Custom Client Configuration

The default imports (openai, anthropic) create clients using environment variables. To pass custom configuration like api_key or base_url, use the class constructors:

OpenAI

from agentracer.openai import TrackedOpenAI

openai = TrackedOpenAI(api_key="sk-...", base_url="https://custom.api/v1")

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

Anthropic

from agentracer.anthropic import TrackedAnthropic

anthropic = TrackedAnthropic(api_key="sk-ant-...")

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

Async Support

For async applications, use the async client wrappers:

Async OpenAI

from agentracer.openai import TrackedAsyncOpenAI

client = TrackedAsyncOpenAI()

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

Async Anthropic

from agentracer.anthropic import TrackedAsyncAnthropic

client = TrackedAsyncAnthropic()

async def chat(message: str):
    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
        feature_tag="async-summarizer"
    )
    return response.content[0].text

Streaming

All providers support streaming. Token usage is automatically tracked after the stream completes.

OpenAI Streaming

from agentracer.openai import openai

stream = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
    feature_tag="poet"
)

for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="")
# Telemetry is sent automatically after the stream ends

Anthropic Streaming

from agentracer.anthropic import anthropic

stream = anthropic.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
    feature_tag="poet"
)

for event in stream:
    if event.type == "content_block_delta":
        print(event.delta.text, end="")

Gemini Streaming

from agentracer.gemini import gemini

model = gemini.GenerativeModel("gemini-1.5-pro")

stream = model.generate_content(
    "Write a poem",
    stream=True,
    feature_tag="poet"
)

for chunk in stream:
    print(chunk.text, end="")

Streaming works transparently -- usage is captured from the final chunk (OpenAI), SSE events (Anthropic), or chunk metadata (Gemini), then sent as a single telemetry event after the stream finishes.

Feature Tags

Feature tags let you group and track costs by feature across your application. There are three ways to apply them:

1. Inline (per-call)

Pass feature_tag directly on any tracked call:

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this"}],
    feature_tag="summarizer"
)

2. Decorator

Use the @observe decorator to tag all LLM calls inside a function:

import agentracer

@agentracer.observe(feature_tag="chatbot")
def handle_chat(message: str):
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}]
    )
    return response

# Async functions work too
@agentracer.observe(feature_tag="async-chatbot")
async def handle_chat_async(message: str):
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}]
    )
    return response

3. Context Manager

Use feature_context for block-level tagging:

from agentracer import feature_context

with feature_context("report-generator"):
    # All LLM calls in this block get tagged "report-generator"
    summary = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Summarize the data"}]
    )
    analysis = anthropic.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": "Analyze trends"}]
    )

Agent Runs

Track multi-step agent workflows as a single run with AgentRun. Each LLM call inside the run is automatically recorded as a step.

from agentracer import AgentRun
from agentracer.openai import openai

with AgentRun(run_name="research-agent", feature_tag="research") as run:
    # Step 1: Plan
    plan = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Plan a research outline"}]
    )

    # Step 2: Execute
    result = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Research: {plan.choices[0].message.content}"}]
    )

    # Step 3: Summarize
    summary = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Summarize: {result.choices[0].message.content}"}]
    )
# All 3 steps are tracked under a single run with timing, tokens, and status

Async Agent Runs

AgentRun also works as an async context manager:

from agentracer import AgentRun
from agentracer.openai import TrackedAsyncOpenAI

client = TrackedAsyncOpenAI()

async def run_agent(query: str):
    async with AgentRun(run_name="async-agent", feature_tag="agent") as run:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}]
        )
        return response.choices[0].message.content

AgentRun Parameters

What gets tracked per run

• Start/end timestamps and total duration
• Status: completed or failed (auto-detected from exceptions)
• Each step: model, provider, tokens, latency, success/failure
• Aggregates: total steps, total tokens, total cost, total latency

Manual Tracking

For custom integrations or providers not yet supported, use track() directly:

import agentracer

agentracer.track(
    model="gpt-4o",
    input_tokens=150,
    output_tokens=300,
    latency_ms=420.5,
    feature_tag="custom-pipeline",
    provider="openai",
    end_user_id="user-123",     # optional: track per-user costs
    cached_tokens=50,           # optional: cached/prompt-cache tokens
)

FastAPI Example

from fastapi import FastAPI
import agentracer
from agentracer.openai import TrackedAsyncOpenAI

app = FastAPI()
client = TrackedAsyncOpenAI()

agentracer.init(
    tracker_api_key="at_your_api_key",
    project_id="your_project_id",
    environment="production"
)

@app.post("/chat")
@agentracer.observe(feature_tag="chatbot")
async def chat(message: str):
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}]
    )
    return {"reply": response.choices[0].message.content}

@app.post("/summarize")
async def summarize(text: str):
    with agentracer.feature_context("summarizer"):
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Summarize: {text}"}]
        )
    return {"summary": response.choices[0].message.content}

@app.post("/agent")
async def agent(query: str):
    async with agentracer.AgentRun(run_name="qa-agent", feature_tag="agent") as run:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}]
        )
    return {"answer": response.choices[0].message.content}

Django Example

# settings.py
import agentracer

agentracer.init(
    tracker_api_key="at_your_api_key",
    project_id="your_project_id",
    environment="production"
)

# views.py
from agentracer.openai import openai
from agentracer import feature_context, AgentRun

def chat_view(request):
    message = request.POST.get("message")

    with feature_context("django-chatbot"):
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": message}]
        )

    return JsonResponse({"reply": response.choices[0].message.content})

def agent_view(request):
    query = request.POST.get("query")

    with AgentRun(run_name="django-agent", feature_tag="agent"):
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}]
        )

    return JsonResponse({"answer": response.choices[0].message.content})

Configuration

agentracer.init(
    tracker_api_key="at_your_api_key",   # Required: your API key from the dashboard
    project_id="your_project_id",         # Required: your project ID
    environment="production",             # Optional: environment name (default: "production")
    host="https://api.agentracer.dev",    # Optional: API host (default: production)
    debug=False,                          # Optional: enable debug logging (default: False)
    enabled=True                          # Optional: enable/disable tracking (default: True)
)

What We Track

Every LLM call automatically captures:

We never capture prompts, responses, or any PII. Only metadata and usage metrics.

Troubleshooting

Telemetry not appearing in the dashboard

1. Check your API key and project ID -- ensure they match what's in your dashboard.
2. Enable debug mode to see what's being sent:

   agentracer.init(
       tracker_api_key="at_your_api_key",
       project_id="your_project_id",
       debug=True
   )
   

3. Check that tracking is enabled -- make sure you haven't set enabled=False.
4. Verify network connectivity -- the SDK sends data to https://api.agentracer.dev/api/ingest.

Import errors for provider modules

If you see ImportError: openai package not installed, install the provider extra:

pip install agentracer[openai]
# or
pip install agentracer[anthropic]
# or
pip install agentracer[gemini]

Feature tags showing as "unknown"

Make sure you're setting feature tags using one of the three methods:

• Inline: feature_tag="my-feature" on the call
• Decorator: @agentracer.observe(feature_tag="my-feature")
• Context manager: with agentracer.feature_context("my-feature"):

SDK not throwing errors

This is by design. The SDK uses a fire-and-forget pattern and silently catches all exceptions to ensure it never impacts your application's performance or reliability.

License

MIT