mainwave 0.1.0rc5

Python SDK for the Mainwave AI observability platform

mainwave

Python SDK for the Mainwave AI observability platform.

Install

pip install "mainwave[openai]"        # one platform
pip install "mainwave[langchain]"     # LangChain + a callback handler
pip install "mainwave[all]"           # everything (kick-the-tires only)

Per-platform pinning is recommended in production.

Quickstart

import mainwave

mainwave.init(
    api_key="...",                                       # or MAINWAVE_API_KEY
    use_case_id="00000000-0000-0000-0000-000000000001",  # or MAINWAVE_USE_CASE_ID
    collector_url="https://collector.mainwave.com",      # default
)

with mainwave.run("INV-12345", metadata={"customer": "acme"}):
    # ... do work; auto-instrumented LLM calls land under this run
    ...

mainwave.flush()  # before exit

init() auto-wires every supported instrumentor that's already importable, so the snippet above needs no instrument() call. Reach for instrument() only to wire a framework imported after init() ran (lazy import, CLI subcommand, FastAPI lifespan) — see Public API.

mainwave.run(...) is optional — without it, every OTel trace becomes its own logical run. Wrap explicitly when one logical operation spans multiple traces (loops, fan-out, multiple LLM invocations per request); the integration guide wraps every agent execution for exactly this reason.

Not seeing data?

The SDK never raises into your app, so a misconfigured endpoint or key fails quietly. Turn on its logger to see what's happening:

import logging
logging.getLogger("mainwave").setLevel(logging.INFO)   # add a handler if you have none

At INFO it logs the first successful export, which instrumentors wired, and a WARNING for any framework that didn't. mainwave.flush() returns False on a failed drain — check it before a hard exit. Confirm collector_url and the API key, and that the framework's extra is installed (pip install "mainwave[openai]").

Process exit and signals

init() registers an atexit hook that flushes buffered spans on normal interpreter shutdown. It also installs a SIGTERM handler (main thread only) so containers stopped with SIGTERM still flush before exit. The handler chains: it flushes, then hands control back to whatever SIGTERM disposition was in place before init() ran — if your app had its own handler, it runs after the flush and owns the exit; if SIGTERM was default, the process exits with the conventional 128+signum status; if it was ignored, the SDK stays out of the way. Repeat init() calls don't re-install, so the SDK never chains to itself.

Set MAINWAVE_DISABLE_SIGNAL_HANDLER=1 to skip installing the SIGTERM handler entirely (the atexit flush still registers) — use this if your process manages SIGTERM itself and doesn't want the SDK touching it.

Public API

• mainwave.init(...) — set up the SDK once, at process start.
• mainwave.instrument(name=None) — re-wire instrumentors after late imports.
• mainwave.run(...) / mainwave.arun(...) / mainwave.start_run(...) — open a logical run (sync CM, async CM, imperative).
• mainwave.span(name, attrs=...) — escape hatch for manual spans.
• mainwave.flush(timeout=30.0) / mainwave.aflush(timeout=30.0) — drain buffers.
• mainwave.langchain.CallbackHandler — optional explicit LangChain handler.
• mainwave.testing — reset() + install_in_memory_provider() for tests.

LangChain users

Use the native CallbackHandler (captures model reasoning + step trajectory) OR auto-instrumentation (instrument(name="langchain")) — not both, or every call is double-counted. You don't have to police it: attaching the handler disables the OpenInference LangChain instrumentor automatically, and init(skip_instrumentors=["langchain"]) skips wiring it up front.

Several agents in one process

init() sets one use_case_id and service_name for the whole process. When a single process runs more than one agent — and you want each reported as its own use case or agent — override per invocation on the LangChain handler:

divergence = mainwave.langchain.CallbackHandler(
    use_case_id="…",        # which use case these spans roll up to
    agent_name="divergence",  # which agent within it (stamped as gen_ai.agent.name)
)
chain.invoke(payload, config={"callbacks": [divergence]})

A handler's use_case_id / agent_name apply only to the spans it emits, so several agents can share one use case (distinguished by agent_name) or land in different use cases from the same process. Omit either to inherit the init() default; an empty value is ignored rather than applied. (service_name= is still accepted as a deprecated alias for agent_name.)

Running inside a service that already uses OpenTelemetry

If your process already has its own OpenTelemetry setup (FastAPI/DB auto-instrumentation, background-job spans, …), the SDK coexists with it. It attaches to your existing TracerProvider rather than replacing it, and it exports a trace to Mainwave only when that trace involves an LLM call — your service's own infrastructure traces stay out of Mainwave. When a trace does contain an LLM call, it's sent in full, so the surrounding request span that triggered the call is preserved and correlated. This mirrors how Langfuse filters export on a shared provider; you don't have to isolate the SDK or wire a separate provider.

See also

Spec: docs/agent_sdk/ in the platform monorepo.