trodo-python 2.13.0

Trodo Analytics SDK for Python — server-side event tracking

trodo-python

Server-side Python SDK for Trodo Analytics. Track backend events, identify users, manage people/groups, and instrument AI agents — all unified with your frontend data under the same site_id.

Installation

pip install trodo-python

Requires Python 3.8+.

OpenTelemetry / OTLP path (NEW in 2.4.0)

Already running OTel? Skip the Trodo SDK install entirely and point your existing pipeline at Trodo. Two env vars:

export OTEL_EXPORTER_OTLP_ENDPOINT=https://sdkapi.trodo.ai
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer ${TRODO_SITE_ID}"

The Bearer token is your site_id — same value you'd pass to trodo.init(site_id=...). Get it from the Integration Manager.

Use this when you already run OTel (Datadog, Jaeger, Honeycomb) and want Trodo as an additional destination. Install trodo-python and call trodo.register_otel(site_id=..., mode='otlp') to attach our OTLP exporter without replacing your existing setup. wrap_agent then routes through OTel so auto-instrumented children share the same trace.

# At app startup (after your existing OTel provider is registered)
import os, trodo

trodo.register_otel(
    site_id=os.environ['TRODO_SITE_ID'],
    mode='otlp',
)

mode='otlp' requires the optional [otlp] extras:

pip install 'trodo-python[otlp]'

The SDK raises a friendly install hint if you call mode='otlp' without them.

For richer Trodo features on top (wrap_agent, feedback, track_mcp), continue with the SDK quick start below.

Quick Start

import trodo

trodo.init(site_id='your-site-id')

# User-bound context (recommended)
user = trodo.for_user('user-123')
user.track('purchase_completed', {'amount': 99.99, 'plan': 'pro'})
user.people.set({'plan': 'pro', 'company': 'Acme'})

# Flush before process exit if using batching
trodo.shutdown()

Core API

trodo.init(config)

Call once at app startup.

Parameter	Default	Description
site_id	required	Your Trodo site ID
api_base	https://sdkapi.trodo.ai	API base URL
timeout	10 s	HTTP request timeout
retries	2	Retries on network/5xx errors
auto_events	False	Hook sys.excepthook / threading.excepthook as server_error events
batch_enabled	False	Queue events and flush in batches
batch_size	50	Flush when this many events are queued
batch_flush_interval	5.0 s	Also flush every N seconds
on_error	—	Callable on API errors (silent by default)
debug	False	Log API calls to stderr

trodo.for_user(distinct_id, session_id=None)

Returns a user-bound context. No API call is made until you track an event.

user = trodo.for_user('user-123', session_id=request.cookies.get('trodo_session'))

trodo.identify(identify_id, session_id=None)

Creates the session and fires POST /api/sdk/identify. Use to link a distinct_id to an external identifier (email, DB id). Returns the user context.

user = trodo.identify('[e-mail]', session_id=request.cookies.get('trodo_session'))
# distinct_id is now [e-mail] — merges with browser events
user.track('login')

User context methods

user.track(event_name, properties=None)         # Custom event
user.identify(identify_id)                      # Merge identity
user.wallet_address(address)                    # Set wallet address
user.reset()                                    # Clear session
user.capture_error(exc, severity='error')       # Track server_error ('critical' | 'error' | 'warning')

# People profile
user.people.set(properties)
user.people.set_once(properties)
user.people.unset(keys)
user.people.increment(key, amount=1)
user.people.append(key, values)
user.people.union(key, values)
user.people.remove(key, values)
user.people.track_charge(amount, properties=None)
user.people.clear_charges()
user.people.delete_user()

# Groups
user.set_group(group_key, group_id)
user.add_group(group_key, group_id)
user.remove_group(group_key, group_id)
group = user.get_group(group_key, group_id)
group.set(properties)
group.set_once(properties)
group.increment(key, amount=1)
group.append(key, values)
group.union(key, values)
group.remove(key, values)
group.unset(keys)
group.delete()

Direct call pattern

trodo.track('user-123', 'event_name', {'key': 'value'})
trodo.people_set('user-123', {'plan': 'pro'})
trodo.set_group('user-123', 'company', 'acme')

---

AI Agent Tracing (recommended)

One wrap around your agent captures every LLM call, tool call, and nested step as a tree of spans — token counts, costs, inputs, outputs, errors. Works with any stack: OpenAI, Anthropic, LangChain, LlamaIndex, Gemini, raw HTTP, custom tools. Cost is derived server-side from (provider, model) — the SDK only sends tokens.

30-second quickstart

import trodo
trodo.init(site_id='your-site-id')   # auto-instrument on by default

with trodo.wrap_agent('customer-support',
                       distinct_id=user_id,
                       conversation_id=session_id) as run:
    run.set_input({'query': 'where did sales drop'})
    answer = agent.run(query)        # OpenAI/Anthropic/LangChain auto-captured
    run.set_output(answer)

Open the Agent Runs dashboard — the row shows tokens in/out, cost, span count, tool count, error count, plus the full trace tree.

Auto-instrumentation

trodo.init() calls enable_auto_instrument() which registers every installed OpenTelemetry instrumentor. No extra code required.

Framework	Install
OpenAI	pip install opentelemetry-instrumentation-openai
Anthropic	pip install opentelemetry-instrumentation-anthropic
LangChain	pip install opentelemetry-instrumentation-langchain
LlamaIndex	pip install opentelemetry-instrumentation-llama-index
Google Gemini	pip install opentelemetry-instrumentation-google-generativeai
Vertex AI	pip install opentelemetry-instrumentation-vertexai
Bedrock	pip install opentelemetry-instrumentation-bedrock
Cohere	pip install opentelemetry-instrumentation-cohere
Mistral	pip install opentelemetry-instrumentation-mistralai
Haystack	pip install opentelemetry-instrumentation-haystack
httpx / requests	bundled — generic HTTP spans for raw-HTTP callers

Opt out with trodo.init(site_id=..., auto_instrument=False).

Span helpers

Typed function wrappers for custom code — every call becomes a span with the args auto-captured as input, return value as output, exception as error. Dual-form: helper and decorator.

# trace — generic span
prepared = trodo.trace('prepare', prepare_fn)(payload)

@trodo.trace('step')
def step(): ...

# tool — tool span (auto tool_name, kind='tool')
run_funnel = trodo.tool('run_funnel_query', run_funnel_query)
result = run_funnel(team_id=1, preset='day7')

@trodo.tool(name='fetch_user')
async def fetch_user(uid): ...

# llm — LLM span, auto-extracts OpenAI / Anthropic / Gemini usage
answer = trodo.llm(
    'answer', call_openai, model='gpt-4o-mini', provider='openai',
)(messages)
# Records input_tokens / output_tokens from response['usage'].

# retrieval — vector search / RAG retriever span
search = trodo.retrieval('vector_search', vector_search)
docs = search(query)

LLM span input: the chat-message list

For LLM spans, set the input to the same messages you send to the model — a chat-message list, not one blob:

span.set_input([
    {"role": "system", "content": system_prompt},        # rules & role
    {"role": "context", "content": retrieved_docs},      # RAG docs (Trodo extension)
    {"role": "user", "content": "Where is my order?"},
    {"role": "assistant", "content": None, "tool_calls": [...]},
    {"role": "tool", "content": '{"status": "shipped"}', "tool_call_id": "c1"},
    {"role": "user", "content": "When will it arrive?"},  # multiple turns are fine
])

Roles: the standard system / user / assistant / tool plus context — a Trodo extension for RAG / retrieved documents. Any order, any number per role; aliases developer / model / function normalise automatically. Trodo embeds the input as a whole and each role separately, which powers the AI-score detectors (system → rule adherence; user → trajectory/echo; context else tool+assistant → grounding, contradiction, factual retention). A plain string still works and is embedded as one vector.

Raw-HTTP escape hatches

If your LLM client isn't OTel-instrumented and you can't wrap it as a function, record a span post-hoc:

resp = httpx.post(url, json=body).json()
trodo.track_llm_call(
    model='gemini-2.5-flash', provider='google',
    input_tokens=resp['usageMetadata']['promptTokenCount'],
    output_tokens=resp['usageMetadata']['candidatesTokenCount'],
    prompt=body['messages'],  # the chat-message list sent to the model
    completion=resp,
)

For advanced cases, get a raw OTel tracer — the Trodo processor is already subscribed:

tracer = trodo.get_tracer('my.module')
with tracer.start_as_current_span('custom') as sp:
    sp.set_attribute('gen_ai.system', 'my-llm')

Cost & token reporting (v2.8.0+)

Trodo computes per-span cost from whatever you report. You don't have to send cost — send tokens and Trodo prices them using the team's Model Price config (Configuration → Model Price), falling back to built-in defaults. Resolution per span, highest priority first:

1. Explicit cost (a final USD number) — used as-is, never recomputed.
2. cost_details (per-category USD breakdown) — authoritative.
3. Tokens (usage_details map, or input_tokens/output_tokens) — priced by the team's configured model price → global default → left unset if unknown.

All token categories live in an open usage_details map. input/output are the defaults; add cache_read, cache_write, reasoning, audio, image, or any custom key. Raw provider field names are fine — the backend normalises them (prompt_tokens→input, cache_read_input_tokens→cache_read, …). Custom keys must match the category name you price in the UI.

# (a) Tokens only — Trodo prices it from the model name. The llm() helper
#     auto-forwards the FULL provider usage object, so cache/reasoning tokens
#     are captured with zero config.
answer = trodo.llm('answer', call_anthropic,
                   model='claude-sonnet-4', provider='anthropic')

# (b) Raw usage object via track_llm_call — same auto-normalisation.
trodo.track_llm_call(
    model='gpt-4o', provider='openai',
    usage=resp['usage'],   # {prompt_tokens, completion_tokens, prompt_tokens_details:{cached_tokens}}
    prompt=body, completion=resp,
)

# (c) Explicit usage map + cache shorthands.
trodo.track_llm_call(
    model='claude-sonnet-4', provider='anthropic',
    usage_details={'input': 1000, 'output': 500},
    cache_read_tokens=200, cache_write_tokens=80,   # → cache_read / cache_write
)

# (d) Pass cost straight through (skip server-side pricing).
trodo.track_llm_call(model='gpt-4o', provider='openai', cost=0.0123)

# (e) Per-category cost breakdown (authoritative).
trodo.track_llm_call(
    model='gpt-4o', provider='openai',
    cost_details={'input': 0.0003, 'output': 0.0005, 'cache_read': 0.00001},
)

Inside a wrap_agent / span block, set the same fields on the handle:

s.set_llm(
    model='gpt-4o', provider='openai',
    usage_details={'input': 1000, 'output': 500},
    cache_read_tokens=200,
    # or: cost=0.0123  /  cost_details={'input': ..., 'output': ...}
)

Override auto-extraction with extract_usage (scalar in/out) or extract_usage_map (open map) on trodo.llm(name, fn, ...).

Cross-service runs

When one service calls another, the downstream service joins the caller's run instead of creating its own. All spans nest under a single timeline in the dashboard.

# Caller (FastAPI / Flask / Django / …) — outbound:
import httpx
httpx.post(url, headers=trodo.propagation_headers(), json=body)

# Downstream (FastAPI):
from fastapi import FastAPI
app = FastAPI()
app.middleware('http')(trodo.fastapi_middleware())
# Every LLM call / @tool / trace helper inside handlers now nests under
# the caller's run — no extra wiring.

# Or manually:
with trodo.join_run(
    run_id=headers['x-trodo-run-id'],
    parent_span_id=headers['x-trodo-parent-span-id'],
):
    ...

Long-lived sessions across processes — start_run / end_run

wrap_agent is a context manager — it opens and closes the run in one call stack. For sessions that live across many HTTP requests (an MCP server, a websocket-pinned chat, scheduled jobs that resume on different workers), use start_run to open the run from one process and end_run to finalise it later. Between the two, any process can use join_run to add child spans. Same run_id threads through everything.

# Process A — open the run for an MCP session.
run_id = trodo.start_run(
    'external_mcp_session',
    distinct_id=str(user_id),
    conversation_id=mcp_session_id,
)
redis.set(f"mcp:run:{mcp_session_id}", run_id, ex=3600)

# Process B (later, possibly a different worker) — append a tool span.
run_id = redis.get(f"mcp:run:{mcp_session_id}").decode()
with trodo.join_run(run_id, name='tool.run_funnel_query', kind='tool') as span:
    span.set_input(args)
    span.set_output(result)

# When the session ends (timeout sweeper, explicit close):
trodo.end_run(run_id, status='ok')

Conversation binding & feedback

with trodo.wrap_agent(
    'chat', distinct_id=user_id, conversation_id=session_id,
) as run:
    ...
# Later:
trodo.feedback(run.run_id, satisfaction='positive', rating=5)

Cookbook

Runnable scenarios that double as integration tests live in sandbox/scenarios/ — span_helpers.py, raw_http.py, custom_tools.py, cross_service.py, concurrent_100.py, long_run.py, plus opt-in openai_auto.py, anthropic_auto.py, langchain_chain.py. Run them with python -m sandbox.run_all from the SDK root.

---

Prompt Management (v2.9.0+)

Author and version prompts in the Trodo dashboard, then fetch them at runtime so your application never hard-codes prompt text. A deploy label (e.g. production) points at one version; ship a new prompt by moving the label — no redeploy.

import trodo
trodo.init(site_id="your-site-id")

# Latest version (default), a label, or a pinned version number:
prompt = trodo.get_prompt("refund-agent", label="production")

# Fill {{variables}} — unknown tokens are left intact so a missing value shows.
system = prompt.compile(company="Acme", customer_name="Ada")

resp = openai.chat.completions.create(
    model=prompt.config.get("model", "gpt-4o-mini"),
    temperature=prompt.config.get("temperature", 0.2),
    messages=[{"role": "system", "content": system},
              {"role": "user", "content": query}],
)

get_prompt() returns a ManagedPrompt with name, version, labels, template, config, variables, and a .compile(**vars) method. It raises LookupError if the prompt can't be found (you can't run without it).

trodo.get_prompt("refund-agent")                # latest version
trodo.get_prompt("refund-agent", version=3)     # pinned version
trodo.list_prompts()                            # [PromptSummary(name=..., labels=...), ...]
trodo.compile_prompt(template_or_prompt, {...}) # standalone {{var}} substitution

---

Agent Analytics (legacy event-based API)

The older per-event API below is still supported but superseded by wrap_agent + span helpers above. Use it only if you're already wired into it; new integrations should prefer the tracing API.

Before you start: register your agent in Integrations → AI Agents in the dashboard to get an agent_id (agt_xxxxxxxx).

from trodo import (
    AgentCallProps, ToolUseProps, AgentResponseProps,
    AgentErrorProps, FeedbackProps,
)

track_agent_call — inbound message / LLM invocation

trodo.track_agent_call(AgentCallProps(
    agent_id='agt_abc12345',
    conversation_id='conv_xyz',
    message_id='msg_001',
    prompt=user_message,
    model='claude-3-5-sonnet',
    provider='anthropic',
    system_prompt_version='v2',   # optional — track prompt iterations
    distinct_id=user_id,          # optional — link to a Trodo user
    metadata={'thread_source': 'slack', 'locale': 'en'},  # optional — agent_calls.metadata JSONB
))

track_tool_use — tool/function call within a turn

trodo.track_tool_use(ToolUseProps(
    agent_id='agt_abc12345',
    conversation_id='conv_xyz',
    message_id='msg_001',
    tool_name='fetch_billing_info',
    latency_ms=143,
    status='success',             # 'success' | 'failure'
    input={'user_id': '123'},     # optional
    output={'plan': 'pro'},       # optional
))

track_agent_response — LLM output and token usage

trodo.track_agent_response(AgentResponseProps(
    agent_id='agt_abc12345',
    conversation_id='conv_xyz',
    message_id='msg_001',
    model='claude-3-5-sonnet',
    completion_tokens=response.usage.output_tokens,
    prompt_tokens=response.usage.input_tokens,
    total_tokens=response.usage.input_tokens + response.usage.output_tokens,
    finish_reason=response.stop_reason,
    distinct_id=user_id,
))

track_agent_error — errors and failures

import traceback

trodo.track_agent_error(AgentErrorProps(
    agent_id='agt_abc12345',
    conversation_id='conv_xyz',
    message_id='msg_001',
    error_type='rate_limit',       # 'timeout' | 'rate_limit' | 'guardrail_block' | ...
    error_message=str(exc),
    failed_tool='fetch_billing_info',  # optional
    traceback=traceback.format_exc(),  # optional
))

track_feedback — user thumbs up/down

trodo.track_feedback(FeedbackProps(
    agent_id='agt_abc12345',
    conversation_id='conv_xyz',
    message_id='msg_001',          # same message_id as the response it refers to
    feedback='positive',           # 'positive' | 'negative' | 'unreact'
    distinct_id=user_id,
))

Full turn example

import traceback
from trodo import AgentCallProps, ToolUseProps, AgentResponseProps, AgentErrorProps

def run_agent_turn(user_id, conversation_id, user_message):
    agent_id = 'agt_abc12345'
    message_id = f'msg_{int(time.time() * 1000)}'

    trodo.track_agent_call(AgentCallProps(
        agent_id=agent_id, conversation_id=conversation_id,
        message_id=message_id, prompt=user_message, distinct_id=user_id,
    ))

    try:
        trodo.track_tool_use(ToolUseProps(
            agent_id=agent_id, conversation_id=conversation_id,
            message_id=message_id, tool_name='search', status='success', latency_ms=80,
        ))

        response = llm_client.complete(user_message)

        trodo.track_agent_response(AgentResponseProps(
            agent_id=agent_id, conversation_id=conversation_id, message_id=message_id,
            model=response.model,
            completion_tokens=response.usage.output_tokens,
            prompt_tokens=response.usage.input_tokens,
            total_tokens=response.usage.input_tokens + response.usage.output_tokens,
            distinct_id=user_id,
        ))

        return response.text

    except Exception as exc:
        trodo.track_agent_error(AgentErrorProps(
            agent_id=agent_id, conversation_id=conversation_id, message_id=message_id,
            error_type=type(exc).__name__, error_message=str(exc),
            traceback=traceback.format_exc(), distinct_id=user_id,
        ))
        raise

---

Identity Merging (Cross-SDK)

Call identify() with the same value on the browser and server to merge all events under one user profile:

# Python
user.identify('[e-mail]')   # → [e-mail]

# Browser (same value)
# Trodo.identify('[e-mail]') → [e-mail]
# Events from both sides now appear together in the dashboard

---

Flask / FastAPI Example

# Flask
from flask import Flask, request
import trodo

app = Flask(__name__)
trodo.init(site_id='your-site-id')

@app.route('/purchase', methods=['POST'])
def purchase():
    user = trodo.for_user(request.json['user_id'])
    user.track('purchase_completed', {'amount': request.json['amount']})
    return {'ok': True}

# FastAPI
from fastapi import FastAPI, Request
import trodo

app = FastAPI()
trodo.init(site_id='your-site-id')

@app.post('/purchase')
async def purchase(request: Request):
    body = await request.json()
    user = trodo.for_user(body['user_id'])
    user.track('purchase_completed', {'amount': body['amount']})
    return {'ok': True}

---

Batching

trodo.init(
    site_id='your-site-id',
    batch_enabled=True,
    batch_size=50,
    batch_flush_interval=5.0,
)

# Always flush before process exit
import atexit
atexit.register(trodo.shutdown)

---

Auto Events

trodo.init(site_id='your-site-id', auto_events=True)
# Hooks sys.excepthook and threading.excepthook
# Sends server_error events with distinct_id: 'server_global'

# Toggle at runtime
trodo.enable_auto_events()
trodo.disable_auto_events()

---

Thread Safety

The SDK is thread-safe. SessionManager, EventQueue, and BatchFlusher all use threading.Lock internally. Safe for multi-threaded Flask/Django/FastAPI apps.

License

ISC