# Mantis SDK - Complete Documentation

> This file contains the complete documentation for Mantis SDK.
> Generated automatically from the public documentation pages.

---

## mantisdk

*Source: /*

# mantisdk

**OpenTelemetry-based distributed tracing for AI agents.**

Built by [Metis](https://withmetis.ai), mantisdk provides a simple, zero-config way to add observability to your AI applications. Trace LLM calls, tool executions, and agent workflows with automatic instrumentation for popular frameworks.

## Why mantisdk?

- **Zero-Config Setup** — Auto-detect credentials from environment variables
- **Native Insight Export** — Built-in integration with Metis Insight platform
- **Auto-Instrumentation** — Support for DSPy, Anthropic, Google ADK, LangChain, and more
- **Flexible APIs** — Decorators (`@trace`) and context managers (`span()`, `tool()`)
- **Full Async Support** — Native async/await with `atrace` and `aspan`
- **Rich Metadata** — Capture inputs, outputs, user IDs, sessions, and custom attributes

## Quick Example

```python
import mantisdk.tracing as tracing

# Initialize - auto-detects Insight credentials from environment
tracing.init()

# Use as a decorator - auto-captures input/output
@tracing.trace
def process_query(query: str) -> dict:
    result = my_llm_call(query)
    return {"response": result}

# Use as a context manager - explicit control
with tracing.trace("workflow", input=user_query) as span:
    result = process_query(user_query)
    span.set_attribute(tracing.semconv.USER_ID, "user-123")

# Ensure spans are exported to Insight
tracing.shutdown()
```

## Supported Frameworks

mantisdk provides auto-instrumentation via [OpenInference](https://github.com/Arize-ai/openinference) for all major AI/LLM frameworks:

| Framework | Auto-Instrumentation | Status |
|-----------|---------------------|--------|
| OpenAI | ✅ Yes | Supported |
| Anthropic | ✅ Yes | [Guide](/integrations/anthropic) |
| Claude Code SDK | ✅ Yes | [Guide](/integrations/claude-code-sdk) |
| DSPy | ✅ Yes | [Guide](/integrations/dspy) |
| Google ADK | ✅ Yes | [Guide](/integrations/google-adk) |
| LangChain | ✅ Yes | Supported |
| LlamaIndex | ✅ Yes | Supported |
| LiteLLM | ✅ Yes | Supported |
| Groq | ✅ Yes | Supported |
| Mistral | ✅ Yes | Supported |
| Instructor | ✅ Yes | Supported |

## Get Started

## Connect to Insight

Set these environment variables to export traces to Insight:

```bash
export INSIGHT_HOST="https://insight.withmetis.ai"
export INSIGHT_PUBLIC_KEY="pk-lf-..."
export INSIGHT_SECRET_KEY="sk-lf-..."
```

Then view your traces, analyze LLM performance, and debug agent behavior in the [Insight dashboard](https://insight.withmetis.ai).

## Links

- [PyPI Package](https://pypi.org/project/mantisdk/) — Install via pip
- [Insight Platform](https://insight.withmetis.ai) — AI observability dashboard
- [Metis](https://withmetis.ai) — Company website

## Agent Downloads

- [`/mantis-docs-mdx.zip`](/mantis-docs-mdx.zip) — original public MDX files with directory structure preserved
- [`/llms.txt`](/llms.txt) — compact agent-readable documentation index
- [`/llms-full.txt`](/llms-full.txt) — complete public docs corpus in one text file

---

## Installation

*Source: /installation*

# Installation

## Requirements

- Python 3.10 or higher
- pip or uv package manager

## Basic Installation

Install the core mantisdk package:

```bash
pip install mantisdk
```

Or using uv:

```bash
uv pip install mantisdk
```

## Optional Dependencies

Mantis SDK provides optional dependency groups for specific integrations:

### Agent Frameworks

```bash
# All agent frameworks
pip install mantisdk[agents]

# Specific frameworks
pip install mantisdk[langchain]    # LangChain + LangGraph
pip install mantisdk[dspy]         # DSPy
pip install mantisdk[crewai]       # CrewAI
```

### Training & Optimization

```bash
# Automatic Prompt Optimization
pip install mantisdk[apo]

# VERL training (requires GPU)
pip install mantisdk[verl]

# TRL integration
pip install mantisdk[trl]
```

### LLM Proxy

```bash
# LLMProxy feature (note: has strict dependency pins for boto3, grpcio, uvicorn)
pip install 'mantisdk[proxy]'
```

> **Note**: The `[proxy]` extra includes `litellm[proxy]` which has strict dependency pins
> for `boto3`, `grpcio`, and `uvicorn` that may conflict with other packages in your
> environment. Only install this if you need the LLMProxy feature.

### Storage Backends

```bash
# MongoDB support
pip install mantisdk[mongo]
```

### RAG Support

```bash
# RAG with FAISS and sentence-transformers
pip install mantisdk[rag]
```

## Full Installation

For development or to install all optional dependencies:

```bash
pip install mantisdk[agents,apo,mongo,rag]
```

## Verify Installation

After installation, verify the SDK is working:

```python
import mantisdk.tracing as tracing

# Should print version info
print(f"Mantis SDK installed successfully")

# Initialize (will use no-op tracer if no credentials)
tracing.init()
print("Tracing initialized")
```

## Environment Setup

Create a `.env` file in your project root:

```bash
# Insight credentials (required for trace export)
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# LLM API keys (for your agent)
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
```

Load environment variables in your code:

```python
from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing
tracing.init()  # Auto-detects credentials from environment
```

## Next Steps

- [Quickstart Guide](/quickstart) - Get tracing working in 5 minutes
- [Tracing Overview](/tracing) - Learn the core concepts
- [Integrations](/integrations) - Set up framework-specific instrumentation

---

## Quickstart

*Source: /quickstart*

# Quickstart

Get Mantis SDK tracing working in 5 minutes.

## 1. Install the SDK

```bash
pip install mantisdk python-dotenv
```

## 2. Set Environment Variables

Create a `.env` file:

```bash
# Insight credentials
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# Your LLM API key (example: OpenAI)
OPENAI_API_KEY=sk-your-openai-key
```

## 3. Initialize Tracing

```python
from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing

# Auto-detect credentials from environment
tracing.init()
```

## 4. Add Tracing to Your Code

### Option A: Decorator (Recommended)

Use `@tracing.trace` to automatically capture function inputs and outputs:

```python
import mantisdk.tracing as tracing

@tracing.trace
def answer_question(question: str) -> str:
    # Your LLM logic here
    response = call_llm(question)
    return response

# Usage
result = answer_question("What is the capital of France?")
```

### Option B: Context Manager

Use `tracing.trace()` as a context manager for explicit control:

```python
import mantisdk.tracing as tracing

def process_request(query: str) -> dict:
    with tracing.trace("process_request", input=query) as span:
        # Add custom attributes
        span.set_attribute(tracing.semconv.USER_ID, "user-123")

        # Your logic
        result = analyze(query)

        # Set output
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, str(result))

        return result
```

## 5. Nested Spans

Create child spans for sub-operations:

```python
import mantisdk.tracing as tracing

def my_agent(query: str) -> str:
    with tracing.trace("agent_workflow", input=query) as root:
        # Step 1: Classify
        with tracing.span("classify_query") as s1:
            query_type = classify(query)
            s1.set_attribute("query_type", query_type)

        # Step 2: Process
        with tracing.span("process_query") as s2:
            result = process(query, query_type)

        # Step 3: Generate response
        with tracing.span("generate_response") as s3:
            response = generate(result)
            s3.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, response)

        root.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
        return response
```

## 6. Tool Calls

Trace tool executions with `tracing.tool()`:

```python
import mantisdk.tracing as tracing

def search_database(query: str) -> list:
    with tracing.tool("search_database", input=query) as span:
        results = db.search(query)
        span.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(results)))
        return results
```

## 7. Clean Shutdown

Always call `shutdown()` before your program exits:

```python
import atexit
import mantisdk.tracing as tracing

tracing.init()

# Register shutdown handler
atexit.register(tracing.shutdown)

# Or manually at the end
try:
    main()
finally:
    tracing.shutdown()
```

## Complete Example

```python
#!/usr/bin/env python3
"""Simple agent with Mantis SDK tracing."""

from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing

# Initialize tracing
tracing.init(service_name="my-agent")

@tracing.trace
def classify_query(query: str) -> str:
    """Classify the user query."""
    if "weather" in query.lower():
        return "weather"
    return "general"

@tracing.trace
def answer_query(query: str, query_type: str) -> str:
    """Generate an answer based on query type."""
    if query_type == "weather":
        return "I can help with weather! Which city?"
    return "I can help with that question."

def main():
    query = "What's the weather like?"

    with tracing.trace("agent_session", input=query) as span:
        span.set_attribute(tracing.semconv.USER_ID, "demo-user")
        span.set_attribute(tracing.semconv.SESSION_ID, "session-001")

        # Process query
        query_type = classify_query(query)
        response = answer_query(query, query_type)

        span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
        print(response)

if __name__ == "__main__":
    try:
        main()
    finally:
        tracing.shutdown()
```

## Ingest external agent outputs

If you already have externally produced trace/score results and want to write them directly to Insight, use the SDK ingestion helper: [Insight Ingestion (SDK)](/integrations/insight-ingestion).

## Next Steps

- [Tracing Concepts](/tracing) - Understand traces, spans, and attributes
- [Auto-Instrumentation](/tracing/auto-instrumentation) - Enable framework-specific tracing
- [Integrations](/integrations) - See full examples for DSPy, Anthropic, and Google ADK

---

## Examples

*Source: /examples*

# Examples

Complete working examples demonstrating Mantis SDK tracing with different frameworks.

## Available Examples

| Example | Framework | Features |
|---------|-----------|----------|
| [Reasoning Agent](/examples/reasoning-agent) | DSPy | Chain-of-thought, signatures, modules |
| [Task Planner](/examples/task-planner) | Anthropic | Tool use, async, multi-turn |
| [Research Assistant](/examples/research-assistant) | Google ADK | Multi-tool, sessions |

## Common Patterns

All examples follow these patterns:

### 1. Initialize Tracing First

```python
from dotenv import load_dotenv
load_dotenv()

# Initialize BEFORE importing framework
import mantisdk.tracing as tracing
tracing.init(service_name="my-agent", instrument_default=False)
tracing.instrument(names=["framework_name"])

# NOW import framework
import framework
```

### 2. Setup Function

```python
def setup_tracing(trace_to_console: bool = False):
    from mantisdk.tracing.exporters.insight import is_insight_configured

    exporters = []
    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="my-agent",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )
    tracing.instrument(names=["framework_name"])
```

### 3. Clean Shutdown

```python
if __name__ == "__main__":
    try:
        main()
    finally:
        tracing.shutdown()
```

## Running the Examples

### Prerequisites

1. Install Mantis SDK and the framework used by the example you want to run:

```bash
pip install mantisdk
```

2. Create a `.env` file or export the same variables in your shell:
```bash
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key

# Framework-specific keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
```

### Install Dependencies

Each example page lists the framework dependencies it needs. Install the matching framework package, then copy the example into a local Python file:

```bash
pip install dspy-ai
python reasoning_agent.py
```

For the Anthropic and Google ADK examples, install those framework packages instead:

```bash
pip install anthropic
pip install google-adk
```

## Copyable Examples

The full examples are included in this docs site:

- [DSPy Reasoning Agent](/examples/reasoning-agent)
- [Anthropic Task Planner](/examples/task-planner)
- [Google ADK Research Assistant](/examples/research-assistant)

---

## DSPy Reasoning Agent

*Source: /examples/reasoning-agent*

# DSPy Reasoning Agent

A chain-of-thought reasoning agent built with DSPy and Mantis SDK tracing.

## Overview

This example demonstrates:
- DSPy modules with ChainOfThought
- Custom signatures for different reasoning types
- `@tracing.trace` decorator for function tracing
- Context manager for explicit input/output control
- Auto-instrumentation of DSPy operations

## Features

- **Chain-of-thought reasoning** - Step-by-step reasoning with DSPy
- **Task classification** - Routes queries to appropriate reasoning modules
- **Math reasoning** - Specialized module for mathematical problems
- **General Q&A** - General-purpose question answering
- **Full tracing** - All operations traced to Insight

## Requirements

```bash
pip install mantisdk dspy-ai python-dotenv
```

## Environment Variables

```bash
# .env file
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key
OPENAI_API_KEY=sk-your-openai-key
```

## Full Source Code

```python
#!/usr/bin/env python3
"""DSPy Reasoning Agent with Mantis SDK Tracing."""

import argparse
import asyncio
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with DSPy instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found. "
              "Set INSIGHT_HOST, INSIGHT_PUBLIC_KEY, INSIGHT_SECRET_KEY")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())
        print("[Tracing] Console exporter enabled")

    tracing.init(
        service_name="dspy-reasoning-agent",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["dspy"])
    print("[Tracing] DSPy instrumentation enabled")

# NOW import DSPy (after tracing init)
import dspy

# Define DSPy signatures
class QuestionAnswer(dspy.Signature):
    """Answer a question with clear reasoning."""
    question: str = dspy.InputField(desc="The question to answer")
    reasoning: str = dspy.OutputField(desc="Step-by-step reasoning process")
    answer: str = dspy.OutputField(desc="The final answer")

class MathProblem(dspy.Signature):
    """Solve a math problem step by step."""
    problem: str = dspy.InputField(desc="The math problem to solve")
    steps: str = dspy.OutputField(desc="Step-by-step solution process")
    final_answer: str = dspy.OutputField(desc="The numerical answer")

class AnalyzeArgument(dspy.Signature):
    """Analyze an argument for logical validity."""
    argument: str = dspy.InputField(desc="The argument to analyze")
    analysis: str = dspy.OutputField(desc="Analysis of the argument's logic")
    conclusion: str = dspy.OutputField(desc="Whether the argument is valid/invalid")

class ReasoningAgent:
    """DSPy-based reasoning agent with chain-of-thought."""

    def __init__(self, model: str = "gpt-4o-mini"):
        # Configure DSPy
        self.lm = dspy.LM(f"openai/{model}", temperature=0.7)
        dspy.configure(lm=self.lm)

        # Create reasoning modules
        self.qa_module = dspy.ChainOfThought(QuestionAnswer)
        self.math_module = dspy.ChainOfThought(MathProblem)
        self.argument_module = dspy.ChainOfThought(AnalyzeArgument)

        self.conversation_history = []

    @tracing.trace(name="classify_task")
    def _classify_task(self, query: str) -> str:
        """Classify the type of reasoning task.

        Using @tracing.trace decorator - automatically captures input/output.
        """
        query_lower = query.lower()

        math_keywords = ["calculate", "compute", "solve", "what is",
                        "how much", "%", "+", "-", "*", "/", "="]
        argument_keywords = ["argue", "argument", "therefore", "because",
                            "valid", "invalid", "premise", "conclude"]

        if any(kw in query_lower for kw in math_keywords):
            if any(c.isdigit() for c in query):
                return "math"

        if any(kw in query_lower for kw in argument_keywords):
            return "argument"

        return "general"

    @tracing.trace(name="reason_general")
    def _reason_general(self, question: str) -> dict:
        """General Q&A reasoning - input/output auto-captured."""
        result = self.qa_module(question=question)
        return {
            "type": "general",
            "reasoning": result.reasoning,
            "answer": result.answer
        }

    @tracing.trace(name="reason_math")
    def _reason_math(self, problem: str) -> dict:
        """Mathematical reasoning - input/output auto-captured."""
        result = self.math_module(problem=problem)
        return {
            "type": "math",
            "steps": result.steps,
            "answer": result.final_answer
        }

    @tracing.trace(name="reason_argument")
    def _reason_argument(self, argument: str) -> dict:
        """Argument analysis - input/output auto-captured."""
        result = self.argument_module(argument=argument)
        return {
            "type": "argument",
            "analysis": result.analysis,
            "conclusion": result.conclusion
        }

    async def reason(self, query: str) -> str:
        """Process a query with appropriate reasoning strategy.

        Using trace context manager with explicit input/output capture.
        """
        with tracing.trace("reasoning_task", input=query) as span:
            # Classify the task
            task_type = self._classify_task(query)
            span.set_attribute("task_type", task_type)

            # Execute appropriate reasoning
            if task_type == "math":
                result = self._reason_math(query)
                response = f"**Solution Steps:**\n{result['steps']}\n\n**Answer:** {result['answer']}"
            elif task_type == "argument":
                result = self._reason_argument(query)
                response = f"**Analysis:**\n{result['analysis']}\n\n**Conclusion:** {result['conclusion']}"
            else:
                result = self._reason_general(query)
                response = f"**Reasoning:**\n{result['reasoning']}\n\n**Answer:** {result['answer']}"

            # Store in history
            self.conversation_history.append({
                "query": query,
                "type": task_type,
                "response": response
            })

            # Set output using semantic conventions
            span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
            return response

    def reset(self):
        """Reset conversation history."""
        self.conversation_history = []

async def run_interactive_session(agent: ReasoningAgent):
    """Run an interactive CLI session."""
    print("\n" + "="*60)
    print("  DSPy Reasoning Agent (with Mantis Tracing)")
    print("="*60)
    print("\nI can help you with:")
    print("  - General questions (chain-of-thought reasoning)")
    print("  - Math problems (step-by-step solutions)")
    print("  - Argument analysis (logical validity)")
    print("\nType 'quit' or 'exit' to end the session.")
    print("="*60 + "\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n\nGoodbye!")
            break

        if not user_input:
            continue

        if user_input.lower() in ("quit", "exit", "q"):
            print("\nGoodbye!")
            break

        if user_input.lower() == "clear":
            agent.reset()
            print("Conversation history cleared.\n")
            continue

        print("\nAssistant:\n")
        try:
            response = await agent.reason(user_input)
            print(response)
        except Exception as e:
            print(f"[Error] {e}")
        print()

async def run_single_query(agent: ReasoningAgent, query: str):
    """Run a single query."""
    try:
        response = await agent.reason(query)
        print(response)
    except Exception as e:
        print(f"[Error] {e}")

def main():
    parser = argparse.ArgumentParser(description="DSPy Reasoning Agent")
    parser.add_argument("--model", default="gpt-4o-mini")
    parser.add_argument("--query", "-q", help="Single query to run")
    parser.add_argument("--trace-to-console", action="store_true")
    parser.add_argument("--no-tracing", action="store_true")
    args = parser.parse_args()

    if not args.no_tracing:
        setup_tracing(trace_to_console=args.trace_to_console)

    agent = ReasoningAgent(model=args.model)

    try:
        if args.query:
            asyncio.run(run_single_query(agent, args.query))
        else:
            asyncio.run(run_interactive_session(agent))
    finally:
        if not args.no_tracing:
            tracing.shutdown()

if __name__ == "__main__":
    main()
```

## Usage

### Interactive Mode

```bash
python reasoning_agent.py
```

### Single Query

```bash
python reasoning_agent.py --query "What is 15% of 240?"
```

### With Console Tracing

```bash
python reasoning_agent.py --trace-to-console
```

## Trace Structure

```
reasoning_task
├── classify_task
│   ├── input: "What is 15% of 240?"
│   └── output: "math"
└── reason_math
    └── ChainOfThought.forward (auto-instrumented)
        └── LM.generate
            ├── model: gpt-4o-mini
            ├── input_tokens: 45
            └── output_tokens: 120
```

## Key Patterns Demonstrated

### 1. Decorator for Simple Functions

```python
@tracing.trace(name="classify_task")
def _classify_task(self, query: str) -> str:
    # Input/output automatically captured
    return task_type
```

### 2. Context Manager for Complex Workflows

```python
with tracing.trace("reasoning_task", input=query) as span:
    span.set_attribute("task_type", task_type)
    result = process(query)
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
```

### 3. Auto-Instrumentation

DSPy modules are automatically traced:
- `ChainOfThought.forward`
- `LM.generate`
- Token counts and latency

---

## Google ADK Research Assistant

*Source: /examples/research-assistant*

# Google ADK Research Assistant

A multi-tool research assistant built with Google ADK and Mantis SDK tracing.

## Overview

This example demonstrates:
- Google ADK agent with multiple tools
- Session management
- Auto-instrumentation of Gemini calls
- Interactive CLI interface

## Features

- **Multiple tools** - Time, weather, calculator, knowledge search
- **Session tracking** - User and session IDs
- **Streaming responses** - Async event handling
- **Full tracing** - Agent runs traced to Insight

## Requirements

```bash
pip install mantisdk google-adk openinference-instrumentation-google-adk python-dotenv
```

## Environment Variables

```bash
# .env file
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key
GOOGLE_API_KEY=your-google-api-key
```

## Full Source Code

```python
#!/usr/bin/env python3
"""Google ADK Research Assistant with Mantis SDK Tracing."""

import argparse
import asyncio
import datetime
from typing import Optional
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Google ADK instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="google-adk-research-assistant",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["google_adk"])
    print("[Tracing] Google ADK instrumentation enabled")

# NOW import Google ADK
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

# Tool implementations
def get_current_time(city: str) -> dict:
    """Get the current time in a specified city."""
    timezone_map = {
        "new york": "America/New_York",
        "los angeles": "America/Los_Angeles",
        "london": "Europe/London",
        "paris": "Europe/Paris",
        "tokyo": "Asia/Tokyo",
        "sydney": "Australia/Sydney",
    }

    city_lower = city.lower().strip()
    tz_id = timezone_map.get(city_lower)

    if not tz_id:
        return {
            "status": "error",
            "error_message": f"Unknown city: {city}. "
                           f"Available: {', '.join(timezone_map.keys())}"
        }

    tz = ZoneInfo(tz_id)
    now = datetime.datetime.now(tz)
    return {
        "status": "success",
        "report": f"Current time in {city.title()}: {now.strftime('%Y-%m-%d %H:%M:%S %Z')}"
    }

def get_weather(city: str) -> dict:
    """Get the current weather for a city (simulated)."""
    weather_data = {
        "new york": {"temp": 22, "condition": "Partly cloudy", "humidity": 65},
        "london": {"temp": 15, "condition": "Rainy", "humidity": 80},
        "tokyo": {"temp": 28, "condition": "Sunny", "humidity": 70},
        "paris": {"temp": 18, "condition": "Overcast", "humidity": 72},
    }

    city_lower = city.lower().strip()
    if city_lower not in weather_data:
        return {"status": "error", "error_message": f"No weather data for {city}"}

    data = weather_data[city_lower]
    return {
        "status": "success",
        "report": f"Weather in {city.title()}: {data['condition']}, "
                 f"{data['temp']}°C, Humidity: {data['humidity']}%"
    }

def calculate(expression: str) -> dict:
    """Evaluate a mathematical expression."""
    import math
    safe_dict = {
        "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
        "tan": math.tan, "log": math.log, "exp": math.exp,
        "pi": math.pi, "e": math.e, "abs": abs, "round": round,
    }
    try:
        result = eval(expression.replace("^", "**"),
                     {"__builtins__": {}}, safe_dict)
        return {"status": "success", "result": result}
    except Exception as e:
        return {"status": "error", "error_message": str(e)}

def search_knowledge(query: str, topic: Optional[str] = None) -> dict:
    """Search a knowledge base (simulated)."""
    knowledge = {
        "python": {
            "topic": "technology",
            "content": "Python is a high-level programming language "
                      "known for simplicity and readability."
        },
        "machine learning": {
            "topic": "technology",
            "content": "ML enables systems to learn from data "
                      "without explicit programming."
        },
        "quantum computing": {
            "topic": "technology",
            "content": "Quantum computing uses quantum mechanics "
                      "for computation."
        },
    }

    query_lower = query.lower()
    for key, value in knowledge.items():
        if query_lower in key:
            if topic is None or value["topic"] == topic.lower():
                return {"status": "success", "result": value["content"]}

    return {"status": "not_found", "message": f"No results for '{query}'"}

def create_research_assistant(model: str = "gemini-2.0-flash") -> Agent:
    """Create the research assistant agent."""
    return Agent(
        name="research_assistant",
        model=model,
        description="A research assistant with time, weather, "
                   "calculator, and knowledge tools.",
        instruction="""You are a helpful research assistant with tools:
1. get_current_time - Get time in various cities
2. get_weather - Get weather information
3. calculate - Evaluate math expressions
4. search_knowledge - Search a knowledge base

Use these tools to answer user questions. Be helpful and concise.""",
        tools=[get_current_time, get_weather, calculate, search_knowledge],
    )

async def run_interactive_session(agent: Agent):
    """Run an interactive CLI session with the agent."""
    import uuid

    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service

    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"

    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )

    print("\n" + "="*60)
    print("  Research Assistant (Google ADK + Mantis Tracing)")
    print("="*60)
    print("\nI can help you with:")
    print("  - Time in various cities")
    print("  - Weather information")
    print("  - Mathematical calculations")
    print("  - Knowledge base searches")
    print("\nType 'quit' to exit.")
    print("="*60 + "\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n\nGoodbye!")
            break

        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit", "q"):
            print("\nGoodbye!")
            break

        message = types.Content(
            role="user",
            parts=[types.Part(text=user_input)]
        )

        print("\nAssistant: ", end="", flush=True)

        try:
            # This run is auto-instrumented
            async for event in runner.run_async(
                user_id=user_id,
                session_id=session_id,
                new_message=message,
            ):
                if event.is_final_response() and event.content:
                    for part in event.content.parts:
                        if hasattr(part, "text") and part.text:
                            print(part.text)
        except Exception as e:
            print(f"\n[Error] {e}")
        print()

async def run_single_query(agent: Agent, query: str):
    """Run a single query against the agent."""
    import uuid

    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service

    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"

    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )

    message = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=message,
    ):
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(part.text)

def main():
    parser = argparse.ArgumentParser(description="Google ADK Research Assistant")
    parser.add_argument("--model", default="gemini-2.0-flash")
    parser.add_argument("--query", "-q", help="Single query")
    parser.add_argument("--trace-to-console", action="store_true")
    parser.add_argument("--no-tracing", action="store_true")
    args = parser.parse_args()

    if not args.no_tracing:
        setup_tracing(trace_to_console=args.trace_to_console)

    agent = create_research_assistant(model=args.model)

    try:
        if args.query:
            asyncio.run(run_single_query(agent, args.query))
        else:
            asyncio.run(run_interactive_session(agent))
    finally:
        if not args.no_tracing:
            tracing.shutdown()

if __name__ == "__main__":
    main()
```

## Usage

### Interactive Mode

```bash
python research_assistant_agent.py
```

### Single Query

```bash
python research_assistant_agent.py --query "What time is it in Tokyo?"
```

### Example Queries

```
You: What time is it in New York?
Assistant: The current time in New York is 2024-01-15 14:30:45 EST.

You: Calculate sqrt(144) + 10
Assistant: The result is 22.0.

You: Tell me about machine learning
Assistant: ML enables systems to learn from data without explicit programming.
```

## Trace Structure

```
run_async (auto-instrumented)
├── agent: research_assistant
├── user_id: user_abc123
├── session_id: session_xyz789
├── model: gemini-2.0-flash
└── tool_calls
    └── get_current_time
        ├── input: {"city": "Tokyo"}
        └── output: {"status": "success", "report": "..."}
```

## Key Patterns Demonstrated

### 1. Agent Definition

```python
agent = Agent(
    name="research_assistant",
    model="gemini-2.0-flash",
    tools=[get_current_time, get_weather, calculate, search_knowledge],
)
```

### 2. Session Management

```python
await session_service.create_session(
    app_name="research_app",
    user_id=user_id,
    session_id=session_id,
)
```

### 3. Async Event Handling

```python
async for event in runner.run_async(...):
    if event.is_final_response():
        print(event.content.parts[0].text)
```

### 4. Auto-Instrumentation

All `run_async()` calls are automatically traced with:
- Session context
- Tool executions
- Model interactions

---

## Anthropic Task Planner

*Source: /examples/task-planner*

# Anthropic Task Planner

A task management agent built with Anthropic Claude and Mantis SDK tracing.

## Overview

This example demonstrates:
- Anthropic Claude with tool use
- Multi-turn conversations
- Async patterns with tracing
- Auto-instrumentation of Anthropic API calls

## Features

- **Task management** - Create, list, complete tasks
- **Tool use** - Claude calls tools to manage tasks
- **Multi-turn** - Maintains conversation history
- **Full tracing** - API calls traced to Insight

## Requirements

```bash
pip install mantisdk anthropic python-dotenv
```

## Environment Variables

```bash
# .env file
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key
```

## Full Source Code

```python
#!/usr/bin/env python3
"""Anthropic Task Planner Agent with Mantis SDK Tracing."""

import argparse
import asyncio
import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Anthropic instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="anthropic-task-planner",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["anthropic"])
    print("[Tracing] Anthropic instrumentation enabled")

# NOW import Anthropic
import anthropic

# Task data model
@dataclass
class Task:
    id: str
    title: str
    description: str
    status: str = "pending"
    priority: str = "medium"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    completed_at: Optional[str] = None

class TaskStore:
    """Simple in-memory task store."""

    def __init__(self):
        self.tasks: Dict[str, Task] = {}

    def create_task(self, title: str, description: str,
                    priority: str = "medium") -> Task:
        import uuid
        task_id = str(uuid.uuid4())[:8]
        task = Task(id=task_id, title=title,
                    description=description, priority=priority)
        self.tasks[task_id] = task
        return task

    def get_task(self, task_id: str) -> Optional[Task]:
        return self.tasks.get(task_id)

    def list_tasks(self, status: Optional[str] = None) -> List[Task]:
        tasks = list(self.tasks.values())
        if status:
            tasks = [t for t in tasks if t.status == status]
        return tasks

    def complete_task(self, task_id: str) -> Optional[Task]:
        task = self.tasks.get(task_id)
        if task:
            task.status = "completed"
            task.completed_at = datetime.now().isoformat()
        return task

task_store = TaskStore()

# Tool implementations
def create_task(title: str, description: str, priority: str = "medium") -> dict:
    if priority not in ("low", "medium", "high"):
        return {"status": "error", "error_message": f"Invalid priority: {priority}"}
    task = task_store.create_task(title, description, priority)
    return {
        "status": "success",
        "task": {"id": task.id, "title": task.title, "priority": task.priority}
    }

def list_tasks(status: Optional[str] = None) -> dict:
    if status and status not in ("pending", "in_progress", "completed"):
        return {"status": "error", "error_message": f"Invalid status: {status}"}
    tasks = task_store.list_tasks(status)
    return {
        "status": "success",
        "tasks": [{"id": t.id, "title": t.title, "status": t.status} for t in tasks],
        "count": len(tasks)
    }

def get_task_details(task_id: str) -> dict:
    task = task_store.get_task(task_id)
    if not task:
        return {"status": "error", "error_message": f"Task '{task_id}' not found"}
    return {
        "status": "success",
        "task": {
            "id": task.id, "title": task.title, "description": task.description,
            "status": task.status, "priority": task.priority
        }
    }

def complete_task(task_id: str) -> dict:
    task = task_store.complete_task(task_id)
    if not task:
        return {"status": "error", "error_message": f"Task '{task_id}' not found"}
    return {
        "status": "success",
        "message": f"Task '{task.title}' marked as completed"
    }

TOOL_FUNCTIONS = {
    "create_task": create_task,
    "list_tasks": list_tasks,
    "get_task_details": get_task_details,
    "complete_task": complete_task,
}

TOOLS = [
    {
        "name": "create_task",
        "description": "Create a new task with title, description, and priority",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "The task title"},
                "description": {"type": "string", "description": "Task description"},
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "Task priority"
                }
            },
            "required": ["title", "description"]
        }
    },
    {
        "name": "list_tasks",
        "description": "List all tasks, optionally filtered by status",
        "input_schema": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["pending", "in_progress", "completed"]
                }
            }
        }
    },
    {
        "name": "get_task_details",
        "description": "Get detailed information about a specific task",
        "input_schema": {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"]
        }
    },
    {
        "name": "complete_task",
        "description": "Mark a task as completed",
        "input_schema": {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"]
        }
    }
]

class TaskPlannerAgent:
    """Anthropic-based task planning agent with tool use."""

    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.messages = []
        self.system_prompt = """You are a helpful task planning assistant.
Help users manage their tasks effectively using the available tools."""

    def reset(self):
        self.messages = []

    async def chat(self, user_message: str) -> str:
        """Process a user message - auto-instrumented by Anthropic instrumentor."""
        self.messages.append({"role": "user", "content": user_message})

        # This API call is auto-instrumented
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            messages=self.messages,
            tools=TOOLS,
        )

        if response.stop_reason == "tool_use":
            return await self._handle_tool_use(response)

        text = "".join(b.text for b in response.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": response.content})
        return text

    async def _handle_tool_use(self, response) -> str:
        """Handle tool use in the response."""
        self.messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                func = TOOL_FUNCTIONS.get(block.name)
                if func:
                    result = func(**block.input)
                else:
                    result = {"status": "error", "error_message": "Unknown tool"}

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result)
                })

        self.messages.append({"role": "user", "content": tool_results})

        # Follow-up API call - also auto-instrumented
        final = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            messages=self.messages,
            tools=TOOLS,
        )

        if final.stop_reason == "tool_use":
            return await self._handle_tool_use(final)

        text = "".join(b.text for b in final.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": final.content})
        return text

async def run_interactive_session(agent: TaskPlannerAgent):
    """Run an interactive CLI session."""
    print("\n" + "="*60)
    print("  Anthropic Task Planner (with Mantis Tracing)")
    print("="*60)
    print("\nI can help you manage your tasks:")
    print("  - Create new tasks with priorities")
    print("  - List and filter tasks by status")
    print("  - Mark tasks as complete")
    print("\nType 'quit' to exit.")
    print("="*60 + "\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n\nGoodbye!")
            break

        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit", "q"):
            print("\nGoodbye!")
            break

        print("\nAssistant: ", end="", flush=True)
        try:
            response = await agent.chat(user_input)
            print(response)
        except Exception as e:
            print(f"\n[Error] {e}")
        print()

def main():
    parser = argparse.ArgumentParser(description="Anthropic Task Planner")
    parser.add_argument("--model", default="claude-sonnet-4-20250514")
    parser.add_argument("--query", "-q", help="Single query")
    parser.add_argument("--trace-to-console", action="store_true")
    parser.add_argument("--no-tracing", action="store_true")
    args = parser.parse_args()

    if not args.no_tracing:
        setup_tracing(trace_to_console=args.trace_to_console)

    agent = TaskPlannerAgent(model=args.model)

    try:
        if args.query:
            asyncio.run(agent.chat(args.query).then(print))
        else:
            asyncio.run(run_interactive_session(agent))
    finally:
        if not args.no_tracing:
            tracing.shutdown()

if __name__ == "__main__":
    main()
```

## Usage

### Interactive Mode

```bash
python task_planner_agent.py
```

### Example Conversation

```
You: Create a task to review the Q4 report
Assistant: I've created a new task for you...

You: List all my tasks
Assistant: Here are your current tasks...

You: Mark that task as complete
Assistant: Done! The task has been marked as completed.
```

## Trace Structure

```
chat (function call)
├── anthropic.messages.create (auto-instrumented)
│   ├── model: claude-sonnet-4-20250514
│   ├── input_tokens: 180
│   ├── output_tokens: 45
│   └── stop_reason: tool_use
└── anthropic.messages.create (follow-up)
    ├── model: claude-sonnet-4-20250514
    ├── input_tokens: 250
    └── output_tokens: 80
```

## Key Patterns Demonstrated

### 1. Auto-Instrumentation

All `messages.create()` calls are automatically traced:

```python
# No manual tracing needed - auto-instrumented
response = self.client.messages.create(
    model=self.model,
    max_tokens=1024,
    messages=self.messages,
    tools=TOOLS,
)
```

### 2. Tool Use Handling

Tool calls and results are captured in the trace:

```python
if response.stop_reason == "tool_use":
    return await self._handle_tool_use(response)
```

### 3. Multi-Turn Conversations

Each turn creates a new span with token counts.

---

## Framework Integrations

*Source: /integrations*

# Framework Integrations

Mantis SDK provides auto-instrumentation for popular AI frameworks via [OpenInference](https://github.com/Arize-ai/openinference). This section covers setup and usage for each supported framework.

## Supported Frameworks

| Framework | Instrumentor | Status |
|-----------|-------------|--------|
| [Claude Code SDK](/integrations/claude-code-sdk) | `claude_code_sdk` | Verified |
| [DSPy](/integrations/dspy) | `dspy` | Verified |
| [Anthropic](/integrations/anthropic) | `anthropic` | Verified |
| [Google ADK](/integrations/google-adk) | `google_adk` | Verified |
| OpenAI | `openai` | Supported |
| OpenAI Agents SDK | `openai_agents` | Supported |
| LangChain | `langchain` | Supported |
| LlamaIndex | `llama_index` | Supported |
| LiteLLM | `litellm` | Supported |
| Groq | `groq` | Supported |
| Mistral | `mistral` | Supported |
| Instructor | `instructor` | Supported |

## Quick Setup

### 1. Install Mantis SDK

```bash
pip install mantisdk
```

### 2. Initialize and Instrument

```python
import mantisdk.tracing as tracing

# Initialize tracing
tracing.init(service_name="my-agent", instrument_default=False)

# Instrument your framework
tracing.instrument(names=["your_framework"])
```

### 3. Import Your Framework

```python
# Import AFTER instrumentation
import your_framework

# Use normally - calls are auto-traced
```

## Common Pattern

All integrations follow this pattern:

```python
#!/usr/bin/env python3
from dotenv import load_dotenv
load_dotenv()

# 1. Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

exporters = []
if is_insight_configured():
    exporters.append(tracing.insight_exporter())

tracing.init(
    service_name="framework-agent",
    exporters=exporters if exporters else None,
    instrument_default=False,
)

# 2. Instrument specific framework
tracing.instrument(names=["framework_name"])

# 3. NOW import the framework
import framework

# 4. Use the framework - auto-traced
def main():
    with tracing.trace("workflow") as span:
        # Framework calls are auto-instrumented
        result = framework.do_something()
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

# 5. Clean shutdown
if __name__ == "__main__":
    try:
        main()
    finally:
        tracing.shutdown()
```

## Combining Manual and Auto Tracing

Auto-instrumentation works alongside manual spans:

```python
@tracing.trace
def my_workflow(query: str):
    # Manual span
    with tracing.span("preprocessing"):
        prepared = preprocess(query)

    # Auto-instrumented LLM call (child of my_workflow)
    response = llm_client.generate(prepared)

    # Manual span
    with tracing.span("postprocessing"):
        result = postprocess(response)

    return result
```

## Environment Variables

All integrations use the same Insight credentials:

```bash
# Insight (required for trace export)
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# Framework-specific API keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
```

## Direct Insight Ingestion (SDK)

Need to ingest externally generated traces or scores (for example from batch evaluators, CSV processors, or custom agent workers)? Use the dedicated SDK ingestion helper:

- [Insight Ingestion (SDK)](/integrations/insight-ingestion)

This path complements auto-instrumentation. Use auto-instrumentation for runtime tracing, and the ingestion helper for explicit trace/score submission from external pipelines.

## Agent Tool Access (MCP)

Need Cursor, Claude Desktop, or another agent to inspect Insight data through tools? Use the Mantis MCP server:

- [Mantis MCP Server](/integrations/mantis-mcp-server)

The MCP server exposes read-only project, trace, score, and SQL analytics tools to MCP-compatible AI clients.

## Next Steps

- [Claude Code SDK Integration](/integrations/claude-code-sdk)
- [DSPy Integration](/integrations/dspy)
- [Anthropic Integration](/integrations/anthropic)
- [Google ADK Integration](/integrations/google-adk)

---

## Anthropic Integration

*Source: /integrations/anthropic*

# Anthropic Integration

Trace Anthropic Claude API calls with Mantis SDK, including messages, tool use, and token counts.

## Overview

Mantis SDK auto-instruments the Anthropic Python SDK to capture:

- Message creation calls
- Tool/function calls
- Token usage (input/output)
- Model and parameters

## Installation

```bash
pip install mantisdk anthropic
```

## Quick Start

```python
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing

tracing.init(service_name="anthropic-agent", instrument_default=False)
tracing.instrument(names=["anthropic"])

# NOW import Anthropic
import anthropic

# Create client and use - automatically traced
client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude!"}]
)
print(response.content[0].text)

# Clean up
tracing.shutdown()
```

## Tracing Features

### Auto-Captured Data

Every `messages.create()` call automatically captures:

| Attribute | Description |
|-----------|-------------|
| Model | Model name (e.g., claude-sonnet-4-20250514) |
| Input tokens | Token count for input |
| Output tokens | Token count for output |
| Messages | Input messages |
| Response | Generated response |
| Tool calls | Tool use blocks (if any) |
| Latency | Request duration |

### Tool Use Tracing

Tool calls are captured as part of the span:

```python
import anthropic
import mantisdk.tracing as tracing

client = anthropic.Anthropic()

tools = [
    {
        "name": "get_weather",
        "description": "Get weather for a city",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"}
            },
            "required": ["city"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}]
)
# Tool calls are automatically traced
```

## Complete Example: Task Planner Agent

This example demonstrates a task management agent with tool use:

```python
#!/usr/bin/env python3
"""Anthropic Task Planner Agent with Mantis SDK Tracing."""

import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Anthropic instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="anthropic-task-planner",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["anthropic"])
    print("[Tracing] Anthropic instrumentation enabled")

# NOW import Anthropic
import anthropic

# Task data model
@dataclass
class Task:
    id: str
    title: str
    description: str
    status: str = "pending"
    priority: str = "medium"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

# In-memory task store
class TaskStore:
    def __init__(self):
        self.tasks: Dict[str, Task] = {}

    def create_task(self, title: str, description: str, priority: str = "medium") -> Task:
        import uuid
        task_id = str(uuid.uuid4())[:8]
        task = Task(id=task_id, title=title, description=description, priority=priority)
        self.tasks[task_id] = task
        return task

    def list_tasks(self, status: Optional[str] = None) -> List[Task]:
        tasks = list(self.tasks.values())
        if status:
            tasks = [t for t in tasks if t.status == status]
        return tasks

    def complete_task(self, task_id: str) -> Optional[Task]:
        task = self.tasks.get(task_id)
        if task:
            task.status = "completed"
        return task

task_store = TaskStore()

# Tool implementations
def create_task(title: str, description: str, priority: str = "medium") -> dict:
    task = task_store.create_task(title, description, priority)
    return {"status": "success", "task": {"id": task.id, "title": task.title}}

def list_tasks(status: Optional[str] = None) -> dict:
    tasks = task_store.list_tasks(status)
    return {"status": "success", "tasks": [{"id": t.id, "title": t.title} for t in tasks]}

def complete_task(task_id: str) -> dict:
    task = task_store.complete_task(task_id)
    if task:
        return {"status": "success", "message": f"Completed: {task.title}"}
    return {"status": "error", "error_message": f"Task {task_id} not found"}

TOOL_FUNCTIONS = {
    "create_task": create_task,
    "list_tasks": list_tasks,
    "complete_task": complete_task,
}

TOOLS = [
    {
        "name": "create_task",
        "description": "Create a new task",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "description": {"type": "string"},
                "priority": {"type": "string", "enum": ["low", "medium", "high"]}
            },
            "required": ["title", "description"]
        }
    },
    {
        "name": "list_tasks",
        "description": "List all tasks",
        "input_schema": {
            "type": "object",
            "properties": {
                "status": {"type": "string", "enum": ["pending", "completed"]}
            }
        }
    },
    {
        "name": "complete_task",
        "description": "Mark a task as completed",
        "input_schema": {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"]
        }
    }
]

class TaskPlannerAgent:
    """Anthropic-based task planner with tool use."""

    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.messages = []
        self.system = "You are a task planning assistant. Help users manage tasks."

    async def chat(self, user_message: str) -> str:
        """Process a user message - auto-instrumented."""
        self.messages.append({"role": "user", "content": user_message})

        # This call is auto-instrumented
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system,
            messages=self.messages,
            tools=TOOLS,
        )

        if response.stop_reason == "tool_use":
            return await self._handle_tool_use(response)

        text = "".join(b.text for b in response.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": response.content})
        return text

    async def _handle_tool_use(self, response) -> str:
        """Handle tool calls."""
        self.messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                func = TOOL_FUNCTIONS.get(block.name)
                if func:
                    result = func(**block.input)
                else:
                    result = {"status": "error", "error_message": "Unknown tool"}

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result)
                })

        self.messages.append({"role": "user", "content": tool_results})

        # Follow-up call - also auto-instrumented
        final = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system,
            messages=self.messages,
            tools=TOOLS,
        )

        if final.stop_reason == "tool_use":
            return await self._handle_tool_use(final)

        text = "".join(b.text for b in final.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": final.content})
        return text

async def main():
    import asyncio

    setup_tracing(trace_to_console=False)

    agent = TaskPlannerAgent()

    # Example conversation
    queries = [
        "Create a task to review the Q4 report",
        "List all my tasks",
    ]

    for query in queries:
        print(f"\nUser: {query}")
        response = await agent.chat(query)
        print(f"Assistant: {response}")

    tracing.shutdown()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
```

## Trace Hierarchy

The example produces traces like:

```
chat (function call)
├── anthropic.messages.create (auto-instrumented)
│   ├── model: claude-sonnet-4-20250514
│   ├── input_tokens: 180
│   ├── output_tokens: 45
│   └── tool_use: create_task
└── anthropic.messages.create (follow-up)
    ├── model: claude-sonnet-4-20250514
    ├── input_tokens: 250
    └── output_tokens: 30
```

## Adding Manual Spans

Combine auto-instrumentation with manual tracing:

```python
import anthropic
import mantisdk.tracing as tracing

client = anthropic.Anthropic()

def analyze_text(text: str) -> dict:
    with tracing.trace("analyze_text", input=text) as span:
        span.set_attribute(tracing.semconv.USER_ID, "user-123")

        # Preprocess
        with tracing.span("preprocess"):
            cleaned = preprocess(text)

        # LLM call - auto-instrumented as child span
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": cleaned}]
        )

        result = response.content[0].text
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
        return {"analysis": result}
```

## Next Steps

- [Examples: Task Planner](/examples/task-planner) - Full example code
- [Google ADK Integration](/integrations/google-adk) - Google ADK
- [Auto-Instrumentation](/tracing/auto-instrumentation) - How it works

---

## Claude Code SDK Integration

*Source: /integrations/claude-code-sdk*

# Claude Code SDK Integration

The Claude Code SDK (also known as Claude Agent SDK) provides a powerful framework for building AI agents with Claude. mantisdk provides seamless integration for tracing all Claude agent interactions.

## Installation

## Quick Start

```python
#!/usr/bin/env python3
import asyncio
from typing import Any
from dotenv import load_dotenv
load_dotenv()

# Initialize mantisdk tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

exporters = []
if is_insight_configured():
    exporters.append(tracing.insight_exporter())

tracing.init(
    service_name="claude-agent",
    exporters=exporters if exporters else None,
    instrument_default=False,
)

# Instrument Claude Agent SDK
tracing.instrument(names=["claude_agent_sdk"])

# NOW import the SDK
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
)

# Define a tool
@tool(
    "get_weather",
    "Gets the current weather for a given city",
    {"city": str},
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    city = args["city"]
    # Simulated weather lookup
    weather = "Sunny, 72F"
    return {
        "content": [{"type": "text", "text": f"Weather in {city}: {weather}"}]
    }

async def main():
    # Create MCP server with tools
    weather_server = create_sdk_mcp_server(
        name="weather",
        version="1.0.0",
        tools=[get_weather],
    )

    # Configure agent options
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-20250514",
        system_prompt="You are a helpful assistant with weather information.",
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_weather"],
    )

    # Use the agent - calls are auto-traced
    with tracing.trace("weather-query", input="What's the weather in Paris?") as span:
        async with ClaudeSDKClient(options=options) as client:
            await client.query("What's the weather like in Paris?")

            response_text = []
            async for message in client.receive_response():
                if hasattr(message, "content"):
                    for block in message.content:
                        if hasattr(block, "text"):
                            response_text.append(block.text)

            result = "".join(response_text)
            span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
            print(result)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    finally:
        tracing.shutdown()
```

## What Gets Traced

When you instrument the Claude Agent SDK, mantisdk automatically captures:

| Event | Captured Data |
|-------|---------------|
| **Agent Queries** | Input prompts, model selection |
| **LLM Calls** | Tokens used, model response, latency |
| **Tool Executions** | Tool name, arguments, results |
| **MCP Server Calls** | Server name, tool invocations |
| **Errors** | Exception details, stack traces |

## Adding Custom Attributes

Enhance your traces with user context and metadata:

```python
async def handle_user_query(user_id: str, session_id: str, query: str):
    with tracing.trace("user-query", input=query) as span:
        # Add user context
        span.set_attribute(tracing.semconv.USER_ID, user_id)
        span.set_attribute(tracing.semconv.SESSION_ID, session_id)

        # Run the agent
        async with ClaudeSDKClient(options=options) as client:
            await client.query(query)

            async for message in client.receive_response():
                # Process response
                pass

        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

## Multi-Turn Conversations

Track entire conversation sessions:

```python
async def conversation_session(user_id: str):
    session_id = f"session-{uuid.uuid4()}"

    with tracing.trace("conversation", metadata={"session": session_id}) as session_span:
        session_span.set_attribute(tracing.semconv.USER_ID, user_id)
        session_span.set_attribute(tracing.semconv.SESSION_ID, session_id)

        async with ClaudeSDKClient(options=options) as client:
            # First turn
            with tracing.span("turn-1", input="Hello!"):
                await client.query("Hello!")
                # ... handle response

            # Second turn
            with tracing.span("turn-2", input="What can you help me with?"):
                await client.query("What can you help me with?")
                # ... handle response
```

## Error Handling

Capture and trace errors:

```python
async def safe_agent_call(query: str):
    with tracing.trace("agent-call", input=query) as span:
        try:
            async with ClaudeSDKClient(options=options) as client:
                await client.query(query)

                async for message in client.receive_response():
                    if hasattr(message, "is_error") and message.is_error:
                        span.set_attribute("error", True)
                        span.set_attribute("error.message", str(message))
                        raise Exception(f"Agent error: {message}")

                    # Process successful response

        except Exception as e:
            span.set_attribute("error", True)
            span.set_attribute("error.type", type(e).__name__)
            span.set_attribute("error.message", str(e))
            raise
```

## Best Practices

1. **Initialize Early**: Set up tracing before any SDK imports
2. **Use Sessions**: Group related queries with `SESSION_ID`
3. **Track Users**: Add `USER_ID` for user-level analytics
4. **Handle Errors**: Wrap agent calls in try/except with proper tracing
5. **Shutdown Cleanly**: Always call `tracing.shutdown()` to flush spans

## Viewing Traces in Insight

After running your Claude agent with mantisdk instrumentation:

1. Open [Insight](https://insight.withmetis.ai)
2. Navigate to your project
3. View the trace timeline showing:
   - Agent initialization
   - Each query/response cycle
   - Tool executions with inputs/outputs
   - Token usage and latency metrics

## Next Steps

- [Tracing API Reference](/reference/tracing) — Complete API documentation
- [Semantic Conventions](/tracing/semantic-conventions) — Standard attributes
- [Insight Export](/tracing/insight-export) — Configure trace export

---

## DSPy Integration

*Source: /integrations/dspy*

# DSPy Integration

Trace DSPy modules, signatures, and chain-of-thought reasoning with Mantis SDK.

## Overview

DSPy is a framework for programming—rather than prompting—language models. Mantis SDK automatically instruments:

- DSPy modules (ChainOfThought, ReAct, etc.)
- Signature executions
- LM calls

## Installation

```bash
pip install mantisdk dspy-ai
```

## Quick Start

```python
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing

tracing.init(service_name="dspy-agent", instrument_default=False)
tracing.instrument(names=["dspy"])

# NOW import DSPy
import dspy

# Configure DSPy
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

# Define signature
class QA(dspy.Signature):
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

# Use module - automatically traced
cot = dspy.ChainOfThought(QA)
result = cot(question="What is the capital of France?")
print(result.answer)

# Clean up
tracing.shutdown()
```

## Tracing Features

### Auto-Captured Data

When you use DSPy modules, the following is automatically traced:

- Module name and type
- Input fields
- Output fields
- Reasoning steps (for ChainOfThought)
- LM calls with token counts

### Manual Span Enrichment

Add custom context alongside auto-instrumentation:

```python
import mantisdk.tracing as tracing
import dspy

@tracing.trace
def classify_and_answer(query: str) -> dict:
    """Decorator auto-captures function I/O."""
    # Classify
    query_type = classify_query(query)

    # ChainOfThought call is auto-instrumented
    cot = dspy.ChainOfThought(QA)
    result = cot(question=query)

    return {
        "type": query_type,
        "answer": result.answer,
        "reasoning": result.reasoning,
    }
```

## Complete Example

This example demonstrates a reasoning agent with DSPy and full tracing:

```python
#!/usr/bin/env python3
"""DSPy Reasoning Agent with Mantis SDK Tracing."""

import argparse
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with DSPy instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="dspy-reasoning-agent",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["dspy"])
    print("[Tracing] DSPy instrumentation enabled")

# NOW import DSPy
import dspy

# Define signatures
class QuestionAnswer(dspy.Signature):
    """Answer a question with clear reasoning."""
    question: str = dspy.InputField(desc="The question to answer")
    reasoning: str = dspy.OutputField(desc="Step-by-step reasoning")
    answer: str = dspy.OutputField(desc="The final answer")

class MathProblem(dspy.Signature):
    """Solve a math problem step by step."""
    problem: str = dspy.InputField(desc="The math problem")
    steps: str = dspy.OutputField(desc="Solution steps")
    final_answer: str = dspy.OutputField(desc="The numerical answer")

# Agent class with manual tracing
class ReasoningAgent:
    """DSPy-based reasoning agent with tracing."""

    def __init__(self, model: str = "gpt-4o-mini"):
        self.lm = dspy.LM(f"openai/{model}", temperature=0.7)
        dspy.configure(lm=self.lm)

        self.qa_module = dspy.ChainOfThought(QuestionAnswer)
        self.math_module = dspy.ChainOfThought(MathProblem)

    @tracing.trace(name="classify_task")
    def _classify_task(self, query: str) -> str:
        """Classify the type of reasoning task."""
        query_lower = query.lower()
        math_keywords = ["calculate", "compute", "solve", "%", "+", "-"]

        if any(kw in query_lower for kw in math_keywords):
            if any(c.isdigit() for c in query):
                return "math"
        return "general"

    @tracing.trace(name="reason_general")
    def _reason_general(self, question: str) -> dict:
        """General Q&A reasoning - auto-traced."""
        result = self.qa_module(question=question)
        return {
            "type": "general",
            "reasoning": result.reasoning,
            "answer": result.answer,
        }

    @tracing.trace(name="reason_math")
    def _reason_math(self, problem: str) -> dict:
        """Math reasoning - auto-traced."""
        result = self.math_module(problem=problem)
        return {
            "type": "math",
            "steps": result.steps,
            "answer": result.final_answer,
        }

    def reason(self, query: str) -> str:
        """Process a query with appropriate reasoning."""
        with tracing.trace("reasoning_task", input=query) as span:
            task_type = self._classify_task(query)
            span.set_attribute("task_type", task_type)

            if task_type == "math":
                result = self._reason_math(query)
                response = f"Steps:\n{result['steps']}\n\nAnswer: {result['answer']}"
            else:
                result = self._reason_general(query)
                response = f"Reasoning:\n{result['reasoning']}\n\nAnswer: {result['answer']}"

            span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
            return response

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--query", "-q", default="What is 15% of 240?")
    parser.add_argument("--trace-to-console", action="store_true")
    args = parser.parse_args()

    setup_tracing(trace_to_console=args.trace_to_console)

    agent = ReasoningAgent()
    result = agent.reason(args.query)
    print(result)

    tracing.shutdown()

if __name__ == "__main__":
    main()
```

## Trace Hierarchy

Running the example produces this trace hierarchy:

```
reasoning_task (manual trace)
├── classify_task (manual span)
└── reason_math (manual span)
    └── ChainOfThought.forward (auto-instrumented)
        └── LM.generate (auto-instrumented)
            ├── model: gpt-4o-mini
            ├── input_tokens: 45
            └── output_tokens: 120
```

## Configuration Options

### Model Selection

```python
# OpenAI
lm = dspy.LM("openai/gpt-4o-mini")

# Anthropic
lm = dspy.LM("anthropic/claude-sonnet-4-20250514")

# Local model via LiteLLM
lm = dspy.LM("ollama/llama2")
```

### Temperature and Parameters

```python
lm = dspy.LM(
    "openai/gpt-4o-mini",
    temperature=0.7,
    max_tokens=1000,
)
```

## Next Steps

- [Examples: Reasoning Agent](/examples/reasoning-agent) - Full example code
- [Anthropic Integration](/integrations/anthropic) - Anthropic SDK
- [Auto-Instrumentation](/tracing/auto-instrumentation) - How it works

---

## Google ADK Integration

*Source: /integrations/google-adk*

# Google ADK Integration

Trace Google Agent Development Kit (ADK) agents with Mantis SDK.

## Overview

Google ADK is a framework for building agents with Gemini models. Mantis SDK auto-instruments:

- Agent executions
- Tool/function calls
- Session management
- Gemini API calls

## Installation

```bash
pip install mantisdk google-adk openinference-instrumentation-google-adk
```

## Quick Start

```python
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing

tracing.init(service_name="adk-agent", instrument_default=False)
tracing.instrument(names=["google_adk"])

# NOW import Google ADK
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

# Define a tool
def get_weather(city: str) -> dict:
    return {"status": "success", "weather": f"Sunny in {city}"}

# Create agent
agent = Agent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
)

# Run agent - automatically traced
runner = InMemoryRunner(agent=agent, app_name="weather_app")

# ... use runner.run_async() to interact

tracing.shutdown()
```

## Tracing Features

### Auto-Captured Data

Google ADK instrumentation captures:

| Attribute | Description |
|-----------|-------------|
| Agent name | Name of the agent |
| Model | Gemini model used |
| Session ID | Session identifier |
| User ID | User identifier |
| Tool calls | Function executions |
| Messages | Input/output messages |

### Session Tracking

Sessions are automatically tracked:

```python
import uuid
from google.adk.runners import InMemoryRunner

runner = InMemoryRunner(agent=agent, app_name="my_app")
session_service = runner.session_service

user_id = f"user_{uuid.uuid4().hex[:8]}"
session_id = f"session_{uuid.uuid4().hex[:8]}"

await session_service.create_session(
    app_name="my_app",
    user_id=user_id,
    session_id=session_id,
)

# Sessions are traced
```

## Complete Example: Research Assistant

This example demonstrates a multi-tool research assistant:

```python
#!/usr/bin/env python3
"""Google ADK Research Assistant with Mantis SDK Tracing."""

import datetime
from typing import Optional
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Google ADK instrumentation."""
    exporters = []

    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")

    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())

    tracing.init(
        service_name="google-adk-research-assistant",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )

    tracing.instrument(names=["google_adk"])
    print("[Tracing] Google ADK instrumentation enabled")

# NOW import Google ADK
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

# Define tools
def get_current_time(city: str) -> dict:
    """Get the current time in a city."""
    timezone_map = {
        "new york": "America/New_York",
        "london": "Europe/London",
        "tokyo": "Asia/Tokyo",
        "paris": "Europe/Paris",
    }

    tz_id = timezone_map.get(city.lower())
    if not tz_id:
        return {"status": "error", "error_message": f"Unknown city: {city}"}

    tz = ZoneInfo(tz_id)
    now = datetime.datetime.now(tz)
    return {"status": "success", "time": now.strftime("%Y-%m-%d %H:%M:%S %Z")}

def get_weather(city: str) -> dict:
    """Get weather for a city (simulated)."""
    weather_data = {
        "new york": {"temp": 22, "condition": "Partly cloudy"},
        "london": {"temp": 15, "condition": "Rainy"},
        "tokyo": {"temp": 28, "condition": "Sunny"},
        "paris": {"temp": 18, "condition": "Overcast"},
    }

    data = weather_data.get(city.lower())
    if not data:
        return {"status": "error", "error_message": f"No weather data for {city}"}

    return {
        "status": "success",
        "weather": f"{data['condition']}, {data['temp']}°C"
    }

def calculate(expression: str) -> dict:
    """Evaluate a math expression."""
    import math
    safe_dict = {
        "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
        "pi": math.pi, "e": math.e, "abs": abs, "round": round,
    }
    try:
        result = eval(expression.replace("^", "**"), {"__builtins__": {}}, safe_dict)
        return {"status": "success", "result": result}
    except Exception as e:
        return {"status": "error", "error_message": str(e)}

def search_knowledge(query: str, topic: Optional[str] = None) -> dict:
    """Search a knowledge base (simulated)."""
    knowledge = {
        "python": "Python is a high-level programming language.",
        "machine learning": "ML enables systems to learn from data.",
        "quantum computing": "Uses quantum mechanics for computation.",
    }

    for key, content in knowledge.items():
        if query.lower() in key:
            return {"status": "success", "result": content}

    return {"status": "not_found", "message": f"No results for '{query}'"}

# Create the agent
def create_research_assistant(model: str = "gemini-2.0-flash") -> Agent:
    """Create the research assistant agent."""
    return Agent(
        name="research_assistant",
        model=model,
        description="A research assistant with time, weather, calculator, and knowledge tools.",
        instruction="""You are a helpful research assistant with access to:
1. get_current_time - Get time in various cities
2. get_weather - Get weather information
3. calculate - Evaluate math expressions
4. search_knowledge - Search a knowledge base

Use these tools to help answer user questions.""",
        tools=[get_current_time, get_weather, calculate, search_knowledge],
    )

async def run_query(agent: Agent, query: str):
    """Run a single query against the agent."""
    import uuid

    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service

    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"

    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )

    message = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )

    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=message,
    ):
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    return part.text

    return None

async def main():
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("--query", "-q", default="What time is it in Tokyo?")
    parser.add_argument("--trace-to-console", action="store_true")
    args = parser.parse_args()

    setup_tracing(trace_to_console=args.trace_to_console)

    agent = create_research_assistant()

    print(f"Query: {args.query}")
    response = await run_query(agent, args.query)
    print(f"Response: {response}")

    tracing.shutdown()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
```

## Trace Hierarchy

The example produces traces like:

```
run_async (auto-instrumented)
├── session_id: session_abc123
├── user_id: user_xyz789
├── agent: research_assistant
└── tool_calls
    └── get_current_time
        ├── input: {"city": "Tokyo"}
        └── output: {"status": "success", "time": "..."}
```

## Adding Manual Context

Add manual spans for additional context:

```python
import mantisdk.tracing as tracing
from google.adk.runners import InMemoryRunner

async def enhanced_query(agent, query: str, user_id: str):
    with tracing.trace("research_session", input=query) as span:
        span.set_attribute(tracing.semconv.USER_ID, user_id)
        span.set_attribute(tracing.semconv.SESSION_ID, "session-123")

        runner = InMemoryRunner(agent=agent, app_name="research_app")

        # ADK calls are auto-instrumented as children
        response = await run_agent(runner, query)

        span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
        return response
```

## Interactive CLI Session

For interactive usage:

```python
async def run_interactive_session(agent: Agent):
    """Interactive CLI with session tracking."""
    import uuid

    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service

    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"

    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )

    print("Research Assistant ready. Type 'quit' to exit.")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break

        message = types.Content(
            role="user",
            parts=[types.Part(text=user_input)]
        )

        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=message,
        ):
            if event.is_final_response() and event.content:
                for part in event.content.parts:
                    if hasattr(part, "text") and part.text:
                        print(f"Assistant: {part.text}")
```

## Environment Variables

```bash
# Google API key (required)
GOOGLE_API_KEY=your-google-api-key

# Insight credentials
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-...
INSIGHT_SECRET_KEY=sk-lf-...
```

## Next Steps

- [Examples: Research Assistant](/examples/research-assistant) - Full example
- [DSPy Integration](/integrations/dspy) - DSPy framework
- [Auto-Instrumentation](/tracing/auto-instrumentation) - How it works

---

## Insight Ingestion (SDK)

*Source: /integrations/insight-ingestion*

# Insight Ingestion (SDK)

Send traces and scores directly to Insight from external agent pipelines using the SDK ingestion helpers.

## When to use this vs tracing export

- Use [Insight Export](/tracing/insight-export) when you want automatic OpenTelemetry span export from instrumented app runtime.
- Use this SDK ingestion helper when you need explicit event submission for traces and scores produced outside the runtime.

## Install and authenticate

## API surface

```python
from mantisdk import InsightIngestionClient, InsightIngestionConfig

client = InsightIngestionClient(
    InsightIngestionConfig(
        host="https://insight.withmetis.ai",
        public_key="pk-lf-...",
        secret_key="sk-lf-...",
        max_batch_size=200,
    )
)
```

### `from_env()`

Create a client from `INSIGHT_HOST`, `INSIGHT_PUBLIC_KEY`, and `INSIGHT_SECRET_KEY`.

### `ingest_trace_event(...)`

Send a single `trace-create` event through `/api/public/ingestion`.

### `send_score_api(score)`

Send a single score directly to `/api/public/scores`.

### `send_scores(scores, mode=...)`

Send many scores with two modes:

- `ingestion-batch` (default): score-create events batched to `/api/public/ingestion`
- `scores-api`: one score per request to `/api/public/scores`

## Score target rules (strict)

Insight scores must target exactly one object type:

- `traceId`
- `sessionId`
- `datasetRunId`

Do not send more than one target in the same score payload.

## Production guidance

### Prefer batching for throughput

```python
client.send_scores(
    [
        {"name": "task_success", "value": 0.94, "dataType": "NUMERIC", "traceId": "trace-1"},
        {"name": "safety", "value": 1, "dataType": "BOOLEAN", "traceId": "trace-1"},
    ],
    mode="ingestion-batch",
)
```

### Use direct API mode when per-score response semantics matter

```python
client.send_scores(
    [{"name": "task_success", "value": 0.94, "dataType": "NUMERIC", "traceId": "trace-1"}],
    mode="scores-api",
)
```

### Common failure modes

- `401/403`: invalid keys or wrong project scope
- `400`: invalid score payload (often target ID rule violation)
- `429`: rate limits; retry with backoff
- `5xx`: transient service error; retry safely

## End-to-end example

```python
from mantisdk import InsightIngestionClient, InsightIngestionConfig

client = InsightIngestionClient(
    InsightIngestionConfig(
        host="https://insight.withmetis.ai",
        public_key="pk-lf-...",
        secret_key="sk-lf-...",
    )
)

client.ingest_trace_event(
    trace_id="trace-sdk-001",
    name="External Agent Trace",
    input_data={"query": "help me with refunds"},
    output_data={"answer": "refund initiated"},
    session_id="session-001",
    environment="production",
)

client.send_scores(
    [
        {"name": "task_success", "value": 0.97, "dataType": "NUMERIC", "traceId": "trace-sdk-001"},
        {"name": "policy_compliant", "value": 1, "dataType": "BOOLEAN", "traceId": "trace-sdk-001"},
    ],
    mode="ingestion-batch",
)
```

## Next steps

- [Insight REST API](/reference/insight-rest-api)
- [Insight Export](/tracing/insight-export)
- [Framework Integrations](/integrations)

---

## Mantis MCP Server

*Source: /integrations/mantis-mcp-server*

# Mantis MCP Server

The Mantis MCP server gives AI tools read-only access to Insight traces, observations, scores, and analytics through the Model Context Protocol. Use it with Cursor, Claude Desktop, or any MCP-compatible client that can launch a local stdio server.

## Prerequisites

- An active Insight account.
- Node.js and npm.
- Cursor, Claude Desktop, or another MCP-compatible client.
- A personal access token from Insight, or a project API key pair.

## Create a personal access token

Personal access tokens are the recommended auth mode for MCP clients because they can list every project your user can read.

1. Open the Insight web app.
2. Click your account menu in the lower-left sidebar.
3. Open **Account Settings**.
4. Select **Access Tokens**.
5. Click **Create Token**.
6. Give the token a recognizable name, then copy it immediately.

## Install

### Global command

```bash
npm install -g @mantisai/mcp
mantis-mcp --help
```

### npx

```bash
npx -y @mantisai/mcp --help
```

`MANTIS_URL` defaults to `https://insight.withmetis.ai`, so production setup usually only needs a token.

## Configure Cursor

Open Cursor settings, go to **Tools & MCP**, and add a new MCP server named `mantis`.

Use one of these configurations.

### npx config

```json
{
  "mcpServers": {
    "mantis": {
      "command": "npx",
      "args": ["-y", "@mantisai/mcp", "serve"],
      "env": {
        "MANTIS_API_TOKEN": "pat_..."
      }
    }
  }
}
```

### Global command config

```json
{
  "mcpServers": {
    "mantis": {
      "command": "mantis-mcp",
      "args": ["serve"],
      "env": {
        "MANTIS_API_TOKEN": "pat_..."
      }
    }
  }
}
```

For a local or self-hosted Insight instance, set `MANTIS_URL` too:

```json
{
  "mcpServers": {
    "mantis": {
      "command": "npx",
      "args": ["-y", "@mantisai/mcp", "serve"],
      "env": {
        "MANTIS_URL": "http://localhost:3000",
        "MANTIS_API_TOKEN": "pat_..."
      }
    }
  }
}
```

After saving the config, reload MCP servers or restart Cursor. The `mantis` server should show as connected and expose Insight tools in chat.

## Configure Claude Desktop

Add the same server block to `claude_desktop_config.json`:

```json
{
  "mcpServers": {
    "mantis": {
      "command": "npx",
      "args": ["-y", "@mantisai/mcp", "serve"],
      "env": {
        "MANTIS_API_TOKEN": "pat_..."
      }
    }
  }
}
```

Restart Claude Desktop after editing the config.

## Authentication modes

| Mode | Environment variables | Scope | Use when |
|------|-----------------------|-------|----------|
| Personal access token | `MANTIS_API_TOKEN` | All projects the user can read | Cursor, Claude Desktop, agent workflows |
| Project API key pair | `MANTIS_PUBLIC_KEY` and `MANTIS_SECRET_KEY` | One project | Single-project automation |

PAT mode requires the AI client to pass `project_id` when calling most project-specific tools. Use `list_projects` first to find the right ID.

Project key mode is already scoped by the key pair, so project-specific tools do not need an explicit `project_id`.

## Available tools

Tools are discovered dynamically from the Insight backend. Common tools include:

| Tool | Purpose |
|------|---------|
| `get_context` | Explain the database schema, query patterns, available tools, and recommended workflow. |
| `list_projects` | List projects accessible to a PAT-authenticated user. |
| `describe_project` | Summarize trace counts, score configs, models, tags, and time range. |
| `get_trace` | Fetch traces with observations and scores. |
| `get_session` | Fetch session traces with score and observation summaries. |
| `get_observations` | Fetch observations by ID with full input and output content. |
| `get_score_configs` | Fetch score config metadata. |
| `diff_traces` | Compare two traces side by side. |
| `trace_stats` | Roll up latency, cost, and score stats by trace name. |
| `execute_query` | Run read-only ClickHouse SQL against allowed analytics tables. |
| `search_insights` | Search score reasoning and comments for relevant context. |
| `guide` | Ask for tool, schema, and query guidance. |

## Direct HTTP MCP

MCP-compatible agents can connect directly to the hosted endpoint without running the local stdio proxy.

```bash
curl -X POST "https://insight.withmetis.ai/api/public/mcp?project_id=$MANTIS_PROJECT_ID" \
  -H "Authorization: Bearer $MANTIS_API_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list"
  }'
```

Use direct HTTP MCP when your agent framework supports HTTP MCP transport. Use the stdio server for tools like Cursor and Claude Desktop that launch a local command.

## Verify

### Cursor or Claude Desktop

1. Reload or restart the client.
2. Confirm the `mantis` server is connected.
3. Ask the client to list available Mantis projects.
4. Pick a project ID and ask for a project summary or recent traces.

Example prompt:

```text
Use the Mantis tools to list my Insight projects, then describe the project with the most recent trace data.
```

### Terminal

Run the server with environment variables to confirm it can start and discover projects:

```bash
MANTIS_API_TOKEN="pat_..." npx -y @mantisai/mcp serve
```

The process stays open because stdio MCP servers wait for a client connection. Startup logs should show accessible projects and then that the MCP server is running on stdio.

## Security model

MCP tool calls are read-only. Insight enforces:

- Authentication on every request.
- Project scoping for every tool call.
- ClickHouse `readonly=1` for SQL execution.
- A table allowlist for analytics queries.
- Execution timeouts and result limits.

The MCP package should still be treated as a credentialed local process. Keep tokens out of source control, do not paste PATs into chats, and remove old tokens when a machine or config is no longer trusted.

## Troubleshooting

### `mantis-mcp: command not found`

Use the `npx` config, or reinstall globally:

```bash
npm install -g @mantisai/mcp
```

If global install still fails, make sure your npm global bin directory is on `PATH`.

### Server connects, but no projects appear

The PAT may not have access to any Insight projects. Confirm you can open the project in the Insight web app, then create a new token from that same account.

### `401` or auth errors

The token or key pair is missing, expired, revoked, or copied incorrectly. Delete the old token and create a fresh one from **Account Settings -> Access Tokens**.

### Cursor shows the server as disabled

Toggle the server on from **Settings -> Tools & MCP**. If it turns off again, inspect the MCP server logs and verify the configured command works in a normal terminal.

### Tool calls ask for `project_id`

That is expected in PAT mode. Run `list_projects`, copy the project ID, and include it in the tool call.

## Next steps

- [Insight CLI](/reference/insight-cli)
- [Insight SQL API](/reference/insight-sql-api)
- [Insight REST API](/reference/insight-rest-api)

---

## API Reference

*Source: /reference*

# API Reference

Complete API reference for Mantis SDK tracing module.

## Module: `mantisdk.tracing`

The main tracing module providing all public APIs.

```python
import mantisdk.tracing as tracing
```

## Quick Reference

### Insight Platform

| Page | Description |
|------|-------------|
| [Insight REST API](/reference/insight-rest-api) | Public API for traces, observations, scores, dashboards, datasets, prompts, and project resources |
| [Insight SQL API](/reference/insight-sql-api) | Read-only ClickHouse analytics over traces, observations, and scores |
| [Insight CLI](/reference/insight-cli) | Terminal workflow for API smoke checks, SQL diagnostics, and project operations |
| [OTel Ingestion](/reference/insight-otel-ingestion) | OpenTelemetry trace ingestion endpoint reference |

### Initialization

| Function | Description |
|----------|-------------|
| [`init()`](/reference/tracing#init) | Initialize tracing |
| [`instrument()`](/reference/tracing#instrument) | Enable auto-instrumentation |
| [`shutdown()`](/reference/tracing#shutdown) | Graceful shutdown |
| [`flush()`](/reference/tracing#flush) | Force flush spans |

### Span Creation

| Function | Description |
|----------|-------------|
| [`trace`](/reference/tracing#trace) | Create root spans (decorator/context manager) |
| [`span()`](/reference/tracing#span) | Create child spans (context manager) |
| [`tool()`](/reference/tracing#tool) | Create tool spans (context manager) |
| [`atrace()`](/reference/tracing#atrace) | Async root spans |
| [`aspan()`](/reference/tracing#aspan) | Async child spans |

### Utilities

| Function | Description |
|----------|-------------|
| [`get_current_span()`](/reference/tracing#get_current_span) | Get active span |
| [`insight_exporter()`](/reference/tracing#insight_exporter) | Create Insight exporter |

### Semantic Conventions

| Module | Description |
|--------|-------------|
| [`semconv`](/reference/semconv) | Standard attribute names |

## Import Patterns

```python
# Full module import (recommended)
import mantisdk.tracing as tracing
tracing.init()
tracing.trace("name")

# Specific imports
from mantisdk.tracing import init, trace, span, semconv

# Semantic conventions
from mantisdk.tracing.semconv import USER_ID, SESSION_ID, ObservationType

# Exporter utilities
from mantisdk.tracing.exporters.insight import is_insight_configured
```

## Type Annotations

```python
from opentelemetry.trace import Span, SpanKind
from opentelemetry.sdk.trace.export import SpanExporter

# Function signatures
def init(
    *,
    service_name: str | None = None,
    instrument_default: bool = True,
    exporters: list[SpanExporter] | None = None,
    force: bool = False,
) -> None: ...

def trace(
    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,
) -> Span | Callable: ...
```

## Next Steps

- [Tracing Module API](/reference/tracing) - Full function reference
- [Semantic Conventions](/reference/semconv) - Attribute constants
- [Environment Variables](/reference/environment) - Configuration reference

---

## Environment Variables

*Source: /reference/environment*

# Environment Variables

Reference for all environment variables used by Mantis SDK.

## Insight Configuration

Required for exporting traces to Mantis Insight.

### `INSIGHT_HOST`

The Insight server URL.

| Property | Value |
|----------|-------|
| Required | Yes (for Insight export) |
| Example | `https://insight.withmetis.ai` |

```bash
export INSIGHT_HOST=https://insight.withmetis.ai
```

### `INSIGHT_PUBLIC_KEY`

The public key for authentication.

| Property | Value |
|----------|-------|
| Required | Yes (for Insight export) |
| Format | `pk-lf-...` |

```bash
export INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
```

### `INSIGHT_SECRET_KEY`

The secret key for authentication.

| Property | Value |
|----------|-------|
| Required | Yes (for Insight export) |
| Format | `sk-lf-...` |

```bash
export INSIGHT_SECRET_KEY=sk-lf-your-secret-key
```

### `INSIGHT_OTLP_ENDPOINT`

Override the OTLP endpoint (optional).

| Property | Value |
|----------|-------|
| Required | No |
| Default | Derived from `INSIGHT_HOST` |

```bash
export INSIGHT_OTLP_ENDPOINT=https://custom.endpoint/v1/traces
```

---

## Trace Metadata via init()

In addition to environment variables, you can set trace-level metadata programmatically via `tracing.init()`. This is useful for:
- Filtering traces by environment (production, staging, development)
- Tracking app versions for regression debugging

```python
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    environment="production",  # Filter traces by environment in Insight
    version="1.2.3",           # Track app version for regression debugging
)
```

These values are set as OpenTelemetry resource attributes and apply to **all traces** from this process:
- `insight.environment` - Maps to the Environment filter in Insight
- `insight.version` - Maps to the Version column in Insight

## Insight Credentials via init()

You can also pass Insight credentials directly to `init()` instead of using environment variables:

```python
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    environment="production",
    version="1.2.3",
    # Pass credentials directly instead of env vars
    insight_host="https://insight.withmetis.ai",
    insight_public_key="pk-lf-...",
    insight_secret_key="sk-lf-...",
)
```

This is useful when:
- You want to configure credentials programmatically
- You're running in an environment where setting env vars is inconvenient
- You need to switch between different Insight projects dynamically

**Priority:** Parameters passed to `init()` take precedence over environment variables

### Langfuse-compatible aliasing (optional)

The SDK ingestion helper can mirror credentials into both env namespaces:

- `INSIGHT_HOST`, `INSIGHT_PUBLIC_KEY`, `INSIGHT_SECRET_KEY`
- `LANGFUSE_HOST`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`

Use this when downstream tooling expects `LANGFUSE_*` variables but you still want to target Insight credentials consistently.

---

## LLM API Keys

Required for the LLM providers you use.

### OpenAI

```bash
export OPENAI_API_KEY=sk-your-openai-key
```

### Anthropic

```bash
export ANTHROPIC_API_KEY=sk-ant-your-anthropic-key
```

### Google (Gemini)

```bash
export GOOGLE_API_KEY=your-google-api-key
```

### Groq

```bash
export GROQ_API_KEY=your-groq-key
```

### Mistral

```bash
export MISTRAL_API_KEY=your-mistral-key
```

---

## Complete `.env` Example

```bash
# Insight configuration
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# LLM API keys (set the ones you need)
OPENAI_API_KEY=sk-your-openai-key
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key
GOOGLE_API_KEY=your-google-api-key
```

---

## Loading Environment Variables

### Using python-dotenv

```python
from dotenv import load_dotenv
load_dotenv()  # Loads from .env file

import mantisdk.tracing as tracing
tracing.init()  # Auto-detects credentials
```

### Direct Shell Export

```bash
export INSIGHT_HOST=https://insight.withmetis.ai
export INSIGHT_PUBLIC_KEY=pk-lf-...
export INSIGHT_SECRET_KEY=sk-lf-...
python my_agent.py
```

### In Code

```python
import os
os.environ["INSIGHT_HOST"] = "https://insight.withmetis.ai"
os.environ["INSIGHT_PUBLIC_KEY"] = "pk-lf-..."
os.environ["INSIGHT_SECRET_KEY"] = "sk-lf-..."

import mantisdk.tracing as tracing
tracing.init()
```

---

## Checking Configuration

```python
from mantisdk.tracing.exporters.insight import is_insight_configured
import os

# Check if all required vars are set
print(f"Insight configured: {is_insight_configured()}")

# Debug individual variables
print(f"INSIGHT_HOST: {os.environ.get('INSIGHT_HOST', 'NOT SET')}")
print(f"INSIGHT_PUBLIC_KEY: {os.environ.get('INSIGHT_PUBLIC_KEY', 'NOT SET')[:15]}...")
print(f"INSIGHT_SECRET_KEY: {'SET' if os.environ.get('INSIGHT_SECRET_KEY') else 'NOT SET'}")
```

---

## Security Best Practices

1. **Never commit `.env` files** - Add `.env` to `.gitignore`
2. **Use secrets management** - In production, use proper secrets management
3. **Rotate keys regularly** - Follow your organization's key rotation policy
4. **Limit permissions** - Use keys with minimal required permissions

```gitignore
# .gitignore
.env
.env.local
*.env
```

---

## Insight CLI

*Source: /reference/insight-cli*

# Insight CLI

Use the Insight CLI to operate an Insight project from the terminal. It talks only to the public REST API, so requests use the same authentication, project scoping, and rate limits as browser and SDK traffic.

## Install

The CLI command is `insight`. Install it from npm:

```bash
npm install -g @mantisai/insight-cli
```

```bash
insight --help
```

## Authenticate

Use project API keys from Insight. The CLI accepts credentials from either environment variables or a saved profile.

### Environment variables

```bash
export INSIGHT_HOST="https://insight.withmetis.ai"
export INSIGHT_PUBLIC_KEY="pk-lf-..."
export INSIGHT_SECRET_KEY="sk-lf-..."

insight auth whoami
```

### Saved profile

```bash
insight auth login \
  --host "https://insight.withmetis.ai" \
  --public-key "pk-lf-..." \
  --secret-key "sk-lf-..."

insight auth whoami
```

Saved profiles live in `~/.insight/config.json`. Environment variables override saved profiles, which is usually the cleanest path for CI and smoke tests.

## Read-only smoke test

Run these commands when you want to confirm the CLI can authenticate, reach the hosted Insight API, and query trace data without creating or deleting anything.

```bash
insight health --json
insight ready --json
insight traces list --limit 3 --fields core
insight projects list
insight dashboards list --limit 5
insight sessions list --limit 3
insight observations list --limit 3
insight scores list --limit 3
insight trace-viewer-configs list
```

Check ClickHouse-backed SQL access with a small date range query:

```bash
insight sql \
  "SELECT count() AS total, min(timestamp) AS earliest, max(timestamp) AS latest FROM traces" \
  --json
```

If the project has stale data, a "last 14 days" query can return zero even when the CLI is working correctly:

```bash
insight sql \
  "SELECT count() AS recent_14d FROM traces WHERE timestamp >= now() - INTERVAL 14 DAY" \
  --json
```

## Common commands

### Trace exploration

```bash
insight traces list --limit 20
insight traces list --name "conversation_turn" --limit 10
insight traces get "<trace-id>" --json
```

Use `--from` and `--to` for old projects so the query window matches the data:

```bash
insight traces list \
  --from "2026-01-01T00:00:00Z" \
  --to "2026-02-01T00:00:00Z" \
  --limit 20
```

### SQL analytics

```bash
insight sql "SELECT name, count() AS cnt FROM traces GROUP BY name ORDER BY cnt DESC LIMIT 10"
insight sql "SELECT type, count() AS cnt FROM observations GROUP BY type ORDER BY cnt DESC" --json
insight sql "SELECT name, avg(value) AS avg_value FROM scores GROUP BY name ORDER BY name" --csv
```

The CLI SQL command uses raw table names: `traces`, `observations`, and `scores`.

### Dashboards and viewer configs

```bash
insight dashboards list
insight widgets list
insight trace-viewer-configs list
```

Create, update, and delete commands are also available, but keep smoke tests read-only unless you intentionally want to mutate the project.

### Skills

```bash
insight skills list
insight skills run health-check
insight skills run analytics
```

Built-in skills use the same REST and SQL APIs as the rest of the CLI. They are useful for guided diagnostics, but they may read more data than the minimal smoke commands above.

## OTLP endpoint note

`INSIGHT_OTLP_ENDPOINT` is used by SDK and OpenTelemetry export flows. The `insight` CLI does not read that variable. For trace export, see [OTel Ingestion](/reference/insight-otel-ingestion) and [Insight Export](/tracing/insight-export).

## Troubleshooting

### `fetch failed`

The CLI could not reach the host. Check network access, DNS, proxy settings, and that `INSIGHT_HOST` has the correct scheme:

```bash
export INSIGHT_HOST="https://insight.withmetis.ai"
```

### `401` or `403`

The project keys are missing, copied incorrectly, revoked, or scoped to a different project. Regenerate keys in Insight and retry with fresh environment variables.

### Empty trace results

Empty results do not always mean auth is broken. Check the project date range:

```bash
insight sql \
  "SELECT count() AS total, min(timestamp) AS earliest, max(timestamp) AS latest FROM traces" \
  --json
```

Then rerun trace and SQL commands with date filters that overlap the returned range.

## Next steps

- [Insight REST API](/reference/insight-rest-api)
- [Insight SQL API](/reference/insight-sql-api)
- [OTel Ingestion](/reference/insight-otel-ingestion)

---

## Insight OTel Ingestion Semantic Conventions

*Source: /reference/insight-otel-ingestion*

# Insight OTel Ingestion Semantic Conventions

Insight accepts OpenTelemetry (OTLP) trace data and converts it into traces and observations. This page documents every span attribute the ingestion endpoint recognizes, how attributes are resolved across frameworks, and how observation types, usage, and costs are extracted.

## How It Works

When you send OTLP spans to Insight, the ingestion processor:

1. Extracts **resource attributes** (set once per exporter, e.g. `service.name`, `deployment.environment`)
2. Extracts **span attributes** (set per span)
3. Resolves each concept (input, output, model name, etc.) by checking multiple attribute keys in **priority order** -- the first non-empty value wins
4. Maps the span to an **observation type** (GENERATION, TOOL, AGENT, etc.) using a priority-ordered mapper registry
5. Extracts **usage details** (token counts) and **cost details** from framework-specific attributes
6. Produces a trace event (from the root span or spans with trace-level attributes) and an observation event (from every span)

### Attribute Priority

For each concept, the processor checks attribute keys in a defined order. For example, to resolve the model name it checks `insight.observation.model.name` first, then `langfuse.observation.model.name`, then `gen_ai.response.model`, and so on. **The first non-empty value found is used.**

Span attributes are checked before resource attributes. If a value is not found on the span, certain attributes (environment, session ID, run ID, tags) fall back to resource attributes.

---

## Supported Frameworks

Insight recognizes attributes from the following SDKs and frameworks out of the box. You do not need to do any special configuration -- just send OTLP spans and the processor will detect the format automatically.

| Framework / SDK | Attribute Namespace | Notes |
|---|---|---|
| **Mantis Insight SDK** | `insight.*` | Primary namespace, highest priority |
| **Langfuse SDK** | `langfuse.*` | Compatibility with upstream Langfuse |
| **OpenInference** | `openinference.*`, `input.value`, `output.value`, `llm.*` | Used by Arize, Agno, BeeAI, and others |
| **OTel GenAI Semconv** | `gen_ai.*` | Standard OpenTelemetry semantic conventions for generative AI |
| **Vercel AI SDK** | `ai.*`, `operation.name` | Vercel's AI SDK telemetry |
| **Pydantic AI** | `gen_ai.*`, `tool_arguments`, `tool_response` | Via Logfire or direct OTel export |
| **MLFlow** | `mlflow.spanInputs`, `mlflow.spanOutputs` | MLFlow tracing |
| **TraceLoop** | `traceloop.entity.*` | OpenLLMetry / TraceLoop SDK |
| **Google ADK** | `gcp.vertex.agent.*` | Google Agent Developer Kit |
| **LiveKit** | `lk.*` | LiveKit Agents framework |
| **Logfire** | `prompt`, `all_messages_events`, `logfire.msg` | Pydantic Logfire |
| **SmolAgents** | `input.value`, `output.value` | HuggingFace SmolAgents |
| **Pipecat** | `input`, `output` | Pipecat framework |
| **AgentOps** | `workflow.*`, `agent.*`, `final_output` | AgentOps SDK |
| **CrewAI** | `crewai.*`, `task_input`, `task_output` | CrewAI framework |
| **Semantic Kernel** | `gen_ai.content.prompt`, `gen_ai.content.completion` | Via span events (legacy) |

---

## Attribute Reference

The tables below list every attribute key that Insight recognizes, grouped by category. Within each group, keys are listed in **priority order** -- earlier entries take precedence.

### Trace-Level Attributes

These attributes are extracted from the **root span** (or any span that sets them) and applied to the trace.

#### Trace Name

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.name` | Insight SDK |
| 2 | `langfuse.trace.name` | Langfuse SDK |

Falls back to the OTel span name if neither is set.

#### Trace Input

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.input` | Insight SDK |
| 2 | `langfuse.trace.input` | Langfuse SDK |

#### Trace Output

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.output` | Insight SDK |
| 2 | `langfuse.trace.output` | Langfuse SDK |

#### Trace System Prompt

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.system_prompt` | Insight SDK |

Stored in trace metadata under the `system_prompt` key.

#### Trace Metadata

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.metadata` | Insight SDK |
| 2 | `langfuse.trace.metadata` | Langfuse SDK |

Accepts a JSON string. Nested metadata keys are also supported via the `langfuse.trace.metadata.*` prefix pattern (e.g. `langfuse.trace.metadata.myKey`).

#### Trace Public

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.public` | Insight SDK |
| 2 | `langfuse.trace.public` | Langfuse SDK |
| 3 | `langfuse.public` | Langfuse SDK (legacy) |

Boolean. Accepts `true`, `"true"`, or truthy values.

#### Trace Tags

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.trace.tags` | Insight SDK |
| 2 | `langfuse.trace.tags` | Langfuse SDK |
| 3 | `langfuse.tags` | Langfuse SDK (legacy) |
| 4 | `langfuse.observation.metadata.langfuse_tags` | Langfuse SDK (metadata) |
| 5 | `langfuse.trace.metadata.langfuse_tags` | Langfuse SDK (metadata) |
| 6 | `ai.telemetry.metadata.tags` | Vercel AI SDK |
| 7 | `tag.tags` | Generic |

Accepts a JSON array string (e.g. `'["prod", "v2"]'`) or an array value. Also checked on resource attributes as a fallback.

---

### User and Session

#### User ID

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.user.id` | Insight SDK |
| 2 | `langfuse.user.id` | Langfuse SDK |
| 3 | `user.id` | Generic OTel |
| 4 | `langfuse.observation.metadata.langfuse_user_id` | Langfuse SDK (metadata) |
| 5 | `langfuse.trace.metadata.langfuse_user_id` | Langfuse SDK (metadata) |
| 6 | `ai.telemetry.metadata.userId` | Vercel AI SDK |

#### Session ID

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.session.id` | Insight SDK |
| 2 | `langfuse.session.id` | Langfuse SDK |
| 3 | `session.id` | Generic OTel |
| 4 | `gen_ai.conversation.id` | OTel GenAI Semconv |
| 5 | `langfuse.observation.metadata.langfuse_session_id` | Langfuse SDK (metadata) |
| 6 | `langfuse.trace.metadata.langfuse_session_id` | Langfuse SDK (metadata) |
| 7 | `ai.telemetry.metadata.sessionId` | Vercel AI SDK |

Also checked on resource attributes as a fallback.

#### Run ID

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.run.id` | Insight SDK |
| 2 | `run.id` | Generic |
| 3 | `mantis.run.id` | Mantis SDK |
| 4 | `experiment.id` | Generic |

Also checked on resource attributes as a fallback. Automatically set when traces are created within a `mantisdk.run` context.

---

### Environment and Version

#### Environment

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.environment` | Insight SDK |
| 2 | `langfuse.environment` | Langfuse SDK |
| 3 | `deployment.environment.name` | OTel Resource Semconv |
| 4 | `deployment.environment` | OTel Resource Semconv (legacy) |

Checked on both span and resource attributes. Defaults to `"default"` if not set.

#### Version

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.version` | Insight SDK |
| 2 | `langfuse.version` | Langfuse SDK |
| 3 | `service.version` | OTel Resource Semconv |

#### Release

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.release` | Insight SDK |
| 2 | `langfuse.release` | Langfuse SDK |

---

### Observation-Level Attributes

These attributes are extracted from **every span** and applied to the observation.

#### Observation Type

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.type` | Insight SDK |
| 2 | `langfuse.observation.type` | Langfuse SDK |
| 3 | `openinference.span.kind` | OpenInference |
| 4 | `gen_ai.operation.name` | OTel GenAI Semconv |

See [Observation Type Mapping](#observation-type-mapping) below for the full mapping rules.

#### Observation Input

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.input` | Insight SDK |
| 2 | `langfuse.observation.input` | Langfuse SDK |
| 3 | `input.value` | OpenInference / SmolAgents |
| 4 | `input` | Pydantic AI / Pipecat |
| 5 | `traceloop.entity.input` | TraceLoop |
| 6 | `mlflow.spanInputs` | MLFlow |
| 7 | `gen_ai.input.messages` | OTel GenAI Semconv |
| 8 | `gen_ai.tool.call.arguments` | OTel GenAI (tools) |
| 9 | `ai.prompt.messages` | Vercel AI SDK |
| 10 | `ai.prompt` | Vercel AI SDK |
| 11 | `ai.toolCall.args` | Vercel AI SDK |
| 12 | `gcp.vertex.agent.llm_request` | Google ADK |
| 13 | `gcp.vertex.agent.tool_call_args` | Google ADK (tools) |
| 14 | `lk.input_text` | LiveKit |
| 15 | `prompt` | Logfire |
| 16 | `tool_arguments` | Pydantic AI (tools) |
| 17 | `workflow.input` | AgentOps |
| 18 | `agent.input` | AgentOps |
| 19 | `crewai.task.input` | CrewAI |
| 20 | `crewai.agent.input` | CrewAI |
| 21 | `task_input` | CrewAI (alt) |

Additional input sources via span events and prefix-based patterns are documented in [Span Events](#span-events) and [Prefix-Based Input/Output](#prefix-based-inputoutput).

#### Observation Output

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.output` | Insight SDK |
| 2 | `langfuse.observation.output` | Langfuse SDK |
| 3 | `output.value` | OpenInference / SmolAgents |
| 4 | `output` | Pydantic AI / Pipecat |
| 5 | `traceloop.entity.output` | TraceLoop |
| 6 | `mlflow.spanOutputs` | MLFlow |
| 7 | `gen_ai.output.messages` | OTel GenAI Semconv |
| 8 | `gen_ai.tool.call.result` | OTel GenAI (tools) |
| 9 | `ai.response.text` | Vercel AI SDK |
| 10 | `ai.result.text` | Vercel AI SDK (legacy, < 4.0) |
| 11 | `ai.toolCall.result` | Vercel AI SDK |
| 12 | `ai.response.object` | Vercel AI SDK |
| 13 | `ai.result.object` | Vercel AI SDK (legacy, < 4.0) |
| 14 | `ai.response.toolCalls` | Vercel AI SDK |
| 15 | `ai.result.toolCalls` | Vercel AI SDK (legacy, < 4.0) |
| 16 | `gcp.vertex.agent.llm_response` | Google ADK |
| 17 | `gcp.vertex.agent.tool_response` | Google ADK (tools) |
| 18 | `lk.function_tool.output` | LiveKit |
| 19 | `lk.response.text` | LiveKit |
| 20 | `all_messages_events` | Logfire |
| 21 | `tool_response` | Pydantic AI (tools) |
| 22 | `workflow.output` | AgentOps |
| 23 | `final_output` | AgentOps |
| 24 | `agent.output` | AgentOps |
| 25 | `crewai.task.output` | CrewAI |
| 26 | `crewai.agent.output` | CrewAI |
| 27 | `task_output` | CrewAI (alt) |
| 28 | `result` | Generic fallback |

**Vercel AI SDK special case:** When both `ai.response.text` and `ai.response.toolCalls` are present, the output is combined into a structured object with `role`, `content`, and `tool_calls` fields.

#### Observation Metadata

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.metadata` | Insight SDK |
| 2 | `langfuse.observation.metadata` | Langfuse SDK |
| 3 | `langfuse.metadata` | Langfuse SDK (shorthand) |
| 4 | `ai.telemetry.metadata` | Vercel AI SDK |

Accepts a JSON string. Nested keys via the `langfuse.observation.metadata.*` prefix are also supported.

#### Observation Level

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.level` | Insight SDK |
| 2 | `langfuse.observation.level` | Langfuse SDK |

Valid values: `DEBUG`, `DEFAULT`, `WARNING`, `ERROR`.

If not set, defaults to `ERROR` when the OTel span status code is `2` (ERROR), otherwise `DEFAULT`.

#### Observation Status Message

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.status_message` | Insight SDK |
| 2 | `langfuse.observation.status_message` | Langfuse SDK |

Falls back to the OTel span `status.message` field.

---

### Model and Generation

#### Model Name

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.model.name` | Insight SDK |
| 2 | `langfuse.observation.model.name` | Langfuse SDK |
| 3 | `gen_ai.response.model` | OTel GenAI Semconv |
| 4 | `ai.model.id` | Vercel AI SDK |
| 5 | `gen_ai.request.model` | OTel GenAI Semconv |
| 6 | `llm.response.model` | OpenInference |
| 7 | `llm.model_name` | OpenInference (legacy) |
| 8 | `model` | Generic |

#### Model Parameters

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `insight.observation.model.parameters` | Insight SDK (JSON string) |
| 2 | `langfuse.observation.model.parameters` | Langfuse SDK (JSON string) |
| 3 | `llm.invocation_parameters` | OpenInference (JSON string) |
| 4 | `model_config` | Generic (JSON string) |

Additionally, any attributes matching the `gen_ai.request.*` prefix (excluding `gen_ai.request.model`) are collected as model parameters.

**Vercel AI SDK special case:** When the instrumentation scope is `ai`, the following attributes are extracted into model parameters:

| Attribute | Parameter Key |
|---|---|
| `ai.settings.maxSteps` | `maxSteps` |
| `ai.prompt.toolChoice` | `toolChoice` |
| `gen_ai.request.max_tokens` | `maxTokens` |
| `gen_ai.response.finish_reasons` or `gen_ai.finishReason` | `finishReason` |
| `gen_ai.system` or `ai.model.provider` | `system` |
| `ai.settings.maxRetries` | `maxRetries` |
| `ai.settings.mode` | `mode` |
| `gen_ai.request.temperature` | `temperature` |

#### Completion Start Time

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `langfuse.observation.completion_start_time` | Langfuse SDK |

For Vercel AI SDK, computed from `ai.response.msToFirstChunk` or `ai.stream.msToFirstChunk` added to the span start time.

---

### Prompt Management

#### Prompt Name

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `langfuse.observation.prompt.name` | Langfuse SDK |
| 2 | `langfuse.prompt.name` | Langfuse SDK (shorthand) |

Also extracted from Vercel AI SDK's `ai.telemetry.metadata.langfusePrompt` JSON field.

#### Prompt Version

| Priority | Attribute Key | Source |
|---|---|---|
| 1 | `langfuse.observation.prompt.version` | Langfuse SDK |
| 2 | `langfuse.prompt.version` | Langfuse SDK (shorthand) |

Also extracted from Vercel AI SDK's `ai.telemetry.metadata.langfusePrompt` JSON field.

---

## Observation Type Mapping

Every span is mapped to an observation type using a priority-ordered registry. Mappers are evaluated from lowest priority number (highest precedence) to highest. The first mapper that both matches and returns a type wins.

### Valid Observation Types

`SPAN`, `GENERATION`, `EVENT`, `EMBEDDING`, `AGENT`, `TOOL`, `CHAIN`, `RETRIEVER`, `GUARDRAIL`, `EVALUATOR`

If no mapper matches, the span defaults to type `SPAN`.

### Mapper Priority Order

#### Priority 0 -- Python SDK v3.3.0 and Earlier Override

Handles a known bug in older Langfuse Python SDK versions where generation-like spans were incorrectly labeled as `span`. If the scope name is `langfuse-sdk`, the SDK language is `python`, and version is 3.3.0 or earlier, the presence of generation attributes (model, usage, cost, etc.) overrides the type to `GENERATION`.

#### Priority 1 -- Direct Type Attribute

Reads `langfuse.observation.type` or `insight.observation.type` and maps directly:

| Attribute Value | Observation Type |
|---|---|
| `span` | SPAN |
| `generation` | GENERATION |
| `event` | EVENT |
| `embedding` | EMBEDDING |
| `agent` | AGENT |
| `tool` | TOOL |
| `chain` | CHAIN |
| `retriever` | RETRIEVER |
| `guardrail` | GUARDRAIL |
| `evaluator` | EVALUATOR |

#### Priority 2 -- OpenInference

Reads `openinference.span.kind`:

| Attribute Value | Observation Type |
|---|---|
| `CHAIN` | CHAIN |
| `RETRIEVER` | RETRIEVER |
| `LLM` | GENERATION |
| `EMBEDDING` | EMBEDDING |
| `AGENT` | AGENT |
| `TOOL` | TOOL |
| `GUARDRAIL` | GUARDRAIL |
| `EVALUATOR` | EVALUATOR |

#### Priority 3 -- OTel GenAI Operation

Reads `gen_ai.operation.name`:

| Attribute Value | Observation Type |
|---|---|
| `chat` | GENERATION |
| `completion` | GENERATION |
| `text_completion` | GENERATION |
| `generate_content` | GENERATION |
| `generate` | GENERATION |
| `embeddings` | EMBEDDING |
| `invoke_agent` | AGENT |
| `create_agent` | AGENT |
| `execute_tool` | TOOL |

#### Priority 4 -- Vercel AI SDK (Generation / Embedding)

Matches when model information is present (`ai.model.id`, `gen_ai.request.model`, etc.) **and** the operation name matches one of:

| Operation Name Prefix | Observation Type |
|---|---|
| `ai.generateText.doGenerate` | GENERATION |
| `ai.streamText.doStream` | GENERATION |
| `ai.generateObject.doGenerate` | GENERATION |
| `ai.streamObject.doStream` | GENERATION |
| `ai.embedMany.doEmbed` | EMBEDDING |
| `ai.embed.doEmbed` | EMBEDDING |

Checked via both `operation.name` (prefix match) and `ai.operationId` (exact match).

#### Priority 5 -- Vercel AI SDK (Tool Calls)

Matches any `ai.*` operation that is **not** a generation or embedding:

| Operation Name Prefix | Observation Type |
|---|---|
| `ai.toolCall` | TOOL |

#### Priority 6 -- GenAI Tool Call Detection

Matches when `gen_ai.tool.name` or `gen_ai.tool.call.id` is present. Used by Pydantic AI and other frameworks that set GenAI tool attributes without setting `gen_ai.operation.name`.

Result: **TOOL**

#### Priority 7 -- Model-Based Fallback

If any model-related attribute is present (`langfuse.observation.model.name`, `gen_ai.request.model`, `gen_ai.response.model`, `llm.model_name`, or `model`), the span is classified as a **GENERATION**.

---

## Usage and Cost Extraction

### Token Usage

Token usage is extracted differently depending on the instrumentation source. The processor normalizes all formats into a standard `usageDetails` object with keys like `input`, `output`, `total`, `input_cached_tokens`, `output_reasoning_tokens`, etc.

#### Langfuse / Insight SDK

| Attribute | Description |
|---|---|
| `langfuse.observation.usage_details` | JSON string with full usage details |

#### Vercel AI SDK (scope name: `ai`)

| Attribute | Maps To |
|---|---|
| `gen_ai.usage.prompt_tokens` or `gen_ai.usage.input_tokens` | `input` |
| `gen_ai.usage.completion_tokens` or `gen_ai.usage.output_tokens` | `output` |
| `ai.usage.tokens` | `total` |
| `ai.usage.cachedInputTokens` | `input_cached_tokens` |
| `ai.usage.reasoningTokens` | `output_reasoning_tokens` |

**Provider metadata fallback:** If the extended usage attributes are not present, the processor reads `ai.response.providerMetadata` (a JSON string) and extracts provider-specific details:

- **OpenAI:** `cachedPromptTokens`, `acceptedPredictionTokens`, `rejectedPredictionTokens`, `reasoningTokens`
- **Anthropic:** `cacheCreationInputTokens`, `cacheReadInputTokens`
- **Bedrock:** `usage.cacheReadInputTokens`, `usage.cacheWriteInputTokens`, `usage.cacheCreationInputTokens`

Cached token counts are subtracted from the total `input` count to avoid double-counting.

#### Pydantic AI (scope name: `pydantic-ai`)

| Attribute | Maps To |
|---|---|
| `gen_ai.usage.input_tokens` | `input` |
| `gen_ai.usage.output_tokens` | `output` |
| `gen_ai.usage.cache_read_tokens` or `gen_ai.usage.details.cache_read_input_tokens` | `input_cache_read` |
| `gen_ai.usage.cache_write_tokens` or `gen_ai.usage.details.cache_creation_input_tokens` | `input_cache_creation` |

#### Generic (all other frameworks)

Any attributes matching `gen_ai.usage.*` or `llm.token_count.*` are collected. Standard key mappings:

| Raw Key Suffix | Normalized Key |
|---|---|
| `prompt_tokens` | `input` |
| `completion_tokens` | `output` |
| `total_tokens` | `total` |
| `input_tokens` | `input` |
| `output_tokens` | `output` |
| `prompt` | `input` |
| `completion` | `output` |

All other suffixes are kept as-is (e.g. `gen_ai.usage.reasoning_tokens` becomes `reasoning_tokens`).

### Cost

| Attribute | Description |
|---|---|
| `langfuse.observation.cost_details` | JSON string with cost breakdown (Langfuse SDK only) |
| `gen_ai.usage.cost` | Total cost as a number, stored as `{ total: value }` |

---

## Span Events

In addition to span attributes, the processor extracts input and output from **OTel span events** (the `events` array on a span). This is used by frameworks that follow the OTel GenAI event-based conventions.

### GenAI Message Events

These events are collected as the **input** to the observation:

| Event Name | Role |
|---|---|
| `gen_ai.system.message` | `system` |
| `gen_ai.user.message` | `user` |
| `gen_ai.assistant.message` | `assistant` |
| `gen_ai.tool.message` | `tool` |

Each event's attributes are included in the message object.

### GenAI Choice Events

| Event Name | Usage |
|---|---|
| `gen_ai.choice` | Extracted as the **output** |

If a single choice event exists, it is returned as the output directly. Multiple choice events are returned as an array.

### Legacy Semantic Kernel Events

| Event Name | Usage |
|---|---|
| `gen_ai.content.prompt` | Input |
| `gen_ai.content.completion` | Output |

### Logfire Events Array

Logfire sends a stringified `events` attribute containing GenAI events. The processor parses this and extracts `gen_ai.choice` events as output and all other events as input.

---

## Prefix-Based Input/Output

Some frameworks use flattened attribute keys with numeric indices to represent structured messages. The processor reassembles these into nested objects:

### GenAI Prompt/Completion Prefix

Attributes starting with `gen_ai.prompt.*` are reassembled into an input object. Attributes starting with `gen_ai.completion.*` are reassembled into an output object. Used by TraceLoop and similar frameworks.

Example: `gen_ai.prompt.0.content` + `gen_ai.prompt.0.role` becomes `[{ content: "...", role: "..." }]`

### OpenInference LLM Messages Prefix

Attributes starting with `llm.input_messages.*` are reassembled into input. Attributes starting with `llm.output_messages.*` are reassembled into output. Used by Agno, BeeAI, and OpenInference instrumentors.

---

## Observation Name Extraction

The observation name is determined by this priority:

1. `gen_ai.tool.name` -- for tool-type observations
2. `logfire.msg` -- for Pydantic AI / Logfire spans
3. `ai.toolCall.name` -- for Vercel AI SDK tool calls
4. `ai.telemetry.functionId` + `ai.operationId` -- for Vercel AI SDK operations (formatted as `functionId:operationId`)
5. The OTel **span name** -- default fallback

---

## Resource Attributes

Resource attributes are set once per exporter instance and apply to all spans in the batch. The following attributes are checked on resource attributes as a **fallback** when not found on span attributes:

- `environment` (`deployment.environment.name`, `deployment.environment`)
- `session.id`
- `run.id`
- `tags`
- `version` (`service.version`)
- `release`

Resource attributes are also stored in the observation and trace metadata under the `resourceAttributes` key for debugging purposes.

---

## Quick Reference

### Insight SDK Attributes (Primary Namespace)

These are the recommended attributes when instrumenting with the Mantis Insight SDK:

```
insight.trace.name
insight.trace.input
insight.trace.output
insight.trace.system_prompt
insight.trace.metadata
insight.trace.tags
insight.trace.public
insight.user.id
insight.session.id
insight.run.id
insight.environment
insight.release
insight.version
insight.observation.type
insight.observation.input
insight.observation.output
insight.observation.metadata
insight.observation.level
insight.observation.status_message
insight.observation.model.name
insight.observation.model.parameters
```

### Using with MantisDK

If you are using the MantisDK Python SDK, these attributes are available as constants:

```python
import mantisdk.tracing as tracing

with tracing.trace("my-workflow", input=query) as span:
    span.set_attribute(tracing.semconv.USER_ID, "user-123")
    span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

See the [MantisDK Semantic Conventions](/tracing/semantic-conventions) page for the full SDK constant reference.

---

## Insight REST API

*Source: /reference/insight-rest-api*

# Insight REST API

Fetch traces, observations, scores, and sessions from Mantis Insight via the public REST API.

## Overview

Mantis Insight exposes a set of authenticated REST endpoints under `/api/public/` that let you programmatically retrieve data from your projects. This is useful for:

- Building custom dashboards or reports
- Exporting data for offline analysis
- Integrating Insight data into CI/CD pipelines
- Feeding evaluation results back into training workflows

## Authentication

All endpoints use **HTTP Basic Auth** with your project's API keys:

| Credential | Value |
|---|---|
| Username | Your **public key** (`pk-lf-...`) |
| Password | Your **secret key** (`sk-lf-...`) |

You can find these in the Insight UI under **Project Settings → API Keys**.

```bash
# Basic Auth header
curl -u "pk-lf-your-public-key:sk-lf-your-secret-key" \
  https://insight.withmetis.ai/api/public/traces
```

```python
import requests

PUBLIC_KEY = "pk-lf-..."
SECRET_KEY = "sk-lf-..."
HOST = "https://insight.withmetis.ai"

response = requests.get(
    f"{HOST}/api/public/traces",
    auth=(PUBLIC_KEY, SECRET_KEY),
)
traces = response.json()
```

API keys are scoped to a single project — all returned data belongs to that project.

---

## Traces

### List Traces

```
GET /api/public/traces
```

Returns a paginated list of traces.

#### Query Parameters

| Parameter | Type | Description |
|---|---|---|
| `page` | `integer` | Page number (default: 1) |
| `limit` | `integer` | Items per page (default: 50) |
| `userId` | `string` | Filter by user ID |
| `name` | `string` | Filter by trace name |
| `tags` | `string \| string[]` | Filter by tags |
| `environment` | `string \| string[]` | Filter by environment (e.g. `production`, `staging`) |
| `sessionId` | `string` | Filter by session ID |
| `version` | `string` | Filter by version |
| `release` | `string` | Filter by release |
| `fromTimestamp` | `ISO 8601` | Start of time range |
| `toTimestamp` | `ISO 8601` | End of time range |
| `orderBy` | `string` | Sort field and direction, e.g. `timestamp.desc` |
| `fields` | `string` | Comma-separated field groups: `core`, `io`, `scores`, `observations`, `metrics` |
| `filter` | `JSON string` | Advanced filter expression (JSON array) |

#### Response

```json
{
  "data": [
    {
      "id": "trace-abc-123",
      "timestamp": "2026-02-23T10:30:00.000Z",
      "name": "research-agent",
      "userId": "user-42",
      "sessionId": "session-789",
      "environment": "production",
      "tags": ["v2", "research"],
      "release": "1.2.0",
      "version": "2",
      "public": false,
      "bookmarked": false,
      "input": { "query": "What is quantum computing?" },
      "output": { "answer": "Quantum computing is..." },
      "metadata": { "source": "api" },
      "projectId": "proj-...",
      "createdAt": "2026-02-23T10:30:00.000Z",
      "updatedAt": "2026-02-23T10:30:05.000Z",
      "totalCost": 0.0023,
      "latency": 4.5,
      "htmlPath": "/project/proj-.../traces/trace-abc-123",
      "observations": ["obs-1", "obs-2"],
      "scores": ["score-1"]
    }
  ],
  "meta": {
    "page": 1,
    "limit": 50,
    "totalItems": 142,
    "totalPages": 3
  }
}
```

#### Examples

**Fetch the 10 most recent traces:**

```bash
curl -u "$PK:$SK" \
  "https://insight.withmetis.ai/api/public/traces?limit=10&orderBy=timestamp.desc"
```

**Filter by session and time range:**

```bash
curl -u "$PK:$SK" \
  "https://insight.withmetis.ai/api/public/traces?sessionId=sess-123&fromTimestamp=2026-02-01T00:00:00Z&toTimestamp=2026-02-23T23:59:59Z"
```

**Python — paginate through all traces:**

```python
import requests

PUBLIC_KEY = "pk-lf-..."
SECRET_KEY = "sk-lf-..."
HOST = "https://insight.withmetis.ai"

all_traces = []
page = 1

while True:
    resp = requests.get(
        f"{HOST}/api/public/traces",
        auth=(PUBLIC_KEY, SECRET_KEY),
        params={"page": page, "limit": 100, "orderBy": "timestamp.desc"},
    )
    resp.raise_for_status()
    data = resp.json()

    all_traces.extend(data["data"])

    if page >= data["meta"]["totalPages"]:
        break
    page += 1

print(f"Fetched {len(all_traces)} traces")
```

**Filter by tags:**

```python
resp = requests.get(
    f"{HOST}/api/public/traces",
    auth=(PUBLIC_KEY, SECRET_KEY),
    params={"tags": ["production", "v2"], "limit": 50},
)
```

---

### Get Single Trace

```
GET /api/public/traces/{traceId}
```

Returns a single trace with its full observation tree and scores.

#### Path Parameters

| Parameter | Type | Description |
|---|---|---|
| `traceId` | `string` | The trace ID |

#### Response

Returns the trace object plus:
- `observations` — full array of observation objects (LLM calls, tool calls, spans)
- `scores` — array of score objects attached to this trace
- `latency` — total latency in seconds
- `totalCost` — aggregated cost across all observations
- `htmlPath` — relative URL to view the trace in the Insight UI

#### Examples

```bash
curl -u "$PK:$SK" \
  "https://insight.withmetis.ai/api/public/traces/trace-abc-123"
```

```python
trace_id = "trace-abc-123"
resp = requests.get(
    f"{HOST}/api/public/traces/{trace_id}",
    auth=(PUBLIC_KEY, SECRET_KEY),
)
trace = resp.json()

print(f"Trace: {trace['name']}")
print(f"Latency: {trace['latency']}s")
print(f"Cost: ${trace['totalCost']:.4f}")
print(f"Observations: {len(trace['observations'])}")
print(f"Scores: {len(trace['scores'])}")
```

---

### Delete Traces

**Single trace:**

```
DELETE /api/public/traces/{traceId}
```

**Bulk delete (up to 1,000):**

```
DELETE /api/public/traces
Content-Type: application/json

{ "traceIds": ["trace-1", "trace-2", "trace-3"] }
```

---

## Observations

### List Observations

```
GET /api/public/observations
```

Returns a paginated list of observations (LLM generations, spans, tool calls).

#### Query Parameters

| Parameter | Type | Description |
|---|---|---|
| `page` | `integer` | Page number |
| `limit` | `integer` | Items per page |
| `traceId` | `string` | Filter by parent trace |
| `userId` | `string` | Filter by user ID |
| `name` | `string` | Filter by observation name |
| `type` | `string` | Filter by type (`GENERATION`, `SPAN`, `EVENT`) |
| `level` | `string` | Filter by level (`DEBUG`, `DEFAULT`, `WARNING`, `ERROR`) |
| `environment` | `string` | Filter by environment |
| `parentObservationId` | `string` | Filter by parent observation |
| `fromStartTime` | `ISO 8601` | Start of time range |
| `toStartTime` | `ISO 8601` | End of time range |
| `version` | `string` | Filter by version |

#### Example

```python
resp = requests.get(
    f"{HOST}/api/public/observations",
    auth=(PUBLIC_KEY, SECRET_KEY),
    params={"traceId": "trace-abc-123", "type": "GENERATION"},
)
generations = resp.json()["data"]

for gen in generations:
    print(f"{gen['name']}: {gen['model']} — {gen['totalTokens']} tokens")
```

### Get Single Observation

```
GET /api/public/observations/{observationId}
```

---

## Scores

### List Scores

```
GET /api/public/scores
```

Returns a paginated list of scores.

#### Query Parameters

| Parameter | Type | Description |
|---|---|---|
| `page` | `integer` | Page number |
| `limit` | `integer` | Items per page |
| `userId` | `string` | Filter by user ID |
| `name` | `string` | Filter by score name |
| `configId` | `string` | Filter by score config |
| `dataType` | `string` | Filter by data type |
| `source` | `string` | Filter by source (`API`, `EVAL`, `ANNOTATION`) |
| `fromTimestamp` | `ISO 8601` | Start of time range |
| `toTimestamp` | `ISO 8601` | End of time range |
| `environment` | `string` | Filter by environment |
| `value` | `number` | Filter by value |
| `operator` | `string` | Comparison operator for value filter |

#### Example

```python
resp = requests.get(
    f"{HOST}/api/public/scores",
    auth=(PUBLIC_KEY, SECRET_KEY),
    params={"name": "correctness", "source": "EVAL", "limit": 100},
)
scores = resp.json()["data"]

avg_score = sum(s["value"] for s in scores) / len(scores) if scores else 0
print(f"Average correctness: {avg_score:.2f}")
```

### Create Score

```
POST /api/public/scores
Content-Type: application/json

{
  "traceId": "trace-abc-123",
  "name": "correctness",
  "value": 0.95,
  "dataType": "NUMERIC",
  "comment": "Factually accurate response"
}
```

### Batch score-create via ingestion endpoint

When throughput matters, send score-create events in batches through `/api/public/ingestion` instead of issuing one `/api/public/scores` request per score.

```
POST /api/public/ingestion
Content-Type: application/json

{
  "batch": [
    {
      "id": "event-1",
      "type": "score-create",
      "timestamp": "2026-03-18T00:00:00.000Z",
      "body": {
        "id": "score-1",
        "name": "task_success",
        "value": 0.94,
        "dataType": "NUMERIC",
        "traceId": "trace-abc-123",
        "environment": "production"
      }
    },
    {
      "id": "event-2",
      "type": "score-create",
      "timestamp": "2026-03-18T00:00:00.000Z",
      "body": {
        "id": "score-2",
        "name": "policy_compliant",
        "value": 1,
        "dataType": "BOOLEAN",
        "traceId": "trace-abc-123",
        "environment": "production"
      }
    }
  ]
}
```

### Direct scores API vs ingestion-batch

| Mode | Endpoint | Best For | Tradeoff |
|---|---|---|---|
| Direct | `/api/public/scores` | Simpler single-score writes and immediate per-score responses | More HTTP overhead at scale |
| Batch | `/api/public/ingestion` (`score-create` events) | High-volume score ingestion pipelines | Slightly more payload shaping required |

Use direct mode when workflow simplicity matters. Use ingestion-batch mode for efficient bulk ingestion.

---

## Sessions

### Get Session

```
GET /api/public/sessions/{sessionId}
```

Returns a session with all its traces.

#### Example

```python
session_id = "session-789"
resp = requests.get(
    f"{HOST}/api/public/sessions/{session_id}",
    auth=(PUBLIC_KEY, SECRET_KEY),
)
session = resp.json()

print(f"Session: {session['id']}")
print(f"Environment: {session['environment']}")
print(f"Traces: {len(session['traces'])}")
```

---

## Complete Example: Export Traces for Analysis

```python
#!/usr/bin/env python3
"""Export traces and scores from Insight for offline analysis."""

import json
from datetime import datetime, timedelta
import requests

PUBLIC_KEY = "pk-lf-..."
SECRET_KEY = "sk-lf-..."
HOST = "https://insight.withmetis.ai"
AUTH = (PUBLIC_KEY, SECRET_KEY)

def fetch_recent_traces(days=7, limit=100):
    """Fetch traces from the last N days."""
    from_ts = (datetime.utcnow() - timedelta(days=days)).isoformat() + "Z"

    all_traces = []
    page = 1

    while True:
        resp = requests.get(
            f"{HOST}/api/public/traces",
            auth=AUTH,
            params={
                "page": page,
                "limit": limit,
                "fromTimestamp": from_ts,
                "orderBy": "timestamp.desc",
            },
        )
        resp.raise_for_status()
        data = resp.json()
        all_traces.extend(data["data"])

        if page >= data["meta"]["totalPages"]:
            break
        page += 1

    return all_traces

def fetch_trace_details(trace_id):
    """Fetch a single trace with observations and scores."""
    resp = requests.get(
        f"{HOST}/api/public/traces/{trace_id}",
        auth=AUTH,
    )
    resp.raise_for_status()
    return resp.json()

def main():
    print("Fetching recent traces...")
    traces = fetch_recent_traces(days=7)
    print(f"Found {len(traces)} traces")

    results = []
    for i, trace in enumerate(traces[:10]):
        detail = fetch_trace_details(trace["id"])
        results.append({
            "id": detail["id"],
            "name": detail["name"],
            "latency": detail["latency"],
            "cost": detail["totalCost"],
            "num_observations": len(detail["observations"]),
            "scores": {
                s["name"]: s["value"] for s in detail["scores"]
            },
        })
        print(f"  [{i+1}/{min(len(traces), 10)}] {detail['name']} — "
              f"{detail['latency']:.1f}s, ${detail['totalCost']:.4f}")

    with open("trace_export.json", "w") as f:
        json.dump(results, f, indent=2, default=str)

    print(f"\nExported {len(results)} traces to trace_export.json")

if __name__ == "__main__":
    main()
```

## Error Handling

The API returns standard HTTP status codes:

| Code | Meaning |
|---|---|
| `200` | Success |
| `401` | Invalid or missing API keys |
| `403` | API key doesn't have access to the requested resource |
| `404` | Resource not found (e.g. invalid trace ID) |
| `429` | Rate limited — back off and retry |
| `500` | Internal server error |

```python
resp = requests.get(
    f"{HOST}/api/public/traces/{trace_id}",
    auth=AUTH,
)

if resp.status_code == 404:
    print(f"Trace {trace_id} not found")
elif resp.status_code == 429:
    print("Rate limited, retrying after backoff...")
elif resp.status_code != 200:
    print(f"Error {resp.status_code}: {resp.text}")
else:
    trace = resp.json()
```

## Rate Limits

API endpoints are rate-limited per project. If you hit a rate limit (`429` response), back off exponentially before retrying. For bulk exports, use reasonable page sizes (50–100) and add a small delay between requests.

## Next Steps

- [Insight Export](/tracing/insight-export) — Send traces to Insight from your application
- [Tracing Overview](/tracing) — Understand traces, spans, and attributes
- [Semantic Conventions](/tracing/semantic-conventions) — Standard attribute names

---

## Insight SQL API

*Source: /reference/insight-sql-api*

# Insight SQL API

Run read-only SQL queries directly against your ClickHouse analytics data.

## Overview

The SQL API (`POST /api/public/sql`) gives you direct SQL access to your traces, observations, and scores in ClickHouse. Unlike the REST API which returns pre-shaped objects, the SQL API lets you write arbitrary `SELECT` queries for custom analytics, aggregations, and cross-table joins.

Use it when you need:

- Custom aggregations the REST API doesn't support (percentiles, time-bucketed rollups, conditional counts)
- Cross-table joins (e.g. traces with their scores and observation costs in one query)
- Bulk data exploration with ClickHouse-native functions
- Building custom dashboards or reporting pipelines

---

## Authentication

Same Basic Auth as all other Insight public API endpoints — use your project API keys.

### Request

```
POST /api/public/sql
Content-Type: application/json
```

| Field   | Type    | Required | Default | Constraints      |
|---------|---------|----------|---------|------------------|
| `sql`   | string  | Yes      | —       | Max 10,000 chars |
| `limit` | integer | No       | 1,000   | Max 10,000       |
| `dedup` | boolean | No       | `true`  | Deduplicate rows at query time. See [Deduplication](#deduplication). |

### Response

```json
{
  "data": [
    { "id": "abc-123", "name": "my-trace", "timestamp": "2025-01-15 10:30:00.000" }
  ],
  "meta": {
    "rowCount": 1,
    "executionTimeMs": 42,
    "limitApplied": 5
  }
}
```

On error, the response includes an `error` field:

```json
{
  "data": [],
  "meta": { "rowCount": 0, "executionTimeMs": 0, "limitApplied": 1 },
  "error": "Table 'users' is not allowed. Allowed tables: traces, observations, scores."
}
```

---

## Available Tables

You have access to exactly three tables. All other tables are blocked.

| Table          | Primary key columns                              | Description                          |
|----------------|--------------------------------------------------|--------------------------------------|
| `traces`       | `project_id`, `toDate(timestamp)`, `id`          | Top-level LLM trace records          |
| `observations` | `project_id`, `type`, `toDate(start_time)`, `id` | Spans, generations, events in traces |
| `scores`       | `project_id`, `toDate(timestamp)`, `name`, `id`  | Eval, annotation, and API scores     |

---

## Best Practices

### Always filter by date

Tables are partitioned by month. Date filters let ClickHouse skip entire partitions. The performance impact is significant — on a dataset with ~2.8M traces:

| Query | Time |
|-------|------|
| `SELECT count() FROM traces` (no filter) | 194 ms |
| `... WHERE timestamp >= '2026-04-10'` (7 days) | 93 ms |
| `... WHERE timestamp >= '2026-04-17'` (1 day) | 76 ms |

The gap grows with dataset size. Without a date filter on multi-million-row tables, you risk hitting the 30-second timeout.

```sql
-- Day-level (uses partition key directly — fastest)
WHERE toDate(timestamp) = '2026-04-17'

-- Range
WHERE timestamp >= '2026-04-10' AND timestamp < '2026-04-17'
```

### Select only what you need

The `input` and `output` columns are large ZSTD-compressed text blobs. Selecting them across many rows wastes bandwidth and memory budget.

```sql
-- Good: lightweight columns only
SELECT id, name, timestamp, user_id, session_id, tags
FROM traces ORDER BY timestamp DESC

-- OK when needed: use length() to check size without reading full content
SELECT id, name, length(input) as input_len, length(output) as output_len
FROM traces WHERE timestamp >= '2026-04-17'
```

### Use `count()` not `count(*)`

ClickHouse's `count()` (no arguments) is faster than `count(*)`.

### Use LIMIT

Always include a `LIMIT` unless you specifically need all rows. The API enforces one via the `limit` parameter, but an explicit `LIMIT` in your SQL can short-circuit query execution earlier.

---

## Query Patterns

### Basic queries

```sql
-- Recent traces
SELECT id, name, timestamp, user_id, session_id
FROM traces
WHERE timestamp >= '2026-04-10'
ORDER BY timestamp DESC
LIMIT 100

-- Count observations by type
SELECT type, count() as cnt
FROM observations
WHERE start_time >= '2026-04-10'
GROUP BY type

-- Score overview
SELECT name, data_type, avg(value) as avg_val, count() as cnt
FROM scores
WHERE timestamp >= '2026-04-01'
GROUP BY name, data_type
ORDER BY cnt DESC
```

### JOINs

Two-table JOINs work well. Always filter both sides by date.

```sql
-- Cost per trace
SELECT
    t.id,
    t.name,
    count() as span_count,
    sum(o.total_cost) as total_cost
FROM traces t
JOIN observations o ON t.id = o.trace_id
WHERE t.timestamp >= '2026-04-10'
  AND o.start_time >= '2026-04-10'
GROUP BY t.id, t.name
ORDER BY total_cost DESC NULLS LAST
LIMIT 10
```

For three-table JOINs, expect 5–10x slower execution. Use the **aggregate-first** pattern to avoid cartesian explosions:

```sql
-- Pre-aggregate scores, then join (faster than a flat 3-way JOIN)
SELECT t.name, agg.avg_score, agg.score_count
FROM traces t
JOIN (
    SELECT trace_id, avg(value) as avg_score, count() as score_count
    FROM scores
    WHERE name = 'correctness' AND timestamp >= '2026-04-10'
    GROUP BY trace_id
) agg ON t.id = agg.trace_id
WHERE t.timestamp >= '2026-04-10'
ORDER BY agg.score_count DESC
LIMIT 10
```

### Subqueries

Since `WITH` clauses are not supported, use inline subqueries wherever you'd normally use a CTE:

```sql
-- Traces with above-average span count
SELECT name, trace_count, avg_spans FROM (
    SELECT
        t.name,
        count(DISTINCT t.id) as trace_count,
        avg(span_cnt) as avg_spans
    FROM traces t
    JOIN (
        SELECT trace_id, count() as span_cnt
        FROM observations
        WHERE start_time >= '2026-04-10'
        GROUP BY trace_id
    ) o ON t.id = o.trace_id
    WHERE t.timestamp >= '2026-04-10'
    GROUP BY t.name
) sub
WHERE trace_count > 5
ORDER BY trace_count DESC
```

### Map columns

`metadata`, `usage_details`, `cost_details`, and `provided_usage_details` are `Map` types. Access keys with bracket notation.

```sql
-- Discover available metadata keys
SELECT mapKeys(metadata) as keys
FROM traces
WHERE timestamp >= '2026-04-15' AND length(mapKeys(metadata)) > 0
LIMIT 10

-- Token usage by model
SELECT
    provided_model_name,
    sum(usage_details['input'])  as input_tokens,
    sum(usage_details['output']) as output_tokens,
    sum(usage_details['total'])  as total_tokens,
    count() as calls
FROM observations
WHERE type = 'GENERATION'
  AND start_time >= '2026-04-10'
  AND provided_model_name != ''
GROUP BY provided_model_name
ORDER BY total_tokens DESC

-- Cost breakdown by model
SELECT
    provided_model_name,
    sum(cost_details['input'])  as input_cost,
    sum(cost_details['output']) as output_cost,
    sum(total_cost)             as total,
    count() as calls
FROM observations
WHERE type = 'GENERATION'
  AND start_time >= '2026-04-10'
  AND total_cost IS NOT NULL AND total_cost > 0
GROUP BY provided_model_name
ORDER BY total DESC
```

### Array columns

The `tags` column on traces is `Array(String)`.

```sql
-- Explode tags to individual rows
SELECT arrayJoin(tags) as tag, count() as cnt
FROM traces
WHERE timestamp >= '2026-04-10' AND length(tags) > 0
GROUP BY tag ORDER BY cnt DESC

-- Filter traces by specific tag
SELECT id, name FROM traces
WHERE has(tags, 'production') AND timestamp >= '2026-04-10'
LIMIT 10
```

### Time series

ClickHouse has built-in time bucketing functions.

```sql
-- Traces per hour
SELECT toStartOfHour(timestamp) as hour, count() as cnt
FROM traces
WHERE timestamp >= '2026-04-16'
GROUP BY hour ORDER BY hour

-- Daily dashboard with unique counts
SELECT
    toStartOfDay(timestamp) as day,
    count()              as traces,
    uniq(session_id)     as unique_sessions,
    uniq(user_id)        as unique_users
FROM traces
WHERE timestamp >= '2026-04-01'
GROUP BY day ORDER BY day
```

Available bucketing functions: `toStartOfMinute`, `toStartOfFiveMinutes`, `toStartOfFifteenMinutes`, `toStartOfHour`, `toStartOfDay`, `toStartOfWeek`, `toStartOfMonth`.

### Latency & error analysis

```sql
-- p50/p95/p99 latency per generation
SELECT
    name,
    quantile(0.5)(toUnixTimestamp64Milli(end_time) - toUnixTimestamp64Milli(start_time))  AS p50_ms,
    quantile(0.95)(toUnixTimestamp64Milli(end_time) - toUnixTimestamp64Milli(start_time)) AS p95_ms,
    quantile(0.99)(toUnixTimestamp64Milli(end_time) - toUnixTimestamp64Milli(start_time)) AS p99_ms,
    count() AS cnt
FROM observations
WHERE type = 'GENERATION'
  AND start_time >= '2026-04-10'
  AND end_time IS NOT NULL
GROUP BY name
HAVING cnt >= 10
ORDER BY p95_ms DESC

-- Error rate by operation
SELECT
    name,
    countIf(level = 'ERROR') as errors,
    count() as total,
    round(countIf(level = 'ERROR') * 100.0 / count(), 2) as error_rate_pct
FROM observations
WHERE start_time >= '2026-04-10' AND name != ''
GROUP BY name
HAVING total >= 10
ORDER BY error_rate_pct DESC
```

### Scores

Scores target either a `trace_id` or a `session_id` (never both). They come in three data types: `NUMERIC`, `BOOLEAN`, `CATEGORICAL`.

```sql
-- Score statistics
SELECT name, data_type, min(value), max(value), avg(value), count()
FROM scores
WHERE timestamp >= '2026-04-01'
GROUP BY name, data_type
ORDER BY count() DESC

-- Session-level scores
SELECT name, source, avg(value) as avg_val, count() as cnt
FROM scores
WHERE session_id IS NOT NULL AND session_id != ''
  AND timestamp >= '2026-04-01'
GROUP BY name, source
ORDER BY cnt DESC

-- Categorical score distribution
SELECT name, string_value, count() as cnt
FROM scores
WHERE data_type = 'CATEGORICAL'
  AND timestamp >= '2026-04-01'
  AND string_value IS NOT NULL
GROUP BY name, string_value
ORDER BY cnt DESC

-- Session duration and trace count
SELECT
    session_id,
    count() as trace_count,
    min(timestamp) as first_trace,
    max(timestamp) as last_trace
FROM traces
WHERE timestamp >= '2026-04-15'
  AND session_id IS NOT NULL AND session_id != ''
GROUP BY session_id
ORDER BY trace_count DESC
LIMIT 10
```

---

## Useful ClickHouse Functions

| Function | Example | Purpose |
|----------|---------|---------|
| `count()` | `count()` | Row count (faster than `count(*)`) |
| `countIf(cond)` | `countIf(level = 'ERROR')` | Conditional count |
| `uniq(col)` | `uniq(user_id)` | Approximate distinct count (fast) |
| `uniqExact(col)` | `uniqExact(session_id)` | Exact distinct count (slower) |
| `quantile(q)(expr)` | `quantile(0.95)(latency)` | Percentile |
| `toStartOfHour(ts)` | `toStartOfHour(timestamp)` | Time bucketing |
| `toDate(ts)` | `toDate(timestamp)` | Extract date |
| `arrayJoin(arr)` | `arrayJoin(tags)` | Explode array to rows |
| `has(arr, val)` | `has(tags, 'prod')` | Check array membership |
| `length(arr)` | `length(tags) > 0` | Array/string length |
| `mapKeys(m)` | `mapKeys(metadata)` | Get map keys |
| `m['key']` | `metadata['env']` | Access map value |
| `round(x, n)` | `round(rate, 2)` | Round to n decimals |
| `if(c, t, f)` | `if(cnt > 0, x/cnt, 0)` | Conditional expression |

---

## Resource Limits

| Limit | Value | HTTP status on breach |
|-------|-------|-----------------------|
| Max execution time | 30 seconds | 408 |
| Max memory | 4 GB | 413 |
| Max result rows | 10,000 | 413 |
| Max SQL length | 10,000 chars | 400 |
| Max `limit` param | 10,000 | 400 |

---

## Error Reference

| HTTP Status | Meaning |
|-------------|---------|
| `200` | Success |
| `400` | Invalid SQL, unknown column, disallowed table, or syntax error |
| `401` | Missing or invalid API keys |
| `403` | Mutation attempt (caught by ClickHouse `readonly: 1`) |
| `408` | Query timed out (30s limit) |
| `413` | Memory limit exceeded or result too large |
| `500` | Unexpected server error |

Error messages are descriptive:

```
"Table 'users' is not allowed. Allowed tables: traces, observations, scores."
"Column 'nonexistent_column' not found. Check table schema."
"Multi-statement SQL is not allowed. Remove semicolons."
"Only SELECT queries are allowed."
"SQL query must not be empty"
```

---

## Deduplication

ClickHouse tables use `ReplacingMergeTree`, which deduplicates rows during background merges. The `dedup` parameter controls whether deduplication is also enforced at query time.

**`dedup: true` (default)** — Applies ClickHouse's `FINAL` modifier, which merge-sorts all project data before returning results. This guarantees no duplicate rows and excludes soft-deleted records, but uses more memory and time on large projects.

**`dedup: false`** — Skips `FINAL` for faster queries with lower memory usage. ClickHouse can use primary key indexes and bloom filters to read only matching rows. The trade-off: very recently ingested data (before background merges complete, typically within seconds) may show duplicate rows, stale versions, or recently deleted records.

---

## Known Limitations

| Pattern | What happens | Workaround |
|---------|-------------|------------|
| `WITH name AS (...) SELECT ... FROM name` | Rejected: "Table 'name' is not allowed" | Use inline subqueries |
| `FROM default.traces` | Rejected: schema-qualified references | Use `FROM traces` |
| `SELECT 1; SELECT 2` | Rejected: multi-statement SQL | One query per request |
| `SETTINGS max_threads=4` | Rejected: SETTINGS clause | Settings are server-controlled |
| `INSERT`, `UPDATE`, `DELETE`, `DROP` | Rejected: read-only API | N/A |
| `FROM system.tables` | Rejected: system table access | N/A |

---

## Column Reference

### traces

| Column | Type | Notes |
|--------|------|-------|
| `id` | String | Trace ID |
| `timestamp` | DateTime64(3) | When the trace occurred |
| `name` | String | Trace name (can be empty) |
| `user_id` | Nullable(String) | End-user identifier |
| `metadata` | Map(String, String) | Key-value metadata |
| `release` | Nullable(String) | Release version |
| `version` | Nullable(String) | Trace version |
| `public` | Bool | Publicly accessible |
| `bookmarked` | Bool | Bookmarked |
| `tags` | Array(String) | Tags array |
| `input` | Nullable(String) | Input payload (large, ZSTD compressed) |
| `output` | Nullable(String) | Output payload (large, ZSTD compressed) |
| `session_id` | Nullable(String) | Session grouping ID |
| `environment` | Nullable(String) | Environment |
| `created_at` | DateTime64(3) | Record creation time |
| `updated_at` | DateTime64(3) | Last update time |

### observations

| Column | Type | Notes |
|--------|------|-------|
| `id` | String | Observation ID |
| `trace_id` | String | Parent trace ID (JOIN key) |
| `type` | LowCardinality(String) | `SPAN`, `GENERATION`, or `EVENT` |
| `parent_observation_id` | Nullable(String) | Parent observation |
| `start_time` | DateTime64(3) | Start timestamp |
| `end_time` | Nullable(DateTime64(3)) | End timestamp |
| `name` | String | Observation name |
| `metadata` | Map(String, String) | Key-value metadata |
| `level` | LowCardinality(String) | `DEFAULT`, `DEBUG`, `WARNING`, `ERROR` |
| `status_message` | Nullable(String) | Status message |
| `input` | Nullable(String) | Input (large, ZSTD compressed) |
| `output` | Nullable(String) | Output (large, ZSTD compressed) |
| `provided_model_name` | Nullable(String) | Model name as provided |
| `internal_model_id` | Nullable(String) | Resolved model ID |
| `provided_usage_details` | Map(String, UInt64) | Token usage as provided |
| `usage_details` | Map(String, UInt64) | Resolved token usage (`input`, `output`, `total`) |
| `provided_cost_details` | Map(String, Decimal64) | Cost as provided |
| `cost_details` | Map(String, Decimal64) | Resolved cost (`input`, `output`) |
| `total_cost` | Nullable(Decimal64(12)) | Total cost |
| `completion_start_time` | Nullable(DateTime64(3)) | Time to first token |
| `prompt_id` | Nullable(String) | Prompt template ID |
| `prompt_name` | Nullable(String) | Prompt template name |
| `prompt_version` | Nullable(UInt16) | Prompt template version |
| `environment` | Nullable(String) | Environment |

### scores

| Column | Type | Notes |
|--------|------|-------|
| `id` | String | Score ID |
| `timestamp` | DateTime64(3) | Score timestamp |
| `trace_id` | Nullable(String) | Target trace (XOR with session_id) |
| `session_id` | Nullable(String) | Target session (XOR with trace_id) |
| `observation_id` | Nullable(String) | Target observation |
| `name` | String | Score name |
| `value` | Float64 | Numeric value |
| `source` | String | `EVAL`, `ANNOTATION`, or `API` |
| `comment` | Nullable(String) | Score comment |
| `data_type` | String | `NUMERIC`, `CATEGORICAL`, `BOOLEAN` |
| `string_value` | Nullable(String) | Categorical value |
| `config_id` | Nullable(String) | Eval config ID |
| `queue_id` | Nullable(String) | Annotation queue ID |

---

## Next Steps

- [Insight REST API](/reference/insight-rest-api) — Fetch pre-shaped trace, observation, and score objects
- [OTel Ingestion](/reference/insight-otel-ingestion) — Send traces to Insight via OpenTelemetry
- [Tracing Overview](/tracing) — Understand traces, spans, and attributes

---

## Semantic Conventions API

*Source: /reference/semconv*

# Semantic Conventions API

Reference for `mantisdk.tracing.semconv` module.

## Overview

Semantic conventions provide standardized attribute names for consistent trace data.

```python
import mantisdk.tracing as tracing

span.set_attribute(tracing.semconv.USER_ID, "user-123")
```

## Trace-Level Attributes

For root spans (traces):

### `TRACE_NAME`

**Value:** `insight.trace.name`

Name of the trace.

```python
span.set_attribute(tracing.semconv.TRACE_NAME, "user-query-workflow")
```

### `TRACE_INPUT`

**Value:** `insight.trace.input`

Input to the trace.

```python
span.set_attribute(tracing.semconv.TRACE_INPUT, user_query)
```

### `TRACE_OUTPUT`

**Value:** `insight.trace.output`

Output from the trace.

```python
span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

### `TRACE_METADATA`

**Value:** `insight.trace.metadata`

Additional metadata (JSON string).

```python
span.set_attribute(tracing.semconv.TRACE_METADATA, '{"source": "api", "version": "1.0"}')
```

### `TRACE_TAGS`

**Value:** `insight.trace.tags`

Tags for filtering (JSON array string).

```python
span.set_attribute(tracing.semconv.TRACE_TAGS, '["production", "high-priority"]')
```

### `TRACE_PUBLIC`

**Value:** `insight.trace.public`

Whether the trace is publicly visible.

```python
span.set_attribute(tracing.semconv.TRACE_PUBLIC, True)
```

---

## User & Session

### `USER_ID`

**Value:** `insight.user.id`

Unique user identifier.

```python
span.set_attribute(tracing.semconv.USER_ID, "user-123")
```

### `SESSION_ID`

**Value:** `insight.session.id`

Session identifier.

```python
span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
```

---

## Environment & Version

### `ENVIRONMENT`

**Value:** `insight.environment`

Deployment environment.

```python
span.set_attribute(tracing.semconv.ENVIRONMENT, "production")
```

### `RELEASE`

**Value:** `insight.release`

Release identifier.

```python
span.set_attribute(tracing.semconv.RELEASE, "v1.2.3")
```

### `VERSION`

**Value:** `insight.version`

Version string.

```python
span.set_attribute(tracing.semconv.VERSION, "1.2.3")
```

---

## Observation-Level Attributes

For child spans (observations):

### `OBSERVATION_TYPE`

**Value:** `insight.observation.type`

Type of observation. Use `ObservationType` enum values.

```python
from mantisdk.tracing.semconv import ObservationType

span.set_attribute(tracing.semconv.OBSERVATION_TYPE, ObservationType.GENERATION)
```

### `OBSERVATION_INPUT`

**Value:** `insight.observation.input`

Input to the operation.

```python
span.set_attribute(tracing.semconv.OBSERVATION_INPUT, input_data)
```

### `OBSERVATION_OUTPUT`

**Value:** `insight.observation.output`

Output from the operation.

```python
span.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, output_data)
```

### `OBSERVATION_METADATA`

**Value:** `insight.observation.metadata`

Additional metadata (JSON string).

```python
span.set_attribute(tracing.semconv.OBSERVATION_METADATA, '{"items": 5}')
```

### `OBSERVATION_LEVEL`

**Value:** `insight.observation.level`

Log level. Use `ObservationLevel` enum values.

```python
from mantisdk.tracing.semconv import ObservationLevel

span.set_attribute(tracing.semconv.OBSERVATION_LEVEL, ObservationLevel.WARNING)
```

### `OBSERVATION_STATUS_MESSAGE`

**Value:** `insight.observation.status_message`

Status message.

```python
span.set_attribute(tracing.semconv.OBSERVATION_STATUS_MESSAGE, "Completed successfully")
```

---

## Model Attributes

### `MODEL_NAME`

**Value:** `insight.observation.model.name`

Model identifier.

```python
span.set_attribute(tracing.semconv.MODEL_NAME, "gpt-4o-mini")
```

### `MODEL_PARAMETERS`

**Value:** `insight.observation.model.parameters`

Model parameters (JSON string).

```python
span.set_attribute(tracing.semconv.MODEL_PARAMETERS, '{"temperature": 0.7, "max_tokens": 1000}')
```

---

## Enums

### `ObservationType`

Valid values for `OBSERVATION_TYPE`:

```python
from mantisdk.tracing.semconv import ObservationType

ObservationType.SPAN        # "span"
ObservationType.GENERATION  # "generation"
ObservationType.EVENT       # "event"
```

### `ObservationLevel`

Valid values for `OBSERVATION_LEVEL`:

```python
from mantisdk.tracing.semconv import ObservationLevel

ObservationLevel.DEBUG    # "DEBUG"
ObservationLevel.DEFAULT  # "DEFAULT"
ObservationLevel.WARNING  # "WARNING"
ObservationLevel.ERROR    # "ERROR"
```

---

## Complete Reference Table

| Constant | Value | Category |
|----------|-------|----------|
| `TRACE_NAME` | `insight.trace.name` | Trace |
| `TRACE_INPUT` | `insight.trace.input` | Trace |
| `TRACE_OUTPUT` | `insight.trace.output` | Trace |
| `TRACE_METADATA` | `insight.trace.metadata` | Trace |
| `TRACE_TAGS` | `insight.trace.tags` | Trace |
| `TRACE_PUBLIC` | `insight.trace.public` | Trace |
| `USER_ID` | `insight.user.id` | User |
| `SESSION_ID` | `insight.session.id` | Session |
| `ENVIRONMENT` | `insight.environment` | Environment |
| `RELEASE` | `insight.release` | Environment |
| `VERSION` | `insight.version` | Environment |
| `OBSERVATION_TYPE` | `insight.observation.type` | Observation |
| `OBSERVATION_INPUT` | `insight.observation.input` | Observation |
| `OBSERVATION_OUTPUT` | `insight.observation.output` | Observation |
| `OBSERVATION_METADATA` | `insight.observation.metadata` | Observation |
| `OBSERVATION_LEVEL` | `insight.observation.level` | Observation |
| `OBSERVATION_STATUS_MESSAGE` | `insight.observation.status_message` | Observation |
| `MODEL_NAME` | `insight.observation.model.name` | Model |
| `MODEL_PARAMETERS` | `insight.observation.model.parameters` | Model |

---

## Tracing Module API

*Source: /reference/tracing*

# Tracing Module API

Complete API reference for `mantisdk.tracing`.

## Initialization Functions

### `init()`

Initialize the tracing system.

```python
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:**

```python
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.

```python
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:**

```python
# Enable specific instrumentors
tracing.instrument(names=["anthropic", "dspy"])

# Enable all except langchain
tracing.instrument(skip=["langchain"])
```

---

### `shutdown()`

Flush and shutdown the tracer provider.

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

**Parameters:**

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

**Example:**

```python
import atexit
import mantisdk.tracing as tracing

tracing.init()
atexit.register(tracing.shutdown)
```

---

### `flush()`

Force flush pending spans without shutting down.

```python
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:**

```python
# 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.

```python
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:**

```python
# 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).

```python
@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:**

```python
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.

```python
@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:**

```python
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.

```python
@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:**

```python
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.

```python
@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:**

```python
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.

```python
@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:**

```python
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.

```python
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:**

```python
# 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.

```python
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.

```python
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.

```python
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.

```python
def get_current_span() -> Span
```

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

**Example:**

```python
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.

```python
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:**

```python
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.

```python
from mantisdk.tracing.exporters.insight import is_insight_configured

def is_insight_configured() -> bool
```

**Returns:** `True` if all required env vars are set.

**Example:**

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

---

## Error Handling Functions

### `record_exception()`

Record an exception on a span with proper attributes.

```python
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:**

```python
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.

```python
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:**

```python
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.

```python
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:**

```python
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.

```python
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:**

```python
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.

```python
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:**

```python
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

```python
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

```python
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

```python
@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

```python
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

```python
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
```

---

## Tracing Overview

*Source: /tracing*

# Tracing Overview

Mantis SDK tracing provides OpenTelemetry-based distributed tracing for AI agents. It enables you to capture the full execution flow of your agent, including LLM calls, tool executions, and custom logic.

## Core Concepts

### Traces

A **trace** represents a complete request or workflow through your system. It's the top-level container that holds all related operations.

```python
with tracing.trace("my-workflow", input=user_query) as span:
    # Everything inside is part of this trace
    result = process(user_query)
```

### Spans

A **span** represents a single operation within a trace. Spans can be nested to show parent-child relationships.

```python
with tracing.trace("workflow") as root:
    with tracing.span("step_1") as s1:
        # Child of root
        pass
    with tracing.span("step_2") as s2:
        # Another child of root
        pass
```

### Attributes

**Attributes** are key-value pairs attached to spans that provide context. Mantis SDK provides semantic conventions for common attributes.

```python
span.set_attribute(tracing.semconv.USER_ID, "user-123")
span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
span.set_attribute("custom_key", "custom_value")
```

## Tracing API

Mantis SDK provides three main ways to create spans:

| Function | Purpose | Use Case |
|----------|---------|----------|
| `trace()` | Create root spans | Main workflow entry points |
| `span()` | Create child spans | Sub-operations within a trace |
| `tool()` | Create tool spans | External calls, API requests |

### Decorators vs Context Managers

All three functions work as both decorators and context managers:

```python
# Decorator - auto-captures input/output
@tracing.trace
def my_function(query: str) -> dict:
    return {"result": query.upper()}

# Context manager - explicit control
with tracing.trace("my-operation", input=query) as span:
    result = do_work()
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

## Data Flow

```
Your Application
      │
      ▼
  tracing.trace() / span() / tool()
      │
      ▼
  OpenTelemetry SDK
      │
      ▼
  BatchSpanProcessor
      │
      ▼
  Insight Exporter (OTLP)
      │
      ▼
  Mantis Insight Backend
```

## Module Structure

```python
import mantisdk.tracing as tracing

# Initialization
tracing.init()           # Initialize tracing
tracing.instrument()     # Enable auto-instrumentation
tracing.shutdown()       # Graceful shutdown
tracing.flush()          # Force flush spans

# Span creation (sync)
tracing.trace            # Root spans (decorator/context manager)
tracing.span             # Child spans (context manager)
tracing.tool             # Tool spans (context manager)

# Span creation (async)
tracing.atrace           # Async root spans
tracing.aspan            # Async child spans

# Utilities
tracing.get_current_span()  # Get active span
tracing.insight_exporter()  # Create Insight exporter

# Semantic conventions
tracing.semconv.USER_ID
tracing.semconv.SESSION_ID
tracing.semconv.TRACE_INPUT
tracing.semconv.TRACE_OUTPUT
# ... and more
```

## Next Steps

- [Initialization](/tracing/initialization) - Configure tracing
- [Traces & Spans](/tracing/traces-and-spans) - Create spans
- [Semantic Conventions](/tracing/semantic-conventions) - Standard attributes
- [Auto-Instrumentation](/tracing/auto-instrumentation) - Framework support

---

## Async Support

*Source: /tracing/async-support*

# Async Support

Mantis SDK provides full async/await support for tracing asynchronous code.

## Async Functions

### Using `@tracing.trace` Decorator

The `@tracing.trace` decorator automatically detects async functions:

```python
import mantisdk.tracing as tracing

@tracing.trace
async def async_process(query: str) -> dict:
    """Async function - decorator works automatically."""
    result = await some_async_operation(query)
    return {"result": result}

# Usage
result = await async_process("hello")
```

### Using `atrace` Context Manager

For explicit async context manager usage:

```python
import mantisdk.tracing as tracing

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

## Async Spans

### Using `aspan` for Child Spans

```python
import mantisdk.tracing as tracing

async def my_workflow(query: str):
    async with tracing.atrace("workflow", input=query):
        # Async child span
        async with tracing.aspan("fetch_data") as s:
            data = await fetch_from_api()
            s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(data)))

        # Another async child span
        async with tracing.aspan("process_data") as s:
            result = await process(data)
            s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, result)

        return result
```

## Mixing Sync and Async

You can use sync spans inside async functions:

```python
import mantisdk.tracing as tracing

async def hybrid_workflow(query: str):
    # Async trace
    async with tracing.atrace("workflow", input=query) as root:
        # Sync span (still works in async context)
        with tracing.span("sync_operation") as s:
            sync_result = sync_function()
            s.set_attribute("sync_result", sync_result)

        # Async span
        async with tracing.aspan("async_operation") as s:
            async_result = await async_function()
            s.set_attribute("async_result", async_result)

        root.set_attribute(tracing.semconv.TRACE_OUTPUT, async_result)
        return async_result
```

## `atrace` vs `aspan`

| Function | Purpose | Use Case |
|----------|---------|----------|
| `atrace` | Async root span | Top-level async workflows |
| `aspan` | Async child span | Sub-operations in async code |

```python
async def main():
    # atrace for the root
    async with tracing.atrace("main_workflow") as root:
        # aspan for children
        async with tracing.aspan("step_1"):
            await step_one()

        async with tracing.aspan("step_2"):
            await step_two()
```

## Parameters

Both `atrace` and `aspan` accept the same parameters:

```python
async with tracing.atrace(
    name,                        # Span name (required)
    input=None,                  # Input value
    kind=SpanKind.INTERNAL,      # OpenTelemetry span kind
    attributes=None,             # Initial attributes dict
    record_exception=True,       # Record exceptions on span
    set_status_on_exception=True # Set ERROR status on exception
) as span:
    ...
```

## Concurrent Operations

Trace concurrent async operations:

```python
import asyncio
import mantisdk.tracing as tracing

async def fetch_all(urls: list[str]):
    async with tracing.atrace("fetch_all", input=str(urls)):
        # Run concurrent fetches
        tasks = []
        for url in urls:
            tasks.append(fetch_url(url))

        results = await asyncio.gather(*tasks)
        return results

async def fetch_url(url: str):
    """Each concurrent task gets its own span."""
    async with tracing.aspan(f"fetch:{url}", input=url) as span:
        result = await http_get(url)
        span.set_attribute("status_code", result.status)
        return result
```

## Complete Async Example

```python
#!/usr/bin/env python3
"""Complete async tracing example."""

import asyncio
from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing

@tracing.trace
async def classify_query(query: str) -> str:
    """Async classification - decorator auto-captures I/O."""
    await asyncio.sleep(0.1)  # Simulate async work
    if "weather" in query.lower():
        return "weather"
    return "general"

@tracing.trace
async def generate_response(query: str, query_type: str) -> str:
    """Async generation - decorator auto-captures I/O."""
    await asyncio.sleep(0.2)  # Simulate LLM call
    return f"Response for {query_type} query: {query}"

async def agent_workflow(user_id: str, query: str):
    """Main async workflow with explicit spans."""
    async with tracing.atrace("agent_session", input=query) as root:
        # Set user context
        root.set_attribute(tracing.semconv.USER_ID, user_id)

        # Classify (uses decorator, auto-traced)
        query_type = await classify_query(query)

        # Generate (uses decorator, auto-traced)
        response = await generate_response(query, query_type)

        # Set output
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, response)

        return response

async def main():
    tracing.init(service_name="async-agent")

    try:
        result = await agent_workflow(
            user_id="user-123",
            query="What's the weather like?"
        )
        print(f"Result: {result}")
    finally:
        tracing.shutdown()

if __name__ == "__main__":
    asyncio.run(main())
```

## Next Steps

- [Traces & Spans](/tracing/traces-and-spans) - Sync span creation
- [Auto-Instrumentation](/tracing/auto-instrumentation) - Framework support

---

## Auto-Instrumentation

*Source: /tracing/auto-instrumentation*

# Auto-Instrumentation

Mantis SDK can automatically instrument popular AI frameworks to capture LLM calls without manual code changes.

## Overview

Auto-instrumentation uses OpenTelemetry instrumentors to automatically trace:
- LLM API calls (tokens, latency, model)
- Tool/function calls
- Framework-specific operations

## Enabling Instrumentation

### Default Instrumentation

Enable the core set of instrumentors:

```python
import mantisdk.tracing as tracing

tracing.init(instrument_default=True)
```

This enables instrumentation for:
- OpenAI
- Anthropic
- LangChain
- LlamaIndex
- LiteLLM

### Selective Instrumentation

Enable specific instrumentors only:

```python
import mantisdk.tracing as tracing

tracing.init(instrument_default=False)

# Enable only what you need
tracing.instrument(names=["anthropic", "dspy"])
```

### Skipping Instrumentors

Enable defaults but skip specific ones:

```python
import mantisdk.tracing as tracing

tracing.init(instrument_default=False)
tracing.instrument(skip=["langchain"])  # Enable all except LangChain
```

## Supported Frameworks

### Core Set (Default)

| Framework | Instrumentor Name | Auto-Captured |
|-----------|------------------|---------------|
| OpenAI | `openai` | Chat completions, embeddings, tokens |
| Anthropic | `anthropic` | Messages, tool use, tokens |
| LangChain | `langchain` | Chains, agents, tools, retrievers |
| LlamaIndex | `llama_index` | Queries, retrievers, indices |
| LiteLLM | `litellm` | Multi-provider LLM calls |

### Additional Frameworks

| Framework | Instrumentor Name | Auto-Captured |
|-----------|------------------|---------------|
| DSPy | `dspy` | Modules, signatures, optimizers |
| Google ADK | `google_adk` | Agents, tools, sessions |
| Mistral | `mistral` | Chat completions, embeddings |
| Groq | `groq` | Chat completions |
| Bedrock | `bedrock` | AWS Bedrock model calls |

## Usage Examples

### DSPy

```python
import mantisdk.tracing as tracing

# Initialize and instrument DSPy
tracing.init(service_name="dspy-agent", instrument_default=False)
tracing.instrument(names=["dspy"])

# Your DSPy code is now automatically traced
import dspy

class MyModule(dspy.Module):
    def forward(self, question):
        # LLM calls are auto-traced
        return self.chain_of_thought(question=question)
```

### Anthropic

```python
import mantisdk.tracing as tracing

tracing.init(service_name="anthropic-agent", instrument_default=False)
tracing.instrument(names=["anthropic"])

# All Anthropic API calls are now traced
import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "Hello!"}]
)
# ^ This call is automatically traced with:
#   - Model name
#   - Input/output tokens
#   - Latency
#   - Response content
```

### Google ADK

```python
import mantisdk.tracing as tracing

tracing.init(service_name="adk-agent", instrument_default=False)
tracing.instrument(names=["google_adk"])

# Google ADK operations are traced
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner

agent = Agent(name="my-agent", model="gemini-2.0-flash", tools=[...])
runner = InMemoryRunner(agent=agent)
# Agent runs are automatically traced
```

### LangChain

```python
import mantisdk.tracing as tracing

tracing.init(instrument_default=False)
tracing.instrument(names=["langchain"])

# LangChain chains and agents are traced
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain

llm = ChatOpenAI()
chain = LLMChain(llm=llm, prompt=my_prompt)
result = chain.invoke({"input": "Hello"})
# ^ Chain execution and LLM calls are traced
```

## Combining Manual and Auto Tracing

Auto-instrumentation works alongside manual tracing:

```python
import mantisdk.tracing as tracing

tracing.init(service_name="my-agent", instrument_default=False)
tracing.instrument(names=["anthropic"])

import anthropic

@tracing.trace
def process_query(query: str) -> str:
    """Manual trace wraps auto-instrumented calls."""
    with tracing.span("prepare") as s:
        prepared = prepare_query(query)
        s.set_attribute("prepared_length", len(prepared))

    # This Anthropic call is auto-traced as a child span
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        messages=[{"role": "user", "content": prepared}]
    )

    return response.content[0].text
```

The resulting trace hierarchy:

```
process_query (manual trace)
├── prepare (manual span)
└── anthropic.messages.create (auto-instrumented)
    ├── model: claude-sonnet-4-20250514
    ├── input_tokens: 15
    └── output_tokens: 42
```

## Instrumentation Timing

Always initialize tracing before importing the framework:

```python
# CORRECT ORDER
import mantisdk.tracing as tracing
tracing.init(instrument_default=False)
tracing.instrument(names=["anthropic"])

import anthropic  # Import AFTER instrumentation
```

```python
# WRONG ORDER - instrumentation may not work
import anthropic  # Imported before instrumentation!

import mantisdk.tracing as tracing
tracing.init(instrument_default=False)
tracing.instrument(names=["anthropic"])
```

## Complete Example

```python
#!/usr/bin/env python3
"""Auto-instrumentation with Anthropic."""

from dotenv import load_dotenv
load_dotenv()

# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

exporters = []
if is_insight_configured():
    exporters.append(tracing.insight_exporter())

tracing.init(
    service_name="anthropic-agent",
    exporters=exporters if exporters else None,
    instrument_default=False,
)
tracing.instrument(names=["anthropic"])

# NOW import Anthropic
import anthropic

def main():
    client = anthropic.Anthropic()

    with tracing.trace("chat_session") as root:
        root.set_attribute(tracing.semconv.USER_ID, "user-123")

        # This call is auto-instrumented
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": "What is the capital of France?"}
            ]
        )

        answer = response.content[0].text
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, answer)
        print(answer)

if __name__ == "__main__":
    try:
        main()
    finally:
        tracing.shutdown()
```

## Next Steps

- [Integrations](/integrations) - Full framework examples
- [Insight Export](/tracing/insight-export) - View auto-traced data

---

## Context Tagging

*Source: /tracing/context-tagging*

# Context Tagging

Attach metadata -- like a call type, tags, or environment -- to every span created inside a code block, without touching each span individually.

## Why Context Tagging?

A typical agent makes many LLM calls: one to plan, one to act, one to judge quality. If every call is auto-instrumented, you need a way to tell Insight *which* call is which. `tracing_context()` lets you wrap a section of code so that all spans inside it carry the same labels.

## Tag All Spans in a Code Block

`tracing_context()` is a context manager. Any span created inside the `with` block automatically receives the metadata you pass in.

```python
import mantisdk.tracing as tracing

tracing.init()

with tracing.tracing_context(call_type="agent-call", tags=["planning", "v2"]):
    # Every LLM call inside this block is tagged as "agent-call"
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Plan the next step."}],
    )
```

In Insight, the spans from that block will show `call_type = agent-call` and `tags = ["planning", "v2"]`, making them easy to filter and compare.

### Parameters

| Parameter | Type | Description |
|-----------|------|-------------|
| `call_type` | `str` | Label for the kind of LLM call -- e.g. `"agent-call"`, `"judge-call"`, `"reflection-call"` |
| `tags` | `list[str]` | Freeform tags attached to every span in the block |
| `environment` | `str` | Environment name for filtering -- e.g. `"production"`, `"staging"` |
| `session_id` | `str` | Session identifier for grouping related spans |

All parameters are optional and keyword-only. Only the values you provide are set; the rest remain unchanged.

## Create Reusable Decorators with `call_type_decorator`

If you always tag certain functions the same way, `call_type_decorator()` creates a decorator you can reuse across your codebase.

```python
import mantisdk.tracing as tracing

tracing.init()

# Define once at module level
agent = tracing.call_type_decorator("agent-call")
judge = tracing.call_type_decorator("judge-call")

@agent
def run_agent(query: str) -> str:
    # All LLM spans inside are tagged as "agent-call"
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
    ).choices[0].message.content

@judge
def evaluate_response(response: str) -> float:
    # All LLM spans inside are tagged as "judge-call"
    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Rate this response: {response}"}],
    )
    return float(result.choices[0].message.content)
```

The decorator works with both sync and async functions -- no special handling needed.

```python
@judge
async def evaluate_response_async(response: str) -> float:
    result = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Rate this response: {response}"}],
    )
    return float(result.choices[0].message.content)
```

## Read the Current Context

Three getter functions let you inspect what context is active. This is useful for logging, assertions, or conditional logic inside shared utilities.

```python
import mantisdk.tracing as tracing

with tracing.tracing_context(call_type="agent-call", tags=["experiment-42"]):
    print(tracing.get_current_call_type())    # "agent-call"
    print(tracing.get_current_tags())          # ["experiment-42"]
    print(tracing.get_current_environment())   # None (not set in this block)
```

Outside any `tracing_context` block, all three return `None`.

| Function | Returns |
|----------|---------|
| `get_current_call_type()` | `str` or `None` |
| `get_current_tags()` | `list[str]` or `None` |
| `get_current_environment()` | `str` or `None` |

## Nest Contexts

Contexts nest cleanly. An inner block overrides the outer block's values for its duration; when it exits, the outer values are restored.

```python
import mantisdk.tracing as tracing

with tracing.tracing_context(call_type="agent-call", tags=["outer"]):
    # Spans here: call_type="agent-call", tags=["outer"]

    with tracing.tracing_context(call_type="judge-call", tags=["inner"]):
        # Spans here: call_type="judge-call", tags=["inner"]
        pass

    # Back to: call_type="agent-call", tags=["outer"]
```

You can also override only some fields. Unspecified fields keep the value from the outer context.

```python
with tracing.tracing_context(call_type="agent-call", tags=["experiment"]):
    # call_type="agent-call", tags=["experiment"]

    with tracing.tracing_context(call_type="judge-call"):
        # call_type="judge-call", tags still=["experiment"]
        pass
```

## Async and Thread Safety

Context tagging uses Python `contextvars`, which means:

- **Async**: Each `asyncio` task gets its own copy of the context. Tagging in one coroutine does not leak into another.
- **Threads**: Each thread has its own context by default. Tagging in one thread does not affect another.

You can safely use `tracing_context()` and `call_type_decorator()` in concurrent code without locking.

```python
import asyncio
import mantisdk.tracing as tracing

async def run_agent(query: str):
    with tracing.tracing_context(call_type="agent-call"):
        return await async_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}],
        )

async def run_judge(response: str):
    with tracing.tracing_context(call_type="judge-call"):
        return await async_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": f"Rate: {response}"}],
        )

# These run concurrently -- each task keeps its own context
agent_result, judge_result = await asyncio.gather(
    run_agent("Plan the next step."),
    run_judge("The agent completed the task."),
)
```

## Session Tracking

Sessions group related traces together in Insight. A chatbot conversation, a multi-step workflow, or a series of agent interactions that belong to the same logical unit can all share a session ID. In the Insight UI, you can filter and browse traces by session to see the full picture.

### Tag All Traces in a Session

Pass `session_id` to `tracing_context()` and every trace created inside the block is linked to that session.

```python
import mantisdk.tracing as tracing

tracing.init()

session_id = "chat-session-abc123"

with tracing.tracing_context(session_id=session_id):
    # Each trace in this block belongs to the same session
    with tracing.trace("user-message-1", input="What is Mantis?") as span:
        response = process("What is Mantis?")
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)

    with tracing.trace("user-message-2", input="How do I install it?") as span:
        response = process("How do I install it?")
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
```

Both traces appear under session `chat-session-abc123` in Insight.

### Set Session on Individual Spans

For more granular control, set `SESSION_ID` directly on a span:

```python
with tracing.trace("workflow", input=query) as span:
    span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
    span.set_attribute(tracing.semconv.USER_ID, "user-123")
    result = do_work(query)
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

### Combine Sessions with Other Context

`session_id` works alongside `call_type`, `tags`, and `environment`. Mix them freely:

```python
with tracing.tracing_context(
    session_id="onboarding-session-001",
    environment="production",
    tags=["onboarding", "v2"],
):
    with tracing.tracing_context(call_type="agent-call"):
        response = agent.run("Help me set up my account")

    with tracing.tracing_context(call_type="judge-call"):
        score = judge.evaluate(response)
```

### Chatbot Example

A typical pattern for a chatbot that maintains session context across multiple user messages:

```python
import uuid
import mantisdk.tracing as tracing

tracing.init()

def handle_conversation(user_id: str):
    session_id = f"chat-{uuid.uuid4().hex[:12]}"

    with tracing.tracing_context(session_id=session_id):
        while True:
            user_input = input("> ")
            if user_input in ("quit", "exit"):
                break

            with tracing.trace("chat-turn", input=user_input) as span:
                span.set_attribute(tracing.semconv.USER_ID, user_id)
                response = agent.respond(user_input)
                span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
                print(response)

    tracing.shutdown()
```

Every turn in the conversation is a separate trace, but they all share the same session ID. In Insight, open the session to see the full conversation flow.

## Complete Example

Putting it all together -- an agent loop that tags planning, execution, and evaluation separately.

```python
import mantisdk.tracing as tracing

tracing.init(service_name="my-agent")

agent = tracing.call_type_decorator("agent-call")
judge = tracing.call_type_decorator("judge-call")

@agent
def plan(goal: str) -> str:
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Plan how to: {goal}"}],
    ).choices[0].message.content

@agent
def execute(plan: str) -> str:
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Execute this plan: {plan}"}],
    ).choices[0].message.content

@judge
def score(result: str) -> float:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Score 0-1: {result}"}],
    )
    return float(response.choices[0].message.content)

def run(goal: str):
    with tracing.tracing_context(tags=["experiment-42"], environment="staging"):
        with tracing.trace("agent-loop", input=goal):
            the_plan = plan(goal)
            result = execute(the_plan)
            quality = score(result)
            return {"result": result, "score": quality}

run("Summarize today's news")
tracing.shutdown()
```

In Insight you can now filter by `call_type` to see agent calls vs judge calls, by `tags` to find a specific experiment, and by `environment` to separate staging from production.

## Next Steps

- [Traces & Spans](/tracing/traces-and-spans) -- Create and nest spans manually
- [Semantic Conventions](/tracing/semantic-conventions) -- Standard attribute names
- [Async Support](/tracing/async-support) -- Async-specific span creation

---

## Initialization

*Source: /tracing/initialization*

# Initialization

Configure and initialize Mantis SDK tracing.

## Basic Initialization

The simplest way to initialize tracing:

```python
import mantisdk.tracing as tracing

# Auto-detect Insight credentials from environment variables
tracing.init()
```

This will:
1. Create an OpenTelemetry TracerProvider
2. Auto-detect Insight credentials from `INSIGHT_*` environment variables
3. Set up a BatchSpanProcessor for efficient export

## Configuration Options

### `tracing.init()`

```python
tracing.init(
    service_name="my-agent",        # Service name in traces
    trace_name="my-workflow",       # Name for the trace (shown in Insight)
    environment="production",       # Environment for filtering (production, staging, dev)
    version="1.2.3",                # App version for regression tracking
    # Insight credentials (alternative to env vars)
    insight_host="https://insight.withmetis.ai",
    insight_public_key="pk-lf-...",
    insight_secret_key="sk-lf-...",
    insight_endpoint=None,          # Custom OTLP endpoint (optional)
    instrument_default=True,        # Auto-instrument common libraries
    trace_config=None,              # OpenInference TraceConfig (optional)
    integrations=None,              # Fine-grained integration control (optional)
    force=False,                    # Replace existing TracerProvider
    debug="INFO",                   # Console logging level (False, True, or level string)
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `service_name` | `str` | `"mantisdk-tracing"` | Name of your service |
| `trace_name` | `str` | `None` | Name for the trace, shown in Insight UI |
| `environment` | `str` | `None` | Environment name (production, staging, dev) for filtering in Insight |
| `version` | `str` | `None` | App version (e.g., "1.2.3", git SHA) for regression tracking |
| `insight_host` | `str` | `None` | Insight server URL. Overrides `INSIGHT_HOST` env var |
| `insight_public_key` | `str` | `None` | Insight public key. Overrides `INSIGHT_PUBLIC_KEY` env var |
| `insight_secret_key` | `str` | `None` | Insight secret key. Overrides `INSIGHT_SECRET_KEY` env var |
| `insight_endpoint` | `str` | `None` | Optional OTLP endpoint. Overrides `INSIGHT_OTLP_ENDPOINT` env var |
| `instrument_default` | `bool` | `True` | Enable auto-instrumentation |
| `exporters` | `list[SpanExporter]` | Auto-detect | Custom exporters |
| `trace_config` | `TraceConfig` | `None` | OpenInference TraceConfig for controlling capture behavior |
| `integrations` | `dict[str, Any]` | `None` | Fine-grained control over individual integrations |
| `force` | `bool` | `False` | Replace existing provider |
| `debug` | `bool \| str` | `False` | Console logging level. See [Debug Logging](#debug-logging) |

### Provider Reuse Behavior

- If no TracerProvider exists, creates one
- If a TracerProvider exists and `force=False`, reuses it and adds processors
- If `force=True`, replaces the existing TracerProvider

## Debug Logging

The `debug` parameter controls how much tracing activity the SDK prints to the console. Use it to verify that tracing is working, diagnose export failures, or inspect individual spans.

```python
import mantisdk.tracing as tracing

# Recommended: see init summary and export results
tracing.init(service_name="my-agent", debug="INFO")

# Minimal: only see problems (export failures, missing credentials)
tracing.init(service_name="my-agent", debug="WARNING")

# Full firehose: every span start/end, all internal plumbing
tracing.init(service_name="my-agent", debug=True)

# Silent (default): no console output
tracing.init(service_name="my-agent", debug=False)
```

### Log Levels

| Level | What you see |
|-------|-------------|
| `False` (default) | Nothing. Silent operation. |
| `"ERROR"` | Only export failures and exceptions. |
| `"WARNING"` | Export failures and missing exporter warnings. |
| `"INFO"` | **Recommended.** Init summary, export success/failure with timing, instrumentor activation, shutdown summary. |
| `"DEBUG"` or `True` | Everything above plus every span start/end, processor setup, and internal plumbing. |

### Example Output

**`debug="INFO"`** -- what most users want:

```
[mantisdk.tracing] Created new TracerProvider with service_name=my-agent
[mantisdk.tracing] Auto-configured Insight exporter from environment variables
[mantisdk.tracing] Instrumented: openai
[mantisdk.tracing] MantisDK tracing initialized successfully
[mantisdk.tracing] Export: 4 span(s) -> SUCCESS (https://insight.withmetis.ai, 85ms)
[mantisdk.tracing] Shutdown: exported 47 span(s) total
```

**`debug=True`** -- full firehose for debugging:

```
[mantisdk.tracing] Created new TracerProvider with service_name=my-agent
[mantisdk.tracing] Added TracingContextSpanProcessor
[mantisdk.tracing] Added LoggingSpanProcessor
[mantisdk.tracing] Auto-configured Insight exporter from environment variables
[mantisdk.tracing] Instrumented: openai
[mantisdk.tracing] MantisDK tracing initialized successfully
[mantisdk.tracing] Span started: "my-workflow"
[mantisdk.tracing] Span started: "ChatCompletion" (auto: openai)
[mantisdk.tracing] Span ended:   "ChatCompletion" (0.61s) trace_id=abc123...
[mantisdk.tracing] Span ended:   "my-workflow" (1.32s) trace_id=abc123...
[mantisdk.tracing] Export: 2 span(s) -> SUCCESS (https://insight.withmetis.ai, 85ms)
[mantisdk.tracing] Flush completed successfully
[mantisdk.tracing] Shutdown: exported 2 span(s) total
```

## Pass Insight Credentials Directly

Instead of relying on environment variables, you can pass Insight credentials straight to `init()`. This is useful in notebooks, one-off scripts, or when you pull secrets from a vault at runtime.

```python
import os
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    insight_host=os.environ.get("INSIGHT_HOST"),
    insight_public_key=os.environ.get("INSIGHT_PUBLIC_KEY"),
    insight_secret_key=os.environ.get("INSIGHT_SECRET_KEY"),
)
```

## Control Capture Behavior with `trace_config`

Pass an OpenInference `TraceConfig` to control what the SDK captures. This is helpful when you need to hide sensitive inputs or outputs.

```python
from openinference.instrumentation import TraceConfig
import mantisdk.tracing as tracing

config = TraceConfig(hide_inputs=True, hide_outputs=False)

tracing.init(
    service_name="my-agent",
    trace_config=config,
    instrument_default=True,
)
```

## Fine-Grained Integration Control with `integrations`

The `integrations` parameter lets you pass configuration to individual auto-instrumentors. Keys are instrumentor names (e.g., `"openai"`, `"anthropic"`), values are dictionaries of options forwarded to that instrumentor.

```python
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    instrument_default=True,
    integrations={
        "openai": {"record_content": False},
        "anthropic": {"record_content": True},
    },
)
```

## Environment Variables

### Insight Configuration

```bash
# Required for Insight export
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# Optional: Override OTLP endpoint
INSIGHT_OTLP_ENDPOINT=https://custom.endpoint/v1/traces
```

### Checking Configuration

```python
from mantisdk.tracing.exporters.insight import is_insight_configured

if is_insight_configured():
    print("Insight credentials found")
else:
    print("Set INSIGHT_HOST, INSIGHT_PUBLIC_KEY, INSIGHT_SECRET_KEY")
```

## Custom Exporters

### Using Insight Exporter Explicitly

```python
import mantisdk.tracing as tracing

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

### Custom OTLP Exporter

```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    exporters=[
        OTLPSpanExporter(endpoint="http://localhost:4317"),
    ]
)
```

## Lifecycle Management

### Shutdown

Always call `shutdown()` before your program exits to ensure all spans are exported:

```python
import atexit
import mantisdk.tracing as tracing

tracing.init()

# Option 1: Register with atexit
atexit.register(tracing.shutdown)

# Option 2: Manual try/finally
try:
    main()
finally:
    tracing.shutdown()
```

### Flush

Force flush pending spans without shutting down:

```python
# Process a batch
for item in batch:
    process(item)

# Ensure spans are exported before continuing
tracing.flush()

# Continue with more work...
```

### Shutdown vs Flush

| Function | Behavior |
|----------|----------|
| `shutdown()` | Flushes spans, closes processors, releases resources. Logs total export count. |
| `flush()` | Flushes spans, keeps provider running |

## Complete Example

```python
#!/usr/bin/env python3
"""Complete initialization example."""

import atexit
from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    environment="production",
    version="1.0.0",
    instrument_default=True,
    debug="INFO",
)
atexit.register(tracing.shutdown)

with tracing.trace("test_trace", input="hello"):
    print("Tracing is working!")
```

## Next Steps

- [Traces & Spans](/tracing/traces-and-spans) - Create spans in your code
- [Auto-Instrumentation](/tracing/auto-instrumentation) - Enable framework support
- [Insight Export](/tracing/insight-export) - Configure Insight backend

---

## Insight Export

*Source: /tracing/insight-export*

# Insight Export

Export traces to Mantis Insight for visualization and analysis.

## Overview

Mantis Insight is the observability backend for Mantis SDK. It provides:
- Trace visualization with span hierarchy
- LLM metrics (tokens, latency, cost)
- User and session tracking
- Search and filtering

## OTLP export vs direct ingestion helper

Use both paths based on where your telemetry is produced:

- **OTLP export (`tracing.init()` + exporter):** Best for automatic span export from instrumented application runtime.
- **Direct ingestion helper (`InsightLangfuseCompatClient`):** Best for explicit trace/score submission from external pipelines or post-processing systems.

For the direct helper flow, see [Insight Ingestion (SDK)](/integrations/insight-ingestion).

## Configuration

### Environment Variables

Set these environment variables to enable Insight export:

```bash
# Required
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-public-key
INSIGHT_SECRET_KEY=sk-lf-your-secret-key

# Optional: Custom OTLP endpoint
INSIGHT_OTLP_ENDPOINT=https://custom.endpoint/v1/traces
```

### Auto-Detection

When you call `tracing.init()`, the SDK automatically detects Insight credentials:

```python
import mantisdk.tracing as tracing

# Auto-detects INSIGHT_* environment variables
tracing.init()
```

### Explicit Configuration

Create the exporter explicitly:

```python
import mantisdk.tracing as tracing

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

## Checking Configuration

```python
from mantisdk.tracing.exporters.insight import is_insight_configured

if is_insight_configured():
    print("Insight credentials found in environment")
else:
    print("Missing INSIGHT_HOST, INSIGHT_PUBLIC_KEY, or INSIGHT_SECRET_KEY")
```

## `insight_exporter()` Function

### Parameters

```python
tracing.insight_exporter(
    host=None,           # Insight host URL (or INSIGHT_HOST env var)
    public_key=None,     # Public key (or INSIGHT_PUBLIC_KEY env var)
    secret_key=None,     # Secret key (or INSIGHT_SECRET_KEY env var)
    otlp_endpoint=None,  # Custom OTLP endpoint (optional)
)
```

### Returns

An OpenTelemetry `SpanExporter` configured for Insight's OTLP endpoint.

## Send OTLP Traces Without MantisDK

If you already have OpenTelemetry tracing set up (e.g., via the OTel Python SDK, JS SDK, or any OTLP-compatible framework), you can send traces directly to Insight without installing MantisDK. All you need is the endpoint and auth headers.

### Endpoint

```
POST https://insight.withmetis.ai/api/public/otel/v1/traces
```

### Authentication

Insight uses HTTP Basic Auth with your project's API keys:

| Header | Value |
|--------|-------|
| `Authorization` | `Basic base64(publicKey:secretKey)` |
| `Content-Type` | `application/json` or `application/x-protobuf` |

You can find your public key (`pk-lf-...`) and secret key (`sk-lf-...`) in your Insight project settings under **API Keys**.

### Python (OTel SDK)

```python
import base64
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Your existing TracerProvider
provider = TracerProvider()

# Add Insight as an exporter
public_key = "pk-lf-..."
secret_key = "sk-lf-..."
credentials = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()

insight_exporter = OTLPSpanExporter(
    endpoint="https://insight.withmetis.ai/api/public/otel/v1/traces",
    headers={"Authorization": f"Basic {credentials}"},
)
provider.add_span_processor(BatchSpanProcessor(insight_exporter))
```

### JavaScript / TypeScript (OTel JS SDK)

```typescript

const publicKey = "pk-lf-...";
const secretKey = "sk-lf-...";
const credentials = Buffer.from(`${publicKey}:${secretKey}`).toString("base64");

const insightExporter = new OTLPTraceExporter({
  url: "https://insight.withmetis.ai/api/public/otel/v1/traces",
  headers: { Authorization: `Basic ${credentials}` },
});

// Add to your existing provider
provider.addSpanProcessor(new BatchSpanProcessor(insightExporter));
```

### Environment Variables (Any OTel SDK)

Most OTel SDKs support standard environment variables. Note that `OTEL_EXPORTER_OTLP_ENDPOINT` should be the base URL — the SDK appends `/v1/traces` automatically:

```bash
OTEL_EXPORTER_OTLP_ENDPOINT=https://insight.withmetis.ai/api/public/otel
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic <base64(pk-lf-...:sk-lf-...)>"
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
```

### Dual-Write to Multiple Backends

If you're already exporting to another backend (e.g., Datadog, Grafana Tempo, Jaeger), just add a second `BatchSpanProcessor` for Insight. Each processor exports independently — a failure in one does not affect the other.

```python
# Existing exporter
provider.add_span_processor(BatchSpanProcessor(your_existing_exporter))

# Add Insight alongside it
provider.add_span_processor(BatchSpanProcessor(insight_exporter))
```

## Multiple Exporters

The `exporters` parameter accepts a list, so you can send traces to multiple backends simultaneously. Each exporter gets its own `BatchSpanProcessor` — they export independently and won't block each other.

### Insight + Another OTEL Backend

Send traces to both Insight and a second OpenTelemetry-compatible backend (e.g., Jaeger, Grafana Tempo, Datadog, Honeycomb):

```python
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    exporters=[
        tracing.insight_exporter(),                                      # Insight
        OTLPSpanExporter(endpoint="https://tempo.internal/v1/traces"),   # Second backend
    ]
)
```

For backends that use gRPC instead of HTTP:

```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    exporters=[
        tracing.insight_exporter(),
        OTLPSpanExporter(endpoint="localhost:4317"),  # gRPC endpoint
    ]
)
```

### Insight + Another OTLP Backend

Add your existing OTLP collector as a second exporter:

```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    service_name="my-agent",
    exporters=[
        tracing.insight_exporter(),                   # Insight (LLM observability)
        OTLPSpanExporter(                             # Your existing OTel backend
            endpoint="otel-collector.example.com:4317",
            insecure=False,
        ),
    ]
)
```

Both exporters run independently with their own `BatchSpanProcessor`, so a failure in one does not affect the other.

### Insight + Console (Debugging)

```python
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    exporters=[
        tracing.insight_exporter(),  # Insight
        ConsoleSpanExporter(),       # Console for debugging
    ]
)
```

## Data Sent to Insight

### Trace-Level Data

- Trace ID and name
- Input and output (via semantic conventions)
- User ID and session ID
- Tags and metadata
- Environment and version

### Span-Level Data

- Span name and kind
- Parent-child relationships
- Start/end timestamps
- Attributes (including semantic conventions)
- Events and exceptions

### LLM-Specific Data (via auto-instrumentation)

- Model name
- Input/output token counts
- Latency
- Tool calls and results

## Viewing Traces in Insight

After traces are exported:

1. Log in to [Mantis Insight](https://insight.withmetis.ai)
2. Navigate to your project
3. View traces in the Traces tab
4. Click a trace to see the span hierarchy
5. Use filters to find specific traces:
   - By user ID
   - By session ID
   - By time range
   - By tags

## Ensuring Export

### Flush Before Exit

Always ensure spans are exported before your program exits:

```python
import atexit
import mantisdk.tracing as tracing

tracing.init()
atexit.register(tracing.shutdown)

# Or use try/finally
try:
    main()
finally:
    tracing.shutdown()
```

### Manual Flush

Force export without shutting down:

```python
# Process batch
for item in batch:
    process_item(item)

# Ensure batch spans are exported
tracing.flush()

# Continue processing...
```

## Troubleshooting

### Traces Not Appearing

1. **Check credentials**: Verify `INSIGHT_HOST`, `INSIGHT_PUBLIC_KEY`, `INSIGHT_SECRET_KEY`
2. **Check network**: Ensure your application can reach the Insight host
3. **Check shutdown**: Ensure `tracing.shutdown()` is called
4. **Enable debug**: Add console exporter to see spans locally

### Debug with Console Exporter

```python
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
import mantisdk.tracing as tracing

tracing.init(
    exporters=[
        tracing.insight_exporter(),
        ConsoleSpanExporter(),  # See spans in terminal
    ]
)
```

### Check Configuration

```python
from mantisdk.tracing.exporters.insight import is_insight_configured
import os

print(f"INSIGHT_HOST: {os.environ.get('INSIGHT_HOST', 'NOT SET')}")
print(f"INSIGHT_PUBLIC_KEY: {os.environ.get('INSIGHT_PUBLIC_KEY', 'NOT SET')[:10]}...")
print(f"INSIGHT_SECRET_KEY: {'SET' if os.environ.get('INSIGHT_SECRET_KEY') else 'NOT SET'}")
print(f"Is configured: {is_insight_configured()}")
```

## Complete Example

```python
#!/usr/bin/env python3
"""Complete Insight export example."""

import atexit
from dotenv import load_dotenv
load_dotenv()

import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured

def setup_tracing():
    """Configure tracing with Insight export."""
    if not is_insight_configured():
        print("Warning: Insight not configured")
        print("Set INSIGHT_HOST, INSIGHT_PUBLIC_KEY, INSIGHT_SECRET_KEY")
        return False

    tracing.init(
        service_name="my-agent",
        exporters=[tracing.insight_exporter()],
    )

    atexit.register(tracing.shutdown)
    print("Tracing configured with Insight export")
    return True

def main():
    with tracing.trace("main_workflow", input="example") as span:
        span.set_attribute(tracing.semconv.USER_ID, "user-123")
        span.set_attribute(tracing.semconv.SESSION_ID, "session-456")

        with tracing.span("step_1"):
            print("Step 1")

        with tracing.span("step_2"):
            print("Step 2")

        span.set_attribute(tracing.semconv.TRACE_OUTPUT, "completed")

if __name__ == "__main__":
    if setup_tracing():
        main()
        print("Traces exported to Insight!")
```

## Next Steps

- [Insight REST API](/reference/insight-rest-api) - Fetch traces back from Insight via the REST API
- [Quickstart](/quickstart) - Get started with tracing
- [Integrations](/integrations) - Framework-specific examples

---

## Semantic Conventions

*Source: /tracing/semantic-conventions*

# Semantic Conventions

Mantis SDK provides semantic conventions for consistent attribute naming across traces.

## Overview

Semantic conventions are standardized attribute names that ensure consistent data across your traces. Use `tracing.semconv` to access all available conventions.

```python
import mantisdk.tracing as tracing

with tracing.trace("workflow", input=query) as span:
    span.set_attribute(tracing.semconv.USER_ID, "user-123")
    span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
```

Constants are organized by namespace:

| Namespace | Prefix | Purpose |
|-----------|--------|---------|
| Insight | `insight.*` | Primary attributes for the Insight UI |
| Langfuse compat | `langfuse.*` | Compatibility with Langfuse SDK spans |
| GenAI | `gen_ai.*` | OpenTelemetry GenAI semantic conventions |
| OpenInference | `openinference.*`, `llm.*`, `input.*`, `output.*`, `tool.*` | Arize OpenInference conventions |
| MantisDK | `mantis.*` | Custom attributes added by MantisDK |
| Claude | `claude.*` | Claude Agent SDK-specific attributes |

For most use cases, the Insight namespace is all you need. The other namespaces are populated automatically by auto-instrumentation or are useful when integrating with third-party tooling.

---

## Insight Namespace

### Trace-Level Attributes

For root spans (traces):

| Constant | Value | Description |
|----------|-------|-------------|
| `TRACE_NAME` | `insight.trace.name` | Name of the trace |
| `TRACE_INPUT` | `insight.trace.input` | Input to the trace |
| `TRACE_OUTPUT` | `insight.trace.output` | Output from the trace |
| `TRACE_METADATA` | `insight.trace.metadata` | Additional metadata (JSON) |
| `TRACE_TAGS` | `insight.trace.tags` | Tags for filtering |
| `TRACE_PUBLIC` | `insight.trace.public` | Whether trace is public |

```python
with tracing.trace("my-workflow", input=query) as span:
    # These are for root traces
    span.set_attribute(tracing.semconv.TRACE_INPUT, query)
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
    span.set_attribute(tracing.semconv.TRACE_METADATA, '{"source": "api"}')
    span.set_attribute(tracing.semconv.TRACE_TAGS, '["production", "v2"]')
```

### User & Session

For tracking users and sessions:

| Constant | Value | Description |
|----------|-------|-------------|
| `USER_ID` | `insight.user.id` | Unique user identifier |
| `SESSION_ID` | `insight.session.id` | Session identifier |

```python
with tracing.trace("workflow") as span:
    span.set_attribute(tracing.semconv.USER_ID, "user-123")
    span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
```

### Run Tracking

For linking traces to Mantis Runs (experiments):

| Constant | Value | Description |
|----------|-------|-------------|
| `RUN_ID` | `insight.run.id` | Run/experiment identifier |
| `RUN_NAME` | `insight.run.name` | Run name |

These attributes are automatically set when traces are created within a `mantisdk.run` context:

```python
from mantisdk import run

with run("my-experiment") as r:
    # All traces created here automatically have RUN_ID set
    with tracing.trace("workflow") as span:
        # RUN_ID is injected automatically
        pass
```

### Environment & Version

For deployment context:

| Constant | Value | Description |
|----------|-------|-------------|
| `ENVIRONMENT` | `insight.environment` | Deployment environment |
| `RELEASE` | `insight.release` | Release identifier |
| `VERSION` | `insight.version` | Version string |

```python
tracing.init()
with tracing.trace("workflow") as span:
    span.set_attribute(tracing.semconv.ENVIRONMENT, "production")
    span.set_attribute(tracing.semconv.RELEASE, "v1.2.3")
    span.set_attribute(tracing.semconv.VERSION, "1.2.3")
```

### Observation-Level Attributes

For child spans (observations):

| Constant | Value | Description |
|----------|-------|-------------|
| `OBSERVATION_TYPE` | `insight.observation.type` | Type of observation |
| `OBSERVATION_INPUT` | `insight.observation.input` | Input to the operation |
| `OBSERVATION_OUTPUT` | `insight.observation.output` | Output from the operation |
| `OBSERVATION_METADATA` | `insight.observation.metadata` | Additional metadata |
| `OBSERVATION_LEVEL` | `insight.observation.level` | Log level |
| `OBSERVATION_STATUS_MESSAGE` | `insight.observation.status_message` | Status message |

```python
with tracing.span("process_data", input=data) as span:
    result = process(data)
    span.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(result))
    span.set_attribute(tracing.semconv.OBSERVATION_METADATA, '{"items": 5}')
```

### Model & Generation

For LLM-specific attributes:

| Constant | Value | Description |
|----------|-------|-------------|
| `MODEL_NAME` | `insight.observation.model.name` | Model identifier |
| `MODEL_PARAMETERS` | `insight.observation.model.parameters` | Model parameters (JSON) |

```python
with tracing.span("llm_call") as span:
    span.set_attribute(tracing.semconv.MODEL_NAME, "gpt-4o-mini")
    span.set_attribute(tracing.semconv.MODEL_PARAMETERS, '{"temperature": 0.7}')
```

### Observation Types

Use `ObservationType` for the `OBSERVATION_TYPE` attribute:

```python
from mantisdk.tracing.semconv import ObservationType

with tracing.span("my_span") as span:
    span.set_attribute(
        tracing.semconv.OBSERVATION_TYPE,
        ObservationType.SPAN       # or "span"
    )

# Available types:
# ObservationType.SPAN       = "span"
# ObservationType.GENERATION = "generation"
# ObservationType.EVENT      = "event"
```

### Observation Levels

Use `ObservationLevel` for log-level-like severity:

```python
from mantisdk.tracing.semconv import ObservationLevel

with tracing.span("warning_case") as span:
    span.set_attribute(
        tracing.semconv.OBSERVATION_LEVEL,
        ObservationLevel.WARNING   # or "WARNING"
    )

# Available levels:
# ObservationLevel.DEBUG   = "DEBUG"
# ObservationLevel.DEFAULT = "DEFAULT"
# ObservationLevel.WARNING = "WARNING"
# ObservationLevel.ERROR   = "ERROR"
```

---

## Langfuse Compatibility

These constants let you emit attributes that Langfuse SDK consumers already expect. If you are migrating from the Langfuse Python SDK, or need traces that are readable by both Insight and Langfuse, use these alongside (or instead of) the Insight equivalents.

### Trace Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `LANGFUSE_TRACE_NAME` | `langfuse.trace.name` | Trace name |
| `LANGFUSE_TRACE_USER_ID` | `user.id` | User identifier |
| `LANGFUSE_TRACE_SESSION_ID` | `session.id` | Session identifier |
| `LANGFUSE_TRACE_TAGS` | `langfuse.trace.tags` | Tags for filtering |
| `LANGFUSE_TRACE_PUBLIC` | `langfuse.trace.public` | Whether trace is public |
| `LANGFUSE_TRACE_METADATA` | `langfuse.trace.metadata` | Additional metadata |
| `LANGFUSE_TRACE_INPUT` | `langfuse.trace.input` | Trace input |
| `LANGFUSE_TRACE_OUTPUT` | `langfuse.trace.output` | Trace output |

### Observation Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `LANGFUSE_OBSERVATION_TYPE` | `langfuse.observation.type` | Observation type |
| `LANGFUSE_OBSERVATION_METADATA` | `langfuse.observation.metadata` | Observation metadata |
| `LANGFUSE_OBSERVATION_LEVEL` | `langfuse.observation.level` | Observation log level |
| `LANGFUSE_OBSERVATION_STATUS_MESSAGE` | `langfuse.observation.status_message` | Status message |
| `LANGFUSE_OBSERVATION_INPUT` | `langfuse.observation.input` | Observation input |
| `LANGFUSE_OBSERVATION_OUTPUT` | `langfuse.observation.output` | Observation output |

### Generation Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `LANGFUSE_OBSERVATION_COMPLETION_START_TIME` | `langfuse.observation.completion_start_time` | Time-to-first-token |
| `LANGFUSE_OBSERVATION_MODEL` | `langfuse.observation.model.name` | Model name |
| `LANGFUSE_OBSERVATION_MODEL_PARAMETERS` | `langfuse.observation.model.parameters` | Model parameters |
| `LANGFUSE_OBSERVATION_USAGE_DETAILS` | `langfuse.observation.usage_details` | Token usage details |
| `LANGFUSE_OBSERVATION_COST_DETAILS` | `langfuse.observation.cost_details` | Cost details |
| `LANGFUSE_OBSERVATION_PROMPT_NAME` | `langfuse.observation.prompt.name` | Prompt template name |
| `LANGFUSE_OBSERVATION_PROMPT_VERSION` | `langfuse.observation.prompt.version` | Prompt template version |

### Environment

| Constant | Value | Description |
|----------|-------|-------------|
| `LANGFUSE_ENVIRONMENT` | `langfuse.environment` | Deployment environment |
| `LANGFUSE_RELEASE` | `langfuse.release` | Release identifier |
| `LANGFUSE_VERSION` | `langfuse.version` | Version string |

---

## OpenTelemetry GenAI

Standard [OpenTelemetry GenAI semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/). These are set automatically by OpenTelemetry-based instrumentors and can be read by any OTel-compatible backend.

### Request & Response

| Constant | Value | Description |
|----------|-------|-------------|
| `GEN_AI_OPERATION_NAME` | `gen_ai.operation.name` | Operation name (e.g. "chat") |
| `GEN_AI_SYSTEM` | `gen_ai.system` | AI system (e.g. "openai") |
| `GEN_AI_REQUEST_MODEL` | `gen_ai.request.model` | Requested model name |
| `GEN_AI_RESPONSE_MODEL` | `gen_ai.response.model` | Model that actually responded |
| `GEN_AI_REQUEST_TEMPERATURE` | `gen_ai.request.temperature` | Sampling temperature |
| `GEN_AI_REQUEST_MAX_TOKENS` | `gen_ai.request.max_tokens` | Maximum tokens requested |
| `GEN_AI_RESPONSE_FINISH_REASONS` | `gen_ai.response.finish_reasons` | Why the model stopped |
| `GEN_AI_CONVERSATION_ID` | `gen_ai.conversation.id` | Conversation identifier |

### Messages

| Constant | Value | Description |
|----------|-------|-------------|
| `GEN_AI_INPUT_MESSAGES` | `gen_ai.input.messages` | Input messages (JSON) |
| `GEN_AI_OUTPUT_MESSAGES` | `gen_ai.output.messages` | Output messages (JSON) |

### Usage

| Constant | Value | Description |
|----------|-------|-------------|
| `GEN_AI_USAGE_INPUT_TOKENS` | `gen_ai.usage.input_tokens` | Input token count |
| `GEN_AI_USAGE_OUTPUT_TOKENS` | `gen_ai.usage.output_tokens` | Output token count |
| `GEN_AI_USAGE_COST` | `gen_ai.usage.cost` | Estimated cost |

### Tool Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `GEN_AI_TOOL_NAME` | `gen_ai.tool.name` | Tool name |
| `GEN_AI_TOOL_CALL_ARGUMENTS` | `gen_ai.tool.call.arguments` | Tool call arguments (JSON) |
| `GEN_AI_TOOL_CALL_RESULT` | `gen_ai.tool.call.result` | Tool call result |

---

## OpenInference

[OpenInference](https://github.com/Arize-ai/openinference) conventions from Arize AI. These are populated by OpenInference instrumentors (e.g. `openinference-instrumentation-openai`).

### Span Kind

| Constant | Value | Description |
|----------|-------|-------------|
| `OPENINFERENCE_SPAN_KIND` | `openinference.span.kind` | Kind of span (see values below) |

Span kind values are available as top-level constants:

| Constant | Value |
|----------|-------|
| `SPAN_KIND_LLM` | `LLM` |
| `SPAN_KIND_TOOL` | `TOOL` |
| `SPAN_KIND_AGENT` | `AGENT` |
| `SPAN_KIND_CHAIN` | `CHAIN` |
| `SPAN_KIND_EMBEDDING` | `EMBEDDING` |
| `SPAN_KIND_RETRIEVER` | `RETRIEVER` |
| `SPAN_KIND_RERANKER` | `RERANKER` |
| `SPAN_KIND_GUARDRAIL` | `GUARDRAIL` |
| `SPAN_KIND_EVALUATOR` | `EVALUATOR` |

```python
from mantisdk.tracing import semconv

span.set_attribute(semconv.OPENINFERENCE_SPAN_KIND, semconv.SPAN_KIND_AGENT)
```

### LLM Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `LLM_MODEL_NAME` | `llm.model_name` | Model name |
| `LLM_INVOCATION_PARAMETERS` | `llm.invocation_parameters` | Parameters as JSON |
| `LLM_INPUT_MESSAGES` | `llm.input_messages` | Input messages |
| `LLM_OUTPUT_MESSAGES` | `llm.output_messages` | Output messages |
| `LLM_TOKEN_COUNT_PROMPT` | `llm.token_count.prompt` | Prompt token count |
| `LLM_TOKEN_COUNT_COMPLETION` | `llm.token_count.completion` | Completion token count |
| `LLM_TOKEN_COUNT_TOTAL` | `llm.token_count.total` | Total token count |
| `LLM_SYSTEM` | `llm.system` | LLM provider system |
| `LLM_PROVIDER` | `llm.provider` | LLM provider name |

### Input / Output

| Constant | Value | Description |
|----------|-------|-------------|
| `INPUT_VALUE` | `input.value` | General input value |
| `OUTPUT_VALUE` | `output.value` | General output value |
| `INPUT_MIME_TYPE` | `input.mime_type` | MIME type of the input |
| `OUTPUT_MIME_TYPE` | `output.mime_type` | MIME type of the output |

MIME type constants for convenience:

| Constant | Value |
|----------|-------|
| `MIME_TYPE_TEXT` | `text/plain` |
| `MIME_TYPE_JSON` | `application/json` |

### Tool Attributes

| Constant | Value | Description |
|----------|-------|-------------|
| `TOOL_NAME` | `tool.name` | Tool name |
| `TOOL_DESCRIPTION` | `tool.description` | Tool description |
| `TOOL_PARAMETERS` | `tool.parameters` | Tool parameters (JSON) |

---

## MantisDK Custom

Attributes specific to MantisDK. These are set automatically by SDK instrumentation and capture data not covered by the standard namespaces.

| Constant | Value | Description |
|----------|-------|-------------|
| `MANTIS_LLM_THINKING` | `mantis.llm.thinking` | Model thinking/reasoning content |
| `MANTIS_LLM_COST_USD` | `mantis.llm.cost_usd` | LLM call cost in USD |
| `MANTIS_DURATION_MS` | `mantis.duration_ms` | Total duration in milliseconds |
| `MANTIS_DURATION_API_MS` | `mantis.duration_api_ms` | API call duration in milliseconds |
| `MANTIS_NUM_TURNS` | `mantis.num_turns` | Number of conversational turns |
| `MANTIS_TOOL_IS_ERROR` | `mantis.tool.is_error` | Whether a tool call errored |

```python
from mantisdk.tracing import semconv

with tracing.span("llm_call") as span:
    span.set_attribute(semconv.MANTIS_LLM_COST_USD, 0.0032)
    span.set_attribute(semconv.MANTIS_DURATION_API_MS, 1250)
```

---

## Claude-Specific

Attributes for traces produced by the Claude Agent SDK. These are populated automatically when using Claude as the underlying model.

| Constant | Value | Description |
|----------|-------|-------------|
| `CLAUDE_API_ERROR` | `claude.api_error` | API error from Claude |
| `CLAUDE_PARENT_TOOL_USE_ID` | `claude.parent_tool_use_id` | Parent tool-use ID |
| `CLAUDE_STRUCTURED_OUTPUT` | `claude.structured_output` | Structured output content |
| `CLAUDE_THINKING_SIGNATURE` | `claude.thinking.signature` | Thinking block signature |
| `CLAUDE_MESSAGE_UUID` | `claude.message.uuid` | Message UUID |
| `CLAUDE_SYSTEM_SUBTYPE` | `claude.system.subtype` | System message subtype |
| `CLAUDE_SYSTEM_DATA` | `claude.system.data` | System message data |

---

## Complete Example

```python
import mantisdk.tracing as tracing
from mantisdk.tracing.semconv import ObservationType, ObservationLevel

def agent_workflow(user_id: str, session_id: str, query: str):
    with tracing.trace("agent_session", input=query) as root:
        # User & session context
        root.set_attribute(tracing.semconv.USER_ID, user_id)
        root.set_attribute(tracing.semconv.SESSION_ID, session_id)

        # Environment context
        root.set_attribute(tracing.semconv.ENVIRONMENT, "production")
        root.set_attribute(tracing.semconv.VERSION, "1.0.0")

        # Child operation with observation attributes
        with tracing.span("llm_generation") as gen_span:
            gen_span.set_attribute(
                tracing.semconv.OBSERVATION_TYPE,
                ObservationType.GENERATION
            )
            gen_span.set_attribute(tracing.semconv.MODEL_NAME, "gpt-4o-mini")
            gen_span.set_attribute(
                tracing.semconv.MODEL_PARAMETERS,
                '{"temperature": 0.7, "max_tokens": 1000}'
            )

            result = call_llm(query)

            gen_span.set_attribute(
                tracing.semconv.OBSERVATION_OUTPUT,
                result
            )

        # Set trace output
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

        return result
```

## All Constants Reference

```python
from mantisdk.tracing import semconv

# --- Insight namespace ---

# Trace-level
semconv.TRACE_NAME
semconv.TRACE_INPUT
semconv.TRACE_OUTPUT
semconv.TRACE_METADATA
semconv.TRACE_TAGS
semconv.TRACE_PUBLIC

# User & Session
semconv.USER_ID
semconv.SESSION_ID

# Run Tracking
semconv.RUN_ID
semconv.RUN_NAME

# Environment
semconv.ENVIRONMENT
semconv.RELEASE
semconv.VERSION

# Observation-level
semconv.OBSERVATION_TYPE
semconv.OBSERVATION_INPUT
semconv.OBSERVATION_OUTPUT
semconv.OBSERVATION_METADATA
semconv.OBSERVATION_LEVEL
semconv.OBSERVATION_STATUS_MESSAGE

# Model
semconv.MODEL_NAME
semconv.MODEL_PARAMETERS

# --- Langfuse compat namespace ---

semconv.LANGFUSE_TRACE_NAME
semconv.LANGFUSE_TRACE_USER_ID
semconv.LANGFUSE_TRACE_SESSION_ID
semconv.LANGFUSE_TRACE_TAGS
semconv.LANGFUSE_TRACE_PUBLIC
semconv.LANGFUSE_TRACE_METADATA
semconv.LANGFUSE_TRACE_INPUT
semconv.LANGFUSE_TRACE_OUTPUT
semconv.LANGFUSE_OBSERVATION_TYPE
semconv.LANGFUSE_OBSERVATION_METADATA
semconv.LANGFUSE_OBSERVATION_LEVEL
semconv.LANGFUSE_OBSERVATION_STATUS_MESSAGE
semconv.LANGFUSE_OBSERVATION_INPUT
semconv.LANGFUSE_OBSERVATION_OUTPUT
semconv.LANGFUSE_OBSERVATION_COMPLETION_START_TIME
semconv.LANGFUSE_OBSERVATION_MODEL
semconv.LANGFUSE_OBSERVATION_MODEL_PARAMETERS
semconv.LANGFUSE_OBSERVATION_USAGE_DETAILS
semconv.LANGFUSE_OBSERVATION_COST_DETAILS
semconv.LANGFUSE_OBSERVATION_PROMPT_NAME
semconv.LANGFUSE_OBSERVATION_PROMPT_VERSION
semconv.LANGFUSE_ENVIRONMENT
semconv.LANGFUSE_RELEASE
semconv.LANGFUSE_VERSION

# --- GenAI namespace ---

semconv.GEN_AI_OPERATION_NAME
semconv.GEN_AI_SYSTEM
semconv.GEN_AI_REQUEST_MODEL
semconv.GEN_AI_RESPONSE_MODEL
semconv.GEN_AI_REQUEST_TEMPERATURE
semconv.GEN_AI_REQUEST_MAX_TOKENS
semconv.GEN_AI_RESPONSE_FINISH_REASONS
semconv.GEN_AI_CONVERSATION_ID
semconv.GEN_AI_INPUT_MESSAGES
semconv.GEN_AI_OUTPUT_MESSAGES
semconv.GEN_AI_USAGE_INPUT_TOKENS
semconv.GEN_AI_USAGE_OUTPUT_TOKENS
semconv.GEN_AI_USAGE_COST
semconv.GEN_AI_TOOL_NAME
semconv.GEN_AI_TOOL_CALL_ARGUMENTS
semconv.GEN_AI_TOOL_CALL_RESULT

# --- OpenInference namespace ---

semconv.OPENINFERENCE_SPAN_KIND
semconv.LLM_MODEL_NAME
semconv.LLM_INVOCATION_PARAMETERS
semconv.LLM_INPUT_MESSAGES
semconv.LLM_OUTPUT_MESSAGES
semconv.LLM_TOKEN_COUNT_PROMPT
semconv.LLM_TOKEN_COUNT_COMPLETION
semconv.LLM_TOKEN_COUNT_TOTAL
semconv.LLM_SYSTEM
semconv.LLM_PROVIDER
semconv.INPUT_VALUE
semconv.OUTPUT_VALUE
semconv.INPUT_MIME_TYPE
semconv.OUTPUT_MIME_TYPE
semconv.TOOL_NAME
semconv.TOOL_DESCRIPTION
semconv.TOOL_PARAMETERS

# Span kind values
semconv.SPAN_KIND_LLM
semconv.SPAN_KIND_TOOL
semconv.SPAN_KIND_AGENT
semconv.SPAN_KIND_CHAIN
semconv.SPAN_KIND_EMBEDDING
semconv.SPAN_KIND_RETRIEVER
semconv.SPAN_KIND_RERANKER
semconv.SPAN_KIND_GUARDRAIL
semconv.SPAN_KIND_EVALUATOR

# MIME types
semconv.MIME_TYPE_TEXT
semconv.MIME_TYPE_JSON

# --- MantisDK namespace ---

semconv.MANTIS_LLM_THINKING
semconv.MANTIS_LLM_COST_USD
semconv.MANTIS_DURATION_MS
semconv.MANTIS_DURATION_API_MS
semconv.MANTIS_NUM_TURNS
semconv.MANTIS_TOOL_IS_ERROR

# --- Claude namespace ---

semconv.CLAUDE_API_ERROR
semconv.CLAUDE_PARENT_TOOL_USE_ID
semconv.CLAUDE_STRUCTURED_OUTPUT
semconv.CLAUDE_THINKING_SIGNATURE
semconv.CLAUDE_MESSAGE_UUID
semconv.CLAUDE_SYSTEM_SUBTYPE
semconv.CLAUDE_SYSTEM_DATA

# --- Classes ---

# from mantisdk.tracing.semconv import ObservationType
# ObservationType.SPAN, ObservationType.GENERATION, ObservationType.EVENT

# from mantisdk.tracing.semconv import ObservationLevel
# ObservationLevel.DEBUG, ObservationLevel.DEFAULT,
# ObservationLevel.WARNING, ObservationLevel.ERROR
```

## Next Steps

- [Traces & Spans](/tracing/traces-and-spans) - Creating spans
- [Insight Export](/tracing/insight-export) - View traces in Insight UI

---

## Traces & Spans

*Source: /tracing/traces-and-spans*

# Traces & Spans

Learn how to create traces and spans to instrument your AI agent.

## The `trace` Class

`tracing.trace` is a class that works as both a **decorator** and a **context manager**. Use whichever style fits your code.

- **Decorator mode** -- wrap a function, and inputs/outputs are captured automatically.
- **Context manager mode** -- open a `with` block and set inputs/outputs yourself.

### Decorator Mode

The simplest way to trace a function. The SDK inspects the function signature, serializes the arguments as input, and captures the return value as output.

```python
import mantisdk.tracing as tracing

@tracing.trace
def process_query(query: str) -> dict:
    """Input and output are captured automatically."""
    result = analyze(query)
    return {"result": result}
```

You can also pass a custom name:

```python
@tracing.trace(name="custom-operation")
def another_function(data):
    process(data)
```

#### Disabling Auto-Capture

Large payloads or sensitive data? Turn off capture for either side:

```python
# Don't record the return value
@tracing.trace(capture_output=False)
def sensitive_function(input_data):
    return process_sensitive(input_data)

# Don't record the function arguments
@tracing.trace(capture_input=False)
def function_with_large_input(big_payload: bytes):
    return summarize(big_payload)

# Turn off both -- only timing and status are recorded
@tracing.trace(capture_input=False, capture_output=False)
def fully_silent(data):
    return transform(data)
```

### Context Manager Mode

When you need full control over what gets recorded, use `trace` as a context manager. You pass `input` up front and set the output on the span directly.

```python
import mantisdk.tracing as tracing

def my_workflow(query: str) -> str:
    with tracing.trace("my-workflow", input=query) as span:
        result = do_work(query)

        # Set output explicitly
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

        # Add custom attributes
        span.set_attribute("custom_key", "custom_value")

        return result
```

The `with` block yields an OpenTelemetry `Span` object. You can call any standard span method on it (`set_attribute`, `add_event`, `record_exception`, etc.).

### Parameters

```python
tracing.trace(
    name_or_func=None,           # Span name (str) or function to decorate
    name=None,                   # Explicit span name (keyword-only)
    input=None,                  # Input value (context manager mode)
    kind=SpanKind.INTERNAL,      # OpenTelemetry span kind
    attributes=None,             # Initial attributes dict
    capture_input=True,          # Auto-capture input (decorator mode only)
    capture_output=True,         # Auto-capture output (decorator mode only)
    record_exception=True,       # Record exceptions on the span
    set_status_on_exception=True # Set ERROR status on exception
)
```

| Parameter | Decorator | Context Manager | Description |
|-----------|:---------:|:---------------:|-------------|
| `name_or_func` | The function | The span name string | First positional argument |
| `name` | Custom span name | -- | Overrides auto-detected function name |
| `input` | -- | Input to record | Ignored in decorator mode (use `capture_input` instead) |
| `capture_input` | Yes | -- | Serialize function args as input |
| `capture_output` | Yes | -- | Serialize return value as output |
| `attributes` | Yes | Yes | Dict of initial span attributes |
| `record_exception` | Yes | Yes | Auto-record exceptions |
| `set_status_on_exception` | Yes | Yes | Set span status to ERROR on exception |

## The `span` Function

`tracing.span()` creates **child spans** within an active trace. Use it to break a trace into logical steps.

```python
import mantisdk.tracing as tracing

def my_workflow(query: str):
    with tracing.trace("workflow", input=query):
        # Step 1 -- child of workflow
        with tracing.span("step.classify", input=query) as s1:
            query_type = classify(query)
            s1.set_attribute("query_type", query_type)

        # Step 2 -- another child of workflow
        with tracing.span("step.process") as s2:
            result = process(query, query_type)
            s2.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, result)

        # Nested spans
        with tracing.span("step.generate"):
            with tracing.span("step.generate.format"):
                formatted = format_output(result)
            with tracing.span("step.generate.validate"):
                validated = validate(formatted)
```

### Parameters

```python
tracing.span(
    name,                        # Span name (required)
    input=None,                  # Input value
    kind=SpanKind.INTERNAL,      # OpenTelemetry span kind
    attributes=None,             # Initial attributes
    record_exception=True,       # Record exceptions
    set_status_on_exception=True # Set ERROR status
)
```

## The `tool` Function

`tracing.tool()` creates spans for tool/function calls with `CLIENT` span kind. This shows up differently in the Insight UI, making it easy to spot external calls.

```python
import mantisdk.tracing as tracing

def search_database(query: str) -> list:
    """Trace database search as a tool call."""
    with tracing.tool("search_database", input=query) as span:
        results = db.execute(query)
        span.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(results)))
        return results

def call_external_api(endpoint: str, payload: dict) -> dict:
    """Trace API call as a tool."""
    with tracing.tool("api_call", input=payload) as span:
        span.set_attribute("endpoint", endpoint)
        response = requests.post(endpoint, json=payload)
        span.set_attribute("status_code", response.status_code)
        return response.json()
```

### Parameters

```python
tracing.tool(
    name,                        # Tool/span name (required)
    input=None,                  # Tool input
    tool_name=None,              # Explicit tool.name attribute
    attributes=None,             # Initial attributes
    record_exception=True,       # Record exceptions
    set_status_on_exception=True # Set ERROR status
)
```

## Working with Spans

### Setting Attributes

```python
with tracing.trace("workflow") as span:
    # Standard attributes
    span.set_attribute("key", "value")
    span.set_attribute("count", 42)
    span.set_attribute("enabled", True)

    # Semantic conventions
    span.set_attribute(tracing.semconv.USER_ID, "user-123")
    span.set_attribute(tracing.semconv.SESSION_ID, "session-456")
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, "result")
```

To group multiple traces into a session (e.g., a multi-turn conversation), use `tracing_context(session_id=...)` instead of setting the attribute on each span individually. See [Context Tagging -- Session Tracking](/tracing/context-tagging#session-tracking) for examples.

### Getting the Current Span

```python
import mantisdk.tracing as tracing

def inner_function():
    """Access the current active span."""
    span = tracing.get_current_span()
    span.set_attribute("inner_data", "value")

with tracing.trace("outer"):
    inner_function()  # Can access the outer span
```

### Recording Exceptions

Exceptions are automatically recorded, but you can also record them manually:

```python
with tracing.span("risky_operation") as span:
    try:
        risky_code()
    except ValueError as e:
        span.record_exception(e)
        span.set_status(StatusCode.ERROR, str(e))
        # Handle or re-raise
```

## Span Hierarchy Example

```python
import mantisdk.tracing as tracing

def agent_workflow(user_query: str):
    """
    Creates this span hierarchy:

    agent_workflow
    +-- classify
    +-- retrieve
    |   +-- search_db (tool)
    |   +-- search_web (tool)
    +-- generate
        +-- format
        +-- validate
    """
    with tracing.trace("agent_workflow", input=user_query) as root:
        root.set_attribute(tracing.semconv.USER_ID, "user-123")

        # Classify
        with tracing.span("classify", input=user_query) as s:
            query_type = classify(user_query)
            s.set_attribute("query_type", query_type)

        # Retrieve (with tool calls)
        with tracing.span("retrieve") as s:
            with tracing.tool("search_db", input=user_query):
                db_results = search_db(user_query)
            with tracing.tool("search_web", input=user_query):
                web_results = search_web(user_query)
            s.set_attribute("total_results", len(db_results) + len(web_results))

        # Generate
        with tracing.span("generate") as s:
            with tracing.span("format"):
                formatted = format_results(db_results, web_results)
            with tracing.span("validate"):
                validated = validate_output(formatted)
            s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, validated)

        root.set_attribute(tracing.semconv.TRACE_OUTPUT, validated)
        return validated
```

## Next Steps

- [Semantic Conventions](/tracing/semantic-conventions) - Standard attribute names
- [Async Support](/tracing/async-support) - Async span creation
- [Auto-Instrumentation](/tracing/auto-instrumentation) - Framework support

---