Custom Python agent

What Dunetrace captures

Behavioral failures in AI agents — tool loops, reasoning stalls, context bloat, and 12 more patterns — within ~15 seconds of a run completing. It never transmits raw prompts or outputs: all user content, tool arguments, and completions are SHA-256 hashed in-process before any data leaves your agent.

What does transmit: model names, token counts, latencies, tool names, finish reasons, step counts.

Step 1. Generate an API key

API keys live in the api_keys table. No UI yet — insert a row directly.

INSERT INTO api_keys (key, agent_id, customer_id)
VALUES ('dt_live_<random>', 'my-production-agent', 'my-company');

Generate a secure suffix:

python3 -c "import secrets; print('dt_live_' + secrets.token_hex(16))"

To revoke: UPDATE api_keys SET active = FALSE WHERE key = 'dt_live_...';

Step 2. Install the SDK

pip install dunetrace
pip install 'dunetrace[otel]'   # if you want OpenTelemetry export

Step 3. Pick an integration path

Path A · Decorator (recommended)

from dunetrace import Dunetrace

dt = Dunetrace(
    endpoint="https://your-dunetrace-ingest",
    api_key="dt_live_...",
)
dt.init(agent_id="my-production-agent")
dt.auto_instrument()   # patches openai, anthropic, httpx, requests

@dt.agent(model="gpt-4o", tools=["web_search", "calculator"])
def run_agent(query: str) -> str:
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content

result = run_agent("What is the capital of France?")
dt.shutdown()

Async agents use identical syntax — async def and await inside.

Path B · FastAPI / ASGI middleware

from dunetrace import Dunetrace, DunetraceASGIMiddleware
from dunetrace.context import get_current_run
from fastapi import FastAPI

dt = Dunetrace(endpoint="https://…", api_key="dt_live_...")
dt.auto_instrument()

app = FastAPI()
app.add_middleware(
    DunetraceASGIMiddleware,
    dt=dt, agent_id="my-api-agent", model="gpt-4o",
)

@app.post("/chat")
async def chat(query: str):
    run = get_current_run()        # opened automatically by middleware
    run.tool_called("db_lookup", {"query": query})
    result = await db.get(query)
    run.tool_responded("db_lookup", success=True, output_length=len(str(result)))
    return result

For Flask/Django, use DunetraceWSGIMiddleware.

Path C · Manual dt.run()

with dt.run("my-agent", user_input=query, model="gpt-4o", tools=TOOLS) as run:
    run.llm_called("gpt-4o", prompt_tokens=150)
    response = call_llm(query)
    run.llm_responded(completion_tokens=30, latency_ms=820,
                      finish_reason="tool_calls", output_length=len(response))

    run.tool_called("web_search", {"query": query})
    result = web_search(query)
    run.tool_responded("web_search", success=True, output_length=len(result), latency_ms=300)

    run.final_answer()

Full RunContext API

# LLM events
run.llm_called(model, prompt_tokens)
run.llm_responded(completion_tokens, latency_ms, finish_reason, output_length)

# Tool events
run.tool_called(tool_name, args)           # args dict is hashed in-process
run.tool_responded(tool_name, success, output_length, latency_ms, error)

# Retrieval
run.retrieval_called(index_name, query_hash)
run.retrieval_responded(index_name, result_count, top_score, latency_ms)

# Infra signals (do not advance step counter)
run.external_signal("rate_limit", source="openai")

# Terminal marker
run.final_answer()

Path D · LangChain

See the dedicated LangChain guide.

Path E · OTel receiver

If your agent already emits gen_ai.* spans via OpenLLMetry:

from dunetrace.integrations.otel_receiver import DunetraceOTelReceiver

DunetraceOTelReceiver.attach(tracer_provider, dt, agent_id="my-agent")

No agent-code changes. See Integrations.

Shutdown gracefully

import atexit
atexit.register(dt.shutdown)
# or
dt.shutdown(timeout=5)

Privacy summary

Checklist

• Generate an API key via INSERT INTO api_keys …
• pip install dunetrace
• Instantiate Dunetrace(endpoint=…, api_key=…)
• dt.init(agent_id=…) and dt.auto_instrument()
• Wrap agent (decorator / middleware / dt.run())
• Call dt.shutdown() on exit
• Set SLACK_WEBHOOK_URL on the server
• Tune detectors.yml if needed
• Test locally against http://localhost:8001 first