tridente 0.1.0

Python SDK for Tridente — report usage, cost, and heartbeat events for AI agents.

tridente

Python SDK for Tridente — report usage, cost, and heartbeat events for AI agents with three lines of code.

Install

pip install tridente

Optional integrations:

pip install "tridente[openai]"      # OpenAI / AsyncOpenAI instrumentation
pip install "tridente[anthropic]"   # Anthropic / AsyncAnthropic instrumentation
pip install "tridente[langchain]"   # TridenteCallbackHandler for LangChain
pip install "tridente[all]"         # everything above

Python 3.10+. Built on httpx for both sync and async callers.

Quick start (3 lines)

import tridente

tridente.init(
    api_url="https://tridente.example.com",
    api_key="tri_live_...",
)
tridente.report_usage("agent:default/my-agent", "run")

Events are queued and drained by a background thread every 5 seconds by default; atexit performs a best-effort final flush. To force an immediate drain, call tridente.flush_sync() (or await tridente.flush() from async code).

Or read the credentials from environment variables (TRIDENTE_API_URL / TRIDENTE_API_KEY):

import os
import tridente

os.environ["TRIDENTE_API_URL"] = "https://tridente.example.com"
os.environ["TRIDENTE_API_KEY"] = "tri_live_..."

tridente.init()
tridente.report_usage("agent:default/my-agent", "run")

Core API

import tridente

tridente.init(api_url="...", api_key="...")

# Usage event. Event types must be one of: run, view, clone, reuse.
tridente.report_usage(
    "agent:default/my-agent",
    "run",
    metadata={"request_id": "req_123"},
)

# Cost event. compute_cost_usd may be 0.0 — the server resolves price
# from the model price table.
tridente.report_cost(
    provider="openai",
    model="gpt-4o",
    token_count_input=300,
    token_count_output=120,
    compute_cost_usd=0.0,
    entity_ref="agent:default/my-agent",
)

# Heartbeat (not batched — posts synchronously).
tridente.report_heartbeat(
    "agent:default/my-agent",
    "healthy",
    metadata={"uptime_seconds": 3600, "version": "1.2.3"},
)

Context manager: run()

Automatically time a block of work and emit a run event with duration and any metadata you attach:

with tridente.run("agent:default/my-agent") as ctx:
    ctx.add_metadata(trace_id="trace-xyz", user_id="u1")
    response = my_agent.invoke(query)

If the block raises, the exception propagates unchanged; the run event is still emitted with metadata.error set to the exception class name. Filter on metadata.error IS NOT NULL to compute failure rate.

Decorator: @track()

Wrap a function so every call emits a run event:

@tridente.track("agent:default/my-agent")
def answer_question(q: str) -> str:
    return my_agent.invoke(q)

answer_question("What is CAC?")

Works on both sync and async functions — the decorator detects async def and uses the async code path automatically.

Integration: OpenAI

Three lines to instrument an OpenAI client so every call auto-reports a cost event:

import tridente
from openai import OpenAI
from tridente.integrations.openai import instrument_openai

tridente.init(api_url="...", api_key="...")
client = OpenAI()
instrument_openai(client, entity_ref="agent:default/my-agent")

# Every call now emits a cost event automatically.
client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Supports .chat.completions.create, legacy .completions.create, and the newer .responses.create API. Works on both OpenAI and AsyncOpenAI clients. Idempotent — calling instrument_openai twice on the same client is a no-op.

Integration: Anthropic

Same pattern for the Anthropic SDK. Supports .messages.create on both sync and async clients:

import tridente
from anthropic import Anthropic
from tridente.integrations.anthropic import instrument_anthropic

tridente.init(api_url="...", api_key="...")
client = Anthropic()
instrument_anthropic(client, entity_ref="agent:default/my-agent")

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

Integration: LangChain

Attach TridenteCallbackHandler to a LangChain chain or agent to emit a run event per chain invocation and a cost event per LLM call:

import tridente
from langchain_core.runnables import RunnableConfig
from tridente.integrations.langchain import TridenteCallbackHandler

tridente.init(api_url="...", api_key="...")

handler = TridenteCallbackHandler(entity_ref="agent:default/my-chain")
config = RunnableConfig(callbacks=[handler])

chain.invoke({"question": "What is CAC?"}, config=config)

The handler infers the provider from the model name (claude-* → anthropic, gpt-* / o1-* → openai, else custom). Callbacks are error-contained — a transient reporting failure never crashes your pipeline.

Read path

import tridente

tridente.init(api_url="...", api_key="...")
client = tridente.get_global_client()

entity = client.get_entity_sync("agent:default/my-agent")
print(entity.name, entity.tags, entity.runtime_status)

results = client.search_sync("variance", kind="agent", limit=20)
for e in results.data:
    print(e.id, e.name)

Or explicitly with an instance:

from tridente import Tridente

with Tridente(api_url="...", api_key="...") as client:
    entity = client.get_entity_sync("agent:default/my-agent")

Async usage

Every method has an async twin. Use the async client when you're already inside an event loop (FastAPI, Jupyter, asyncio):

from tridente import Tridente

async with Tridente(api_url="...", api_key="...") as client:
    entity = await client.get_entity("agent:default/my-agent")

    async with tridente.run_async("agent:default/my-agent"):
        await do_async_work()

    await client.flush()

Testing

tridente.testing.MockTridenteClient is a zero-network stand-in:

from tridente.testing import MockTridenteClient, patch_global_client

def test_my_agent_emits_a_run_event():
    with patch_global_client() as mock:
        my_agent()  # calls tridente.report_usage(...) internally
        assert len(mock.usage_events) == 1
        assert mock.usage_events[0].event_type == "run"

There's also a tridente_mock pytest fixture:

from tridente.testing import tridente_mock  # noqa: F401 — fixture

def test_my_agent(tridente_mock):
    my_agent(client=tridente_mock)
    assert len(tridente_mock.usage_events) == 1

Configuration reference

Environment variables

Variable	Description
TRIDENTE_API_URL	Base URL of your Tridente instance
TRIDENTE_API_KEY	API key (prefixed tri_live_ in prod)

Constructor options

Option	Default	Description
api_url	env	API base URL
api_key	env	API key
flush_interval_seconds	5.0	Background flush cadence
batch_size	50	Max events per batch (1-100)
request_timeout_seconds	10.0	Per-request timeout
http_client	None	Bring your own httpx.AsyncClient if you need to

Batching behavior

• Usage + cost events are queued and drained by a background daemon thread every flush_interval_seconds (default 5s).
• Heartbeats POST synchronously — liveness signals need immediate ack.
• Queue is bounded at 10,000 events. Beyond that, the oldest events are dropped (usage first, then cost) and a TridenteTransportWarning is emitted at most once per minute per client.
• atexit hook performs a best-effort flush at process exit.
• Call tridente.flush_sync() / await tridente.flush() (global) or client.flush_sync() / await client.flush() (instance) to force a drain immediately.

Wire-format parity with the TypeScript SDK

Both tridente (Python) and @tridente/sdk (TypeScript) emit byte-identical HTTP request bodies for every event type. A shared contract test at sdk-contract/golden_events.json pins the format; if either SDK diverges, both parity tests break.

Error handling

from tridente import ApiError, ConfigError, TridenteError

try:
    client.get_entity_sync("agent:default/missing")
except ApiError as exc:
    if exc.code == "NOT_FOUND":
        ...
    else:
        raise

Error hierarchy: TridenteError > ApiError / ConfigError / ClosedClientError.

License

BUSL-1.1 (converts to Apache 2.0 after four years). See LICENSE.