mainwave 0.1.0rc2

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]").

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
    service_name="divergence",  # which agent within it
)
chain.invoke(payload, config={"callbacks": [divergence]})

A handler's use_case_id / service_name apply only to the spans it emits, so several agents can share one use case (distinguished by service_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.

See also

Spec: docs/agent_sdk/ in the platform monorepo.