Custom Python agent

What Dunetrace captures

Structural failures in AI agents — tool loops, cost spikes, session latency, context bloat, and 19 more patterns — within ~15 seconds of a run completing.

Transmitted: model names, token counts, latencies, tool names, finish reasons, step counts, user input, tool arguments and outputs, and LLM prompts/completions.

Step 1. Generate an API key

Keys are minted over HTTP, in two steps. A fresh install has no keys, so the first one comes from the ingest service's bootstrap endpoint. It is gated on ADMIN_API_KEY from your .env, which is your deployment secret rather than a tenant credential. Omitting scopes mints an admin key, the one scope a fresh install cannot get any other way.

curl -X POST http://localhost:8001/v1/keys \
  -H 'Content-Type: application/json' \
  -d '{"org_id": "my-company", "org_name": "My Company", "admin_key": "'"$ADMIN_API_KEY"'"}'

The response carries key once. It is stored only as a SHA-256 hash and is never logged, so save it now. Keep it for operators.

Then mint a narrower key for the agent itself, using the admin key you just created. An agent needs ingest and nothing else:

curl -X POST http://localhost:8002/v1/keys \
  -H "Authorization: Bearer $DUNETRACE_ADMIN_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"scopes": ["ingest"]}'

Give the agent that second key. An ingest key cannot mint keys, write policies, or decide approvals, and POST /v1/keys returns 403 for any scope the calling key does not itself hold, so an agent key cannot escalate to admin.

To revoke, list keys with GET /v1/keys (admin scope) and call DELETE /v1/keys/{id}.

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 · Zero-instrumentation decorators (@dt.trace / @dt.tool)

Decorate your tool functions with @dt.tool and your agent entry point with @dt.trace. No SDK calls are needed inside any function body — everything is tracked automatically.

from dunetrace import Dunetrace

dt = Dunetrace(endpoint="http://localhost:8001")

@dt.tool                                      # auto-emits tool.called / tool.responded
def web_search(query: str) -> list:
    return search_api(query)

@dt.tool("calculator")                        # explicit tool name
def calc(expr: str) -> float:
    return eval(expr)

@dt.trace                                     # agent_id defaults to function name
def my_agent(question: str) -> str:
    results = web_search(question)
    return str(calc(results[0]))

# Or with an explicit agent ID and model:
@dt.trace("research-agent", model="gpt-4o")
async def async_agent(question: str) -> str:
    results = await async_search(question)
    return results[0]

@dt.tool is a no-op when called outside a dt.run() / @dt.trace context — the function still runs normally with no overhead. Both decorators work on sync and async functions identically.

Path B · Decorator (recommended for most agents)

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, mistral, botocore, langchain, crewai, 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 sent as-is
run.tool_responded(tool_name, success, output_length, latency_ms, error)

# Retrieval
run.retrieval_called(index_name, query)
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)

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