Avenza Python SDK

Instrument AI agents in one line. Cost, value, and SLO tracking — automatically.

Install

pip install avenza

Quickstart (3 lines)

from avenza import Agent

agent = Agent(name='Invoice Bot', risk_tier='T2')

with agent.run() as run:
    result = process_invoice(data)
    run.success = result.is_valid
    run.log_value('task_completed', quantity=1, unit_value_usd=1.50)

That's it. If you're using the official Anthropic, OpenAI, or Gemini client, token usage is captured automatically with zero additional code.

Setup wizard

avenza init

Interactive wizard that verifies your API key and writes a working starter script — not a docs page to interpret.

Diagnose

avenza doctor

Self-diagnose connection issues, missing instrumentation, and proxy configuration.

Auto-instrumentation

The SDK patches the official provider clients the moment it's imported:

import anthropic
from avenza import Agent

agent = Agent(name='Support Bot')
client = anthropic.Anthropic()

with agent.run() as run:
    # Token usage captured automatically from this call
    response = client.messages.create(model='claude-sonnet-4-6', ...)
    run.success = True

Works across threads and async/await via Python's contextvars.

Manual token fallback

with agent.run() as run:
    response = my_llm_client.call(...)
    run.set_tokens(response.input_tokens, response.output_tokens)
    run.success = True

LangChain

from avenza.integrations.langchain import AvenzaCallbackHandler
from langchain_anthropic import ChatAnthropic

handler = AvenzaCallbackHandler(agent_name='Support Classifier', risk_tier='T1')
llm = ChatAnthropic(model='claude-sonnet-4-6', callbacks=[handler])
response = llm.invoke('Classify this ticket: ...')

Testing

from avenza.testing import MockAgent

def test_my_agent():
    agent = MockAgent(name='Invoice Bot')
    with agent.run() as run:
        run.success = True
        run.log_value('task_completed', quantity=1, unit_value_usd=1.50)

    assert agent.runs[-1].success is True
    assert agent.runs[-1].value_events[0]['quantity'] == 1

Configuration

Redaction

If your agent handles sensitive data — patient names, SSNs, account numbers — in set_metadata() or escalate(reason=...), tell the SDK which field names to scrub before anything leaves your process:

agent = Agent(name="Intake Bot", redact_fields=["ssn", "patient_name", "email"])

with agent.run() as run:
    run.set_metadata(patient_name="Jane Doe", ssn="123-45-6789", confidence=0.94)
    # payload sent to Avenza has patient_name and ssn replaced with "[REDACTED]"
    # confidence is untouched — only field names you list are affected

Matching is case-insensitive and recursive (checked inside meta and value_events, at any nesting depth). Redaction only affects the outgoing copy — run._metadata in your own process still has the real values, so your agent's own logic is unaffected.

Never blocks. Never raises.

Every network call is fire-and-forget on a background thread. Avenza being down, slow, or returning errors will never crash your agent or add latency to your agent's actual work. Failed sends are buffered to .avenza_buffer.jsonl and retried on next startup.

If a send fails and there's no offline buffer to catch it (offline_buffer=False, or the buffer write itself fails — e.g. read-only filesystem), the payload is genuinely lost rather than silently retried. That's tracked, not hidden:

if agent.dropped_count > 0:
    logger.warning(f"Avenza lost {agent.dropped_count} governance record(s)")

dropped_count should stay 0 in normal operation. A nonzero value under sustained load is a signal to leave offline_buffer=True (the default) rather than disabling it.