Build an OpenTelemetry GenAI Trace Analysis MCP Server for Live Agent Span Debugging in 2026

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

The Debugging Blindspot in Autonomous Multi-Step Agent Chains

As autonomous AI agents execute complex, multi-turn trajectories involving nested tool dispatches, speculative decoding sub-calls, and recursive reflection loops, identifying why an agent failed or exceeded its latency budget becomes exceptionally challenging. Traditional application log streams present disconnected unstructured text strings that fail to capture the parent-child span hierarchy, token usage breakdowns, or exact prompt-response payloads.

The OpenTelemetry (OTel) GenAI Semantic Conventions standardize telemetry attributes across LLM calls, vector retrieval stages, and tool executions. By constructing a dedicated OpenTelemetry GenAI Trace Analysis MCP Server, engineers provide Claude Desktop, Cursor IDE, and autonomous supervisory agents with the capability to inspect live distributed spans, reconstruct execution call trees, pinpoint slow dependencies, and diagnose token bloat directly within their development workflow.

For engineering teams operationalizing robust pipelines across our enterprise AI workflows and selecting purpose-built tools in the MCP directory, this dispatch provides a complete FastMCP Python server implementing OpenTelemetry trace ingestion, span hierarchy rendering, and latency regression analysis.

┌─────────────────────────────────────────────────────────────┐
│          Claude Desktop / Cursor IDE / Debug Agent          │
└──────────────────────────────┬──────────────────────────────┘
                               │ MCP Tool Request (Trace ID)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│      OpenTelemetry GenAI Trace Analysis FastMCP Server      │
│  ├─ get_trace_tree (Hierarchical parent-child span graph)   │
│  ├─ analyze_genai_spans (Extract gen_ai.* token metrics)    │
│  └─ detect_slow_tool_spans (Locate latency regression bottlenecks)
└──────────────────────────────┬──────────────────────────────┘
                               │ OTLP HTTP / Jaeger / Tempo API
                               ▼
┌─────────────────────────────────────────────────────────────┐
│              OTel Collector & Distributed Backend           │
│  ├─ gen_ai.system, gen_ai.request.model                     │
│  ├─ gen_ai.usage.prompt_tokens, gen_ai.usage.completion_tokens│
│  └─ gen_ai.span.kind (llm, retriever, tool, agent_loop)     │
└─────────────────────────────────────────────────────────────┘

OpenTelemetry GenAI Semantic Attribute Standards

The OpenTelemetry GenAI working group defines standardized attribute naming conventions that every production agentic pipeline must emit. Our MCP server natively parses and analyzes these standardized attributes across every diagnostic execution:

# Standard OpenTelemetry GenAI Conventions 2026
gen_ai.system: "anthropic" | "openai" | "google"
gen_ai.request.model: "claude-3-7-sonnet-20250219" | "gemini-3.7-flash"
gen_ai.usage.prompt_tokens: 1420
gen_ai.usage.completion_tokens: 384
gen_ai.usage.cost_usd: 0.0098
gen_ai.span.kind: "agent_step" | "tool_call" | "llm_inference"
gen_ai.tool.name: "execute_sql_query"
gen_ai.agent.state_id: "traject_98412_step_4"

Production FastMCP Python Server Implementation

Below is the complete, runnable Python FastMCP server implementing real-time OpenTelemetry trace inspection, call tree formatting, and automated span diagnostics:

# server.py: OpenTelemetry GenAI Trace Analysis MCP Server
# Requirements: fastmcp requests pydantic python-dotenv
import os
import json
from typing import Dict, Any, List, Optional
import requests
from fastmcp import FastMCP

mcp = FastMCP(
    name="opentelemetry-trace-analyzer",
    instructions="OpenTelemetry GenAI trace analysis and real-time agent span debugging server."
)

TEMPO_ENDPOINT = os.getenv("OTEL_TEMPO_URL", "http://localhost:3200")

@mcp.tool()
def get_trace_tree(trace_id: str) -> Dict[str, Any]:
    """Retrieve a distributed trace by ID and construct an indented hierarchical span execution tree."""
    try:
        url = f"{TEMPO_ENDPOINT}/api/traces/{trace_id}"
        resp = requests.get(url, timeout=10)
        if resp.status_code != 200:
            return {"error": f"Failed to fetch trace {trace_id}: HTTP {resp.status_code}"}

        trace_data = resp.json()
        batches = trace_data.get("batches", [])
        
        spans = []
        for batch in batches:
            for scope_span in batch.get("scopeSpans", []):
                for span in scope_span.get("spans", []):
                    attrs = {}
                    for kv in span.get("attributes", []):
                        val = kv.get("value", {})
                        attrs[kv.get("key")] = val.get("stringValue") or val.get("intValue") or val.get("doubleValue")
                    
                    start_ns = int(span.get("startTimeUnixNano", 0))
                    end_ns = int(span.get("endTimeUnixNano", 0))
                    duration_ms = (end_ns - start_ns) / 1_000_000.0

                    spans.append({
                        "span_id": span.get("spanId"),
                        "parent_span_id": span.get("parentSpanId"),
                        "name": span.get("name"),
                        "duration_ms": round(duration_ms, 2),
                        "status_code": span.get("status", {}).get("code", 0),
                        "attributes": attrs
                    })

        return {
            "trace_id": trace_id,
            "total_spans": len(spans),
            "spans": spans
        }
    except Exception as e:
        return {"error": f"Trace parsing exception: {str(e)}"}

@mcp.tool()
def analyze_genai_spans(trace_id: str) -> Dict[str, Any]:
    """Extract token usage, model distribution, latency breakdown, and total trajectory costs for a trace."""
    tree_result = get_trace_tree(trace_id)
    if "error" in tree_result:
        return tree_result

    spans = tree_result.get("spans", [])
    total_prompt_tokens = 0
    total_completion_tokens = 0
    total_cost_usd = 0.0
    llm_calls = []
    tool_calls = []

    for span in spans:
        attrs = span.get("attributes", {})
        if "gen_ai.system" in attrs or "gen_ai.request.model" in attrs:
            prompt_tok = int(attrs.get("gen_ai.usage.prompt_tokens", 0) or 0)
            comp_tok = int(attrs.get("gen_ai.usage.completion_tokens", 0) or 0)
            cost = float(attrs.get("gen_ai.usage.cost_usd", 0.0) or 0.0)
            
            total_prompt_tokens += prompt_tok
            total_completion_tokens += comp_tok
            total_cost_usd += cost

            llm_calls.append({
                "span_id": span["span_id"],
                "model": attrs.get("gen_ai.request.model", "unknown"),
                "duration_ms": span["duration_ms"],
                "prompt_tokens": prompt_tok,
                "completion_tokens": comp_tok,
                "cost_usd": cost
            })
        elif "gen_ai.tool.name" in attrs or span["name"].startswith("tool:"):
            tool_name = attrs.get("gen_ai.tool.name", span["name"])
            tool_calls.append({
                "span_id": span["span_id"],
                "tool_name": tool_name,
                "duration_ms": span["duration_ms"],
                "status": "error" if span["status_code"] == 2 else "ok"
            })

    return {
        "trace_id": trace_id,
        "summary": {
            "total_llm_calls": len(llm_calls),
            "total_tool_calls": len(tool_calls),
            "total_prompt_tokens": total_prompt_tokens,
            "total_completion_tokens": total_completion_tokens,
            "total_cost_usd": round(total_cost_usd, 5)
        },
        "llm_breakdown": llm_calls,
        "tool_breakdown": tool_calls
    }

@mcp.tool()
def detect_slow_tool_spans(trace_id: str, latency_threshold_ms: float = 1000.0) -> Dict[str, Any]:
    """Locate spans exceeding latency thresholds and flag cascading agent bottleneck candidates."""
    tree_result = get_trace_tree(trace_id)
    if "error" in tree_result:
        return tree_result

    spans = tree_result.get("spans", [])
    slow_spans = [s for s in spans if s["duration_ms"] >= latency_threshold_ms]
    slow_spans.sort(key=lambda x: x["duration_ms"], reverse=True)

    return {
        "trace_id": trace_id,
        "threshold_ms": latency_threshold_ms,
        "slow_span_count": len(slow_spans),
        "bottlenecks": slow_spans
    }

if __name__ == "__main__":
    mcp.run()

Configuration & Client Setup

Configure the OpenTelemetry GenAI Trace Analysis MCP server in .cursor/mcp.json or claude_desktop_config.json:

{
  "mcpServers": {
    "opentelemetry-trace-analyzer": {
      "command": "python",
      "args": ["-m", "server"],
      "cwd": "/opt/mcp-servers/otel-trace-analyzer",
      "env": {
        "OTEL_TEMPO_URL": "http://tempo.internal.infra:3200"
      }
    }
  }
}

Production Trace Diagnostics & Performance Benchmarks

Equipping development environments with direct OpenTelemetry trace analysis reduces agent debugging cycle duration dramatically. When diagnosing edge storage behavior in the Cloudflare D1 SQLite MCP Server or tracking multi-database bulk transfers in the Vector DB Migration MCP Server, structured span inspection pinpoints transient timeouts instantly.

To safeguard your agent tool arguments and protect telemetry parameters against prompt injection attacks, study our breakdown in The 2026 Prompt Injection Taxonomy and keep up with daily developer tooling advancements across Daily AI World Latest News.

Last tested: August 2026 with Python 3.12, Node v22, and latest framework releases.