agentracer 0.1.0

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

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

Parameter	Type	Default	Description
run_name	str	None	Name for this run (shown in dashboard)
feature_tag	str	"unknown"	Feature tag applied to all steps
end_user_id	str	None	ID of the end user triggering this run
run_id	str	auto-generated	Custom run ID (UUID v4 generated if omitted)

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

Parameter	Type	Default	Description
tracker_api_key	str	required	Your API key from the Agentracer dashboard
project_id	str	required	Your project ID
environment	str	"production"	Environment name (e.g., "staging", "development")
host	str	"https://api.agentracer.dev"	API endpoint
debug	bool	False	Print telemetry payloads to console
enabled	bool	True	Set to False to disable all tracking

What We Track

Every LLM call automatically captures:

Field	Description
project_id	Your project identifier
provider	LLM provider (openai, anthropic, gemini, custom)
model	Model name (e.g., gpt-4o, claude-sonnet-4-20250514)
feature_tag	Feature label for cost grouping
input_tokens	Number of input/prompt tokens
output_tokens	Number of output/completion tokens
cached_tokens	Prompt-cached tokens (OpenAI, Anthropic, Gemini)
latency_ms	Request latency in milliseconds
success	Whether the call succeeded
error_type	Exception class name on failure
environment	Deployment environment
end_user_id	End user identifier (when provided)

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