Metadata-Version: 2.4
Name: tridente
Version: 0.1.0
Summary: Python SDK for Tridente — report usage, cost, and heartbeat events for AI agents.
Project-URL: Homepage, https://tridente.dev
Project-URL: Repository, https://github.com/idiaz01/tridente
Project-URL: Issues, https://github.com/idiaz01/tridente/issues
Author: Ivan Diaz
License: BUSL-1.1
License-File: LICENSE
Keywords: ai-agents,cost-tracking,governance,observability,tridente
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.7
Requires-Dist: typing-extensions>=4.10; python_version < '3.11'
Provides-Extra: all
Requires-Dist: anthropic>=0.34; extra == 'all'
Requires-Dist: langchain-core>=0.2; extra == 'all'
Requires-Dist: openai>=1.40; extra == 'all'
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.34; extra == 'anthropic'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2; extra == 'langchain'
Provides-Extra: openai
Requires-Dist: openai>=1.40; extra == 'openai'
Description-Content-Type: text/markdown

# tridente

[![PyPI](https://img.shields.io/pypi/v/tridente)](https://pypi.org/project/tridente/)
[![Python](https://img.shields.io/pypi/pyversions/tridente)](https://pypi.org/project/tridente/)
[![License: BUSL-1.1](https://img.shields.io/badge/license-BUSL--1.1-blue)](../../LICENSE)

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

## Install

```bash
pip install tridente
```

Optional integrations:

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

```python
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`):

```python
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

```python
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:

```python
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:

```python
@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:

```python
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:

```python
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:

```python
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

```python
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:

```python
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):

```python
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:

```python
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:

```python
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`](../../sdk-contract/)
pins the format; if either SDK diverges, both parity tests break.

## Error handling

```python
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](../../LICENSE).
