Python¶
For scripts, scheduled jobs, AI agents, and LangChain chains. The emitter
module is stdlib only: no dependencies to manage, nothing to pin.
Download the single file lumatrack_emitters.py
into your project.
The client¶
from lumatrack_emitters import LumaTrackClient
client = LumaTrackClient("https://your-lumatrack-host", "lmt_...")
# or set LUMATRACK_URL and LUMATRACK_API_KEY and call LumaTrackClient()
client.record_run(
"invoice-agent",
duration_seconds=12,
external_id="job-8841",
metadata={"tokens": 4210},
)
record_run is one POST to /api/v1/runs and returns the response body
as a dict. It raises urllib.error.HTTPError on 4xx/5xx, deliberately:
a tracking call that fails silently is how value quietly stops being
counted. Catch it if your job must not die on a tracking hiccup, but log
what you catch.
Constructor extras: source="python" labels the runs (override per
system, e.g. source="airflow"), and timeout=10 is the HTTP timeout in
seconds. record_run also accepts executed_at (ISO 8601) for evidence
that genuinely happened earlier.
Two robustness guarantees: if the failure report inside track_run itself
fails, your ORIGINAL exception still propagates (the telemetry error goes
to stderr); and the LangChain handler never raises into your chain, and
reports once per tracked invocation even when child chains propagate
callbacks.
Wrap any block: track_run¶
from lumatrack_emitters import LumaTrackClient, track_run
client = LumaTrackClient("https://your-lumatrack-host", "lmt_...")
with track_run(client, "invoice-agent", metadata={"model": "gpt-5"}):
run_my_agent()
The context manager times the block and reports one run. If the block
raises, it reports a failure (with the duration) and re-raises the
exception unchanged. Each entry generates its own external_id, so a
crashed-and-rerun job counts as two honest runs, not a duplicate.
LangChain¶
from lumatrack_emitters import LumaTrackClient, LumaTrackLangChainHandler
client = LumaTrackClient("https://your-lumatrack-host", "lmt_...")
handler = LumaTrackLangChainHandler(client, "support-triage-agent")
chain.invoke(inputs, config={"callbacks": [handler]})
The handler reports on chain end (success) and chain error (failure),
using LangChain's run_id as the external_id and attaching token usage
to metadata when the LLM provider exposes it. It is duck-typed, so the
emitter module stays dependency-free.
OpenAI Agents SDK, CrewADK, everything else¶
Wrap the agent invocation in track_run. That is the whole recipe:
with track_run(client, "ticket-resolver", metadata={"framework": "openai-agents"}):
result = runner.run(agent, task)
If your framework exposes token counts after the call, prefer an explicit
client.record_run(...) after the invocation so you can put them in
metadata.