Darkhunt telemetry for Python

Python SDK for sending LLM traces, generations, and observations to the Darkhunt platform for persistence and security data enrichment. Built on OpenTelemetry primitives (TracerProvider, BatchSpanProcessor, OTLP/protobuf).

This is the Python analog of @darkhunt-security/telemetry (the TypeScript SDK) — same wire contract and routing semantics, adapted to Python idioms (keyword arguments, with context managers).

• DarkhuntTelemetry — the client. One per process.
• Trace — a single user-facing interaction. Carries routing fields.
• Generation — one LLM round-trip under a trace (model, messages, usage, cost).
• Span — anything else (tool calls, retrievals, guardrails, sub-agents).

Requires Python 3.9+.

Let Claude Code wire it up for you. Install the plugin:

/plugin marketplace add darkhunt-security/darkhunt-telemetry-python
/plugin install darkhunt-telemetry-py@darkhunt-py

Then tell Claude "add Darkhunt telemetry to this service" and the darkhunt-telemetry-python-integration skill auto-invokes and does the steps below for you.

Get started

1. Install

pip install darkhunt-telemetry
# optional Temporal handoff interceptors:
pip install "darkhunt-telemetry[temporal]"

2. Create a singleton client

Construct one client for the lifetime of the process — it spins up a TracerProvider + batch processor, so a per-request client leaks resources.

from darkhunt_telemetry import DarkhuntTelemetry

# api_key is read from DARKHUNT_API_KEY if omitted.
dh = DarkhuntTelemetry(
    tenant_id="t1",
    workspace_id="ws-1",
    application_id="app-1",
    service_name="my-service",          # OTel service.name — one per process/agent
)

Several logical agents in one process? service_name is per client, so they would share a node — name them per trace with agent instead.

In-cluster service-to-service callers post to the permitAll /internal/... path and need no key:

dh = DarkhuntTelemetry(internal=True, tenant_id="t1", workspace_id="ws-1", application_id="app-1")

3. Wrap your LLM calls

The ergonomic form — start_active_generation times the span automatically and makes it the active OTel span (so provider auto-instrumentation nests under it):

trace = dh.trace("chat", session_id=session_id, user_id=user_id)

with trace.start_active_generation("answer", model="claude-sonnet-5") as gen:
    gen.update(input_messages=[{"role": "user", "content": prompt}])
    reply = call_llm(prompt)                       # span is ACTIVE here — timing is real
    gen.end(
        model="claude-sonnet-5",
        output_messages=[{"role": "assistant", "content": reply.text}],
        usage={"input_tokens": reply.input_tokens, "output_tokens": reply.output_tokens},
    )

trace.end()

The manual form still exists for streaming or when you hold a span open across calls. If you open the generation after the work started, backdate it with start_time (epoch seconds, e.g. time.time() captured before the call):

import time
start = time.time()
reply = call_llm(prompt)                           # work happens first
gen = trace.generation("answer", model="claude-sonnet-5", start_time=start)
gen.update(input_messages=[{"role": "user", "content": prompt}])
gen.end(output_messages=[{"role": "assistant", "content": reply.text}], usage=...)

update() is for fields known at start; end() for fields known at finish.

4. Drain the buffer on shutdown

Spans batch in the background. The SDK flushes at process exit (via atexit), but signal-driven shutdown must be wired explicitly:

import signal

def _shutdown(*_):
    dh.shutdown()

signal.signal(signal.SIGTERM, _shutdown)
signal.signal(signal.SIGINT, _shutdown)

For one-shot scripts, dh.flush() before returning is enough.

5. Verify it worked

A clean flush() is not proof of ingestion — the batch processor swallows export errors. Probe the exact ingest endpoint the exporter uses (a 400 on an empty body = auth + routing OK; 401 = wrong/absent key; 404 = missing /trace-hub in the base URL):

curl -s -o /dev/null -w '%{http_code}\n' -X POST \
  -H "Authorization: Bearer $DARKHUNT_API_KEY" \
  -H 'Content-Type: application/x-protobuf' \
  -H "X-Workspace-Id: $DARKHUNT_WORKSPACE_ID" \
  -H "X-Application-Id: $DARKHUNT_APPLICATION_ID" \
  --data-binary '' \
  "$DARKHUNT_BASE_URL/otlp/t/$DARKHUNT_TENANT_ID/v1/traces"

Then open the Darkhunt dashboard and confirm the trace, its generation bubbles, routing attributes, and token/cost.

Span types

Spans nest naturally — parent.span(...) / parent.generation(...) makes the child a child in the trace tree. Every factory also has an active-context variant: start_active_span(...) / start_active_generation(...).

session_id and user_id — set them every time

Routing fields are required; session_id and user_id are technically optional but every integration should set them. They unlock conversation visualization (traces sharing a session_id group into one timeline) and per-user guardrails / anomaly detection. Set them late with trace.update(user_id=..., session_id=...) if not known at open — all spans created after inherit the values.

Configuration

Every option resolves constructor arg > env var > default.

Two rules when overriding base_url: use the ingest API host (api…darkhunt.ai), not the dashboard, and keep the /trace-hub path — the exporter posts to {base_url}/otlp/t/{tenant_id}/v1/traces.

Routing fields can be set once on the client (constant per process) or per-trace (multi-tenant): dh.trace(name, tenant_id=req.tenant_id, ...). dh.trace() raises ValueError if tenant/workspace/application is missing after merging.

Note on register_context_manager. Unlike the Node SDK, Python's OpenTelemetry context is contextvars-based and always active, so span nesting works with nothing to register. This option is kept for parity and only ensures a global W3C propagator exists for traceparent inject/extract.

Data masking

The SDK does not mask data client-side: inputs, outputs, messages, system prompts, metadata, tool arguments, names, tags, and status messages are sent verbatim. PII masking happens server-side in the Darkhunt platform on ingest. Server-side masking covers inputs, outputs, messages, system prompts, tool calls, span names and status messages; metadata values, tags and routing IDs are stored as sent, so keep secrets and PII out of them.

Routing identifiers (session_id, user_id, user_email) round-trip verbatim so the dashboard can group and filter by exact match.

Multi-agent topology (agent handoffs)

When your service is one agent in a multi-agent system, Darkhunt reconstructs the agent topology — who handed off to whom — from the cross-service span tree. The safest, lowest-friction way to draw the edges is to nest each agent's trace under its caller by passing the caller's handoff token:

# Upstream agent: expose its entry-span token after doing its work.
trace = dh.trace("research-agent", session_id=sid, user_id=uid)
token = trace.handoff_token()          # opaque W3C traceparent string
# ... work ...
trace.end()

# Downstream agent: nest under the upstream (parent = handoff_from[0]).
trace = dh.trace("analyst-agent", handoff_from=[token], session_id=sid, user_id=uid)

handoff_from[0] becomes the parent edge (the topology arrow) and an agent_handoff link; further entries are supplementary links (fan-in). Give each agent its own service_name — that string is the topology node.

Several agents in one process

service_name sets the OTel Resource, which is fixed per TracerProvider — i.e. per client. A process hosting several logical agents behind one shared client therefore renders as a single node named after the process.

When that's your shape, name the agent per trace instead. One client, one provider, one Resource:

# One client for the process. Routing stays off it when it varies per request.
dh = DarkhuntTelemetry(service_name="alludium-web")

research = dh.trace(
    "research.run",
    agent="research",             # ← the topology node for this trace
    session_id=sid,               # ← must be shared across the agents in one run
    tenant_id=t, workspace_id=w, application_id=a,
)

scoring = dh.trace(
    "score.deal",
    agent="deal-scoring",
    session_id=sid,               # ← same session
    handoff_from=[research.handoff_token()],
    tenant_id=t, workspace_id=w, application_id=a,
)

This composes with multi-tenant routing: agent and the routing fields are independent, so a host serving many customers from one process passes both per trace — tenant_id / workspace_id / application_id from the request context, agent from whichever logical agent is running. Only service_name stays on the client.

agent emits service.name as a span attribute on the root and on every child span. The backend resolves a trace group's identity from merged attributes, where span attributes outrank the Resource — so each agent becomes its own node while service_name stays the fallback for traces that don't set it.

Two consequences worth knowing before adopting it:

An agent-scoped trace is always a new root. Identity is resolved once per trace id, so two agents sharing a trace would collapse onto whichever name merged first. To make that impossible, passing agent ignores both handoff_from[0] and any ambient active span when parenting. handoff_from still records every upstream as an agent_handoff link, and links are what the topology walks to draw the edge — so the graph is unchanged; the cross-agent parent chain is what you give up. (A trace then covers one agent's slice rather than the end-to-end request.)

Share session_id across the agents in one run. With the parent chain gone, edges come from links alone, and links resolve within a session. Different sessions, no edge.

Use a small, stable set of values ("research", "deal-scoring") — never a request id or anything derived from user input. Each distinct value is a permanent topology node.

Carry the token across a transport in its metadata channel, never the business payload (dependency-free helpers included):

from darkhunt_telemetry.transports import (
    handoff_to_http_headers, handoff_from_http_headers,   # HTTP: W3C traceparent header
    handoff_to_message_meta, handoffs_from_messages,      # Queue: out-of-band message metadata
)

# HTTP producer / consumer
requests.post(url, headers=handoff_to_http_headers(trace.handoff_token(), base_headers))
token = handoff_from_http_headers(request.headers)        # -> dh.trace(handoff_from=[token])

# Queue fan-in
tokens = handoffs_from_messages([m1.headers, m2.headers]) # -> dh.trace(handoff_from=tokens)

Temporal (optional extra): register one interceptor on the worker; each activity reads its upstream via current_handoff().

from temporalio.worker import Worker
from darkhunt_telemetry.temporal import HandoffInterceptor, current_handoff, child_args

worker = Worker(client, task_queue="q", workflows=[...], activities=[...],
                interceptors=[HandoffInterceptor()])

# inside an activity:
trace = dh.trace("recon-agent", handoff_from=current_handoff() or [], session_id=task_id)

The token rides a Temporal Header (out of the business args). A coordinator authors a per-edge override with child_args(input_dict, [upstream_token]); the interceptor relocates it to the header and strips it before the child sees it. Never instrument workflow code (deterministic sandbox) — telemetry lives in activities.

Why the topology may show separate nodes (and that's correct)

Darkhunt draws edges from the cross-service parentSpanId chain. Services that are independent processes with no live agent→agent handoff — a repo of standalone scripts, or a producer/consumer pair coupled only through a datastore or batch boundary (classic RAG ingest→answer) — have no such chain, so they render as separate, unconnected nodes. That's the honest picture, not a misconfiguration. Connecting them is an architecture change (carry a handoff_token() across the boundary), not a telemetry setting.

Development

Uses uv for a fast, reproducible dev environment (pinned by uv.lock); the build backend is hatchling.

uv sync --all-extras          # create .venv from the lockfile (dev + temporal)
uv run pytest                 # tests
uv run ruff check . && uv run ruff format --check .
uv run mypy
uv run bandit -c pyproject.toml -r darkhunt_telemetry
uv build                      # sdist + wheel

Plain pip works too, if you'd rather not use uv:

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,temporal]"
pytest

License

Apache-2.0. See LICENSE and NOTICE.