trodo-python 1.2.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+.

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

---

Agent Analytics

Track every step of your LLM agents. Each call counts as one event toward your plan limit.

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

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