Skip to content

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.