Tracing Module API

Complete API reference for mantisdk.tracing.

Initialization Functions

`init()`

Initialize the tracing system.

tracing.init(
    *,
    trace_name: str | None = None,
    service_name: str | None = None,
    environment: str | None = None,
    version: str | None = None,
    insight_host: str | None = None,
    insight_public_key: str | None = None,
    insight_secret_key: str | None = None,
    insight_endpoint: str | None = None,
    instrument_default: bool = True,
    exporters: list[SpanExporter] | None = None,
    trace_config: TraceConfig | None = None,
    force: bool = False,
    integrations: dict[str, Any] | None = None,
) -> None

Parameters:

Parameter	Type	Default	Description
`trace_name`	`str \| None`	`None`	Name for the trace (set as `trace.name` resource attribute)
`service_name`	`str \| None`	`"mantisdk-tracing"`	Service name in traces
`environment`	`str \| None`	`None`	Environment name (e.g., `"production"`, `"staging"`). Sets `insight.environment` resource attribute for filtering in Insight
`version`	`str \| None`	`None`	Application version (e.g., `"1.2.3"`, git SHA). Sets `insight.version` resource attribute for tracking regressions
`insight_host`	`str \| None`	Env var	Insight server URL. Overrides `INSIGHT_HOST` env var
`insight_public_key`	`str \| None`	Env var	Insight public key (`pk-lf-...`). Overrides `INSIGHT_PUBLIC_KEY` env var
`insight_secret_key`	`str \| None`	Env var	Insight secret key (`sk-lf-...`). Overrides `INSIGHT_SECRET_KEY` env var
`insight_endpoint`	`str \| None`	Derived from host	Custom OTLP endpoint URL. Overrides `INSIGHT_OTLP_ENDPOINT` env var
`instrument_default`	`bool`	`True`	Auto-instrument core libraries
`exporters`	`list[SpanExporter] \| None`	Auto-detect	Custom span exporters. If `None` and Insight credentials are available (via params or env vars), an Insight exporter is created automatically
`trace_config`	`TraceConfig \| None`	`None`	OpenInference trace config
`force`	`bool`	`False`	Replace existing TracerProvider. If `False`, reuses existing provider and appends processors
`integrations`	`dict[str, Any] \| None`	`None`	Fine-grained control over integrations

Example:

import mantisdk.tracing as tracing
 
# Minimal -- auto-detect Insight from env vars
tracing.init()
 
# With environment and version for filtering in Insight
tracing.init(
    service_name="my-agent",
    environment="production",
    version="1.2.3",
    instrument_default=True,
)
 
# Explicit Insight credentials (instead of env vars)
tracing.init(
    service_name="my-agent",
    environment="production",
    version="1.2.3",
    insight_host="https://insight.withmetis.ai",
    insight_public_key="pk-lf-...",
    insight_secret_key="sk-lf-...",
)

`instrument()`

Enable auto-instrumentation for specific libraries.

tracing.instrument(
    names: list[str] | None = None,
    skip: list[str] | None = None,
) -> None

Parameters:

Parameter	Type	Default	Description
`names`	`list[str] \| None`	Core set	Instrumentors to enable
`skip`	`list[str] \| None`	`None`	Instrumentors to skip

Core set (enabled by default when names=None):

openai_agents, claude_agent_sdk, openai, anthropic, langchain, llama_index, litellm

Additional available:

google_adk, mistral, groq, bedrock, vertexai, dspy, instructor, crewai

Example:

# Enable specific instrumentors
tracing.instrument(names=["anthropic", "dspy"])
 
# Enable all except langchain
tracing.instrument(skip=["langchain"])

`shutdown()`

Flush and shutdown the tracer provider.

tracing.shutdown(timeout_millis: int = 30000) -> None

Parameters:

Parameter	Type	Default	Description
`timeout_millis`	`int`	`30000`	Maximum wait time in ms

Example:

import atexit
import mantisdk.tracing as tracing
 
tracing.init()
atexit.register(tracing.shutdown)

`flush()`

Force flush pending spans without shutting down.

tracing.flush(timeout_millis: int = 30000) -> bool

Parameters:

Parameter	Type	Default	Description
`timeout_millis`	`int`	`30000`	Maximum wait time in ms

Returns: True if flush succeeded, False otherwise.

Example:

# Process batch
for item in batch:
    process(item)
 
# Ensure batch spans are exported
tracing.flush()

Span Creation

`trace`

A class that creates a root span. Works as both a decorator and a context manager.

As a decorator, it automatically captures function inputs and outputs. As a context manager, it allows explicit input/output control via span attributes.

class trace:
    def __init__(
        self,
        name_or_func: str | Callable | None = None,
        *,
        name: str | None = None,
        input: Any = None,
        kind: SpanKind = SpanKind.INTERNAL,
        attributes: dict[str, Any] | None = None,
        capture_input: bool = True,
        capture_output: bool = True,
        record_exception: bool = True,
        set_status_on_exception: bool = True,
    ): ...

Parameters:

Parameter	Type	Default	Description
`name_or_func`	`str \| Callable \| None`	`None`	Span name or function to decorate
`name`	`str \| None`	`None`	Explicit span name
`input`	`Any`	`None`	Input value (context manager mode)
`kind`	`SpanKind`	`INTERNAL`	OpenTelemetry span kind
`attributes`	`dict \| None`	`None`	Initial attributes
`capture_input`	`bool`	`True`	Auto-capture function input (decorator mode)
`capture_output`	`bool`	`True`	Auto-capture function output (decorator mode)
`record_exception`	`bool`	`True`	Record exceptions
`set_status_on_exception`	`bool`	`True`	Set ERROR status on exception

Example:

# Decorator without arguments
@tracing.trace
def my_function(query: str) -> dict:
    return {"result": query}
 
# Decorator with arguments
@tracing.trace(name="custom-name", capture_output=False)
def another_function(data):
    process(data)
 
# Async decorator
@tracing.trace
async def async_function(query: str) -> dict:
    return await fetch(query)
 
# Context manager
with tracing.trace("my-workflow", input=query) as span:
    result = do_work()
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

`span()`

Create a child span (context manager only).

@contextmanager
def span(
    name: str,
    *,
    input: Any = None,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
    **extra_attributes: Any,
) -> Generator[Span, None, None]

Parameters:

Parameter	Type	Default	Description
`name`	`str`	Required	Span name
`input`	`Any`	`None`	Input value
`kind`	`SpanKind`	`INTERNAL`	Span kind
`attributes`	`dict \| None`	`None`	Initial attributes
`record_exception`	`bool`	`True`	Record exceptions
`set_status_on_exception`	`bool`	`True`	Set ERROR status
`**extra_attributes`	`Any`	-	Additional attributes

Example:

with tracing.trace("workflow"):
    with tracing.span("step_1", input="data") as s:
        result = process()
        s.set_attribute("count", len(result))

`tool()`

Create a tool execution span (context manager only). Uses CLIENT span kind and sets a tool.name attribute.

@contextmanager
def tool(
    name: str,
    *,
    input: Any = None,
    tool_name: str | None = None,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
    **extra_attributes: Any,
) -> Generator[Span, None, None]

Parameters:

Parameter	Type	Default	Description
`name`	`str`	Required	Span name
`input`	`Any`	`None`	Tool input
`tool_name`	`str \| None`	`name`	Explicit tool.name attribute
`attributes`	`dict \| None`	`None`	Initial attributes
`record_exception`	`bool`	`True`	Record exceptions
`set_status_on_exception`	`bool`	`True`	Set ERROR status
`**extra_attributes`	`Any`	-	Additional attributes

Example:

with tracing.tool("search_database", input=query) as s:
    results = db.search(query)
    s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(results)))

`atrace()`

Async context manager for root spans.

@asynccontextmanager
async def atrace(
    name: str,
    *,
    input: Any = None,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
    **extra_attributes: Any,
) -> AsyncGenerator[Span, None]

Example:

async with tracing.atrace("async-workflow", input=query) as span:
    result = await async_operation()
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

`aspan()`

Async context manager for child spans.

@asynccontextmanager
async def aspan(
    name: str,
    *,
    input: Any = None,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
    **extra_attributes: Any,
) -> AsyncGenerator[Span, None]

Example:

async with tracing.atrace("workflow"):
    async with tracing.aspan("fetch_data") as s:
        data = await fetch()
        s.set_attribute("count", len(data))

Tracing Context

`tracing_context()`

Context manager to set tracing metadata for all spans created within the block. The TracingContextSpanProcessor reads these values and injects them as span attributes automatically.

@contextmanager
def tracing_context(
    *,
    call_type: str | None = None,
    tags: list[str] | None = None,
    environment: str | None = None,
    session_id: str | None = None,
) -> Generator[None, None, None]

Parameters:

Parameter	Type	Default	Description
`call_type`	`str \| None`	`None`	LLM call type (e.g., `"agent-call"`, `"judge-call"`, `"reflection-call"`)
`tags`	`list[str] \| None`	`None`	Tags for the traces
`environment`	`str \| None`	`None`	Environment name (e.g., `"mantisdk-gepa"`)
`session_id`	`str \| None`	`None`	Session ID for proxy spans

Example:

with tracing.tracing_context(call_type="agent-call", tags=["gepa", "train"]):
    response = client.chat.completions.create(...)  # Tagged automatically
 
with tracing.tracing_context(call_type="judge-call"):
    score = evaluate(response)  # Tagged as judge-call

`call_type_decorator()`

Factory that creates a decorator to set the call_type context for the duration of the decorated function. Works with both sync and async functions.

def call_type_decorator(call_type: str) -> Callable[[F], F]

Parameters:

Parameter	Type	Default	Description
`call_type`	`str`	Required	The call type to set (e.g., `"agent-call"`, `"judge-call"`)

Returns: A decorator that wraps functions with tracing context.

Example:

# Create reusable decorators
agent = tracing.call_type_decorator("agent-call")
judge = tracing.call_type_decorator("judge-call")
 
@agent
def my_agent_function():
    response = client.chat.completions.create(...)  # Tagged as "agent-call"
 
@judge
async def my_judge_function():
    response = await client.chat.completions.create(...)  # Tagged as "judge-call"

`get_current_call_type()`

Get the current LLM call type from context.

def get_current_call_type() -> str | None

Returns: The current call type string, or None if not set.

`get_current_tags()`

Get the current tracing tags from context.

def get_current_tags() -> list[str] | None

Returns: The current list of tags, or None if not set.

`get_current_environment()`

Get the current tracing environment from context.

def get_current_environment() -> str | None

Returns: The current environment string, or None if not set.

Utility Functions

`get_current_span()`

Get the currently active span.

def get_current_span() -> Span

Returns: The current span, or a non-recording span if none active.

Example:

def inner_function():
    span = tracing.get_current_span()
    span.set_attribute("inner_data", "value")
 
with tracing.trace("outer"):
    inner_function()

`insight_exporter()`

Create an Insight span exporter.

def insight_exporter(
    *,
    host: str | None = None,
    public_key: str | None = None,
    secret_key: str | None = None,
    endpoint: str | None = None,
) -> SpanExporter

Parameters:

Parameter	Type	Default	Description
`host`	`str \| None`	Env var	Insight host URL
`public_key`	`str \| None`	Env var	Public key
`secret_key`	`str \| None`	Env var	Secret key
`endpoint`	`str \| None`	Derived from host	Custom OTLP endpoint

Example:

tracing.init(
    exporters=[
        tracing.insight_exporter(
            host="https://insight.withmetis.ai",
            public_key="pk-lf-...",
            secret_key="sk-lf-...",
        )
    ]
)

Helper Functions

`is_insight_configured()`

Check if Insight credentials are configured.

from mantisdk.tracing.exporters.insight import is_insight_configured
 
def is_insight_configured() -> bool

Returns: True if all required env vars are set.

Example:

if is_insight_configured():
    exporters.append(tracing.insight_exporter())

Error Handling Functions

`record_exception()`

Record an exception on a span with proper attributes.

def record_exception(
    span: Span,
    exception: BaseException,
    *,
    attributes: dict[str, Any] | None = None,
    escaped: bool = False,
) -> None

Parameters:

Parameter	Type	Default	Description
`span`	`Span`	Required	The span to record on
`exception`	`BaseException`	Required	The exception to record
`attributes`	`dict \| None`	`None`	Additional attributes
`escaped`	`bool`	`False`	Whether exception escaped the span's scope

Example:

with tracing.span("operation") as s:
    try:
        risky_operation()
    except ValueError as e:
        tracing.record_exception(s, e, attributes={"input": "bad"})
        # Handle or re-raise

`set_span_error()`

Set a span's status to ERROR with a description.

def set_span_error(
    span: Span,
    message: str,
    *,
    exception: BaseException | None = None,
) -> None

Parameters:

Parameter	Type	Default	Description
`span`	`Span`	Required	The span to mark as error
`message`	`str`	Required	Error description
`exception`	`BaseException \| None`	`None`	Optional exception to record

Example:

with tracing.span("operation") as s:
    result = do_work()
    if not result.success:
        tracing.set_span_error(s, f"Operation failed: {result.error}")

`set_span_ok()`

Set a span's status to OK.

def set_span_ok(span: Span, message: str | None = None) -> None

Parameters:

Parameter	Type	Default	Description
`span`	`Span`	Required	The span to mark as OK
`message`	`str \| None`	`None`	Optional success message

Example:

with tracing.span("operation") as s:
    result = do_work()
    if result.success:
        tracing.set_span_ok(s, "Completed successfully")

Decorator Variants

`trace_decorator()`

Decorator variant for tracing function execution. This is a lower-level alternative to trace -- it creates spans but does not auto-capture inputs and outputs.

def trace_decorator(
    func: Callable | None = None,
    *,
    name: str | None = None,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
) -> Callable

Parameters:

Parameter	Type	Default	Description
`func`	`Callable \| None`	`None`	Function to decorate
`name`	`str \| None`	Function name	Custom span name
`kind`	`SpanKind`	`INTERNAL`	The span kind
`attributes`	`dict \| None`	`None`	Initial attributes
`record_exception`	`bool`	`True`	Record exceptions
`set_status_on_exception`	`bool`	`True`	Set ERROR status on exception

Example:

from mantisdk.tracing.api import trace_decorator
 
@trace_decorator
def process_data(items):
    return [process(item) for item in items]
 
@trace_decorator(name="custom-operation", attributes={"version": "1.0"})
async def async_operation():
    return await do_something()

`tool_decorator()`

Decorator to trace a tool/function execution with CLIENT span kind. Lower-level alternative to the tool() context manager.

def tool_decorator(
    func: Callable | None = None,
    *,
    name: str | None = None,
    tool_name: str | None = None,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
    set_status_on_exception: bool = True,
) -> Callable

Parameters:

Parameter	Type	Default	Description
`func`	`Callable \| None`	`None`	Function to decorate
`name`	`str \| None`	Function name	Custom span name
`tool_name`	`str \| None`	Span name	Tool name attribute
`attributes`	`dict \| None`	`None`	Initial attributes
`record_exception`	`bool`	`True`	Record exceptions
`set_status_on_exception`	`bool`	`True`	Set ERROR status on exception

Example:

from mantisdk.tracing.api import tool_decorator
 
@tool_decorator
def search_database(query: str) -> list:
    return db.execute(query)
 
@tool_decorator(tool_name="web_search")
async def search_web(query: str) -> list:
    return await web_client.search(query)

Types Reference

SpanKind

OpenTelemetry span kinds available:

Kind	Description
`SpanKind.INTERNAL`	Default, internal operations
`SpanKind.SERVER`	Server-side request handling
`SpanKind.CLIENT`	Client-side request (tool calls)
`SpanKind.PRODUCER`	Message producer
`SpanKind.CONSUMER`	Message consumer

Span

The OpenTelemetry Span object exposes:

Method	Description
`set_attribute(key, value)`	Set a single attribute
`set_attributes(attributes)`	Set multiple attributes
`add_event(name, attributes)`	Add a span event
`record_exception(exception)`	Record an exception
`set_status(status)`	Set span status
`end()`	End the span

Usage Patterns

Nested Spans

with tracing.trace("workflow", input=data) as root:
    with tracing.span("step_1") as s1:
        result1 = step_one()
        s1.set_attribute("result_count", len(result1))
 
    with tracing.span("step_2") as s2:
        result2 = step_two(result1)
        s2.set_attribute("result_count", len(result2))
 
    root.set_attribute(tracing.semconv.TRACE_OUTPUT, str(result2))

Async Workflows

async def async_workflow(query: str):
    async with tracing.atrace("async-workflow", input=query) as root:
        async with tracing.aspan("fetch") as s1:
            data = await fetch_data(query)
 
        async with tracing.aspan("process") as s2:
            result = await process_data(data)
 
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
        return result

Mixed Decorator and Context Manager

@tracing.trace
def main_workflow(input_data: dict):
    # Automatic input/output capture from decorator
 
    with tracing.span("validation") as s:
        validate(input_data)
 
    with tracing.tool("database_query", input=input_data["query"]) as t:
        results = db.query(input_data["query"])
        t.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(results)))
 
    return results

Tracing Context with Call Types

agent = tracing.call_type_decorator("agent-call")
judge = tracing.call_type_decorator("judge-call")
 
@tracing.trace
@agent
def run_agent(query: str):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
    )
 
@judge
def evaluate_response(response: str):
    with tracing.tracing_context(tags=["evaluation"]):
        return client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": f"Score this: {response}"}],
        )

Error Handling

with tracing.trace("operation") as span:
    try:
        result = risky_operation()
        tracing.set_span_ok(span)
    except ValidationError as e:
        tracing.record_exception(span, e)
        tracing.set_span_error(span, f"Validation failed: {e}")
        raise
    except Exception as e:
        # Automatically recorded if record_exception=True (default)
        raise

Insight CLI Semantic Conventions