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_key="invoice-workflow",  # or MAINWAVE_USE_CASE_KEY — a stable name; the SDK derives the use-case UUID
    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.

Naming your use case (use_case_key vs use_case_id)

use_case_id means you bring the UUID (created in the Mainwave app); use_case_key means you bring a stable name and the SDK derives the UUID deterministically — same key, same identity, across runs, restarts, and machines. Passing both raises. The first run under a new key surfaces in the app under Unassigned Agents, named by the key, ready to adopt in one click.

Resolution, innermost wins:

1. explicit use_case_id — anywhere
2. use_case_key= on run() / bind_identity() / the LangChain handler (top-level roots)
3. use_case_key= on init(), or MAINWAVE_USE_CASE_KEY
4. (default) an agent_name bound at a run root → f"{service_name}/{agent_name}" — one use case per agent
5. (default) service_name — one use case per deployable

Treat the key like a database key: renaming it forks the use case. Keys are case-sensitive and trimmed; mainwave.use_case_id_for("my-key") (or agent_name="…") predicts the derived UUID for scripts and CI. Set ucid_namespace= / MAINWAVE_UCID_NAMESPACE to your org's issued namespace (shown in the app) so derived identities survive API-key rotation — without it, derivation seeds on the API key and rotating the key moves every derived identity (the SDK warns at init).

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

Each agent reported as its own use case is the default: bind the agent's name at its run root and the per-agent identity derives automatically —

with mainwave.run("INV-12345", agent_name="divergence"):
    ...  # lands in the "your-service/divergence" use case

# or on the LangChain handler:
divergence = mainwave.langchain.CallbackHandler(agent_name="divergence")
chain.invoke(payload, config={"callbacks": [divergence]})

To group several agents under ONE use case instead, bind the identity at the outermost root you own and leave the inner surfaces identity-free — they inherit:

with mainwave.run("INV-12345", use_case_key="invoice-workflow", agent_name="extractor"):
    ...  # every agent inside shares the invoice-workflow use case

An explicit use_case_id= / use_case_key= on a handler wins over an enclosing run's binding (a deliberately multi-use-case run). Omit identity args to inherit; an empty value is ignored rather than applied. (service_name= is still accepted as a deprecated alias for agent_name. If your process identity is an explicit use_case_id, agent names stay a display axis and never move telemetry.)

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/design/agent-sdk/ in the platform monorepo.