darkhunt-telemetry-py

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), with built-in client-side data masking that redacts 66 secret/PII patterns before anything leaves your process.

This is the Python analog of @darkhunt-security/telemetry (the TypeScript SDK) — same wire contract, same routing semantics, same masking ruleset, 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
)

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