How to Trace MCP Servers and Clients

# Install with MCP integration
pip install "honeyhive[openinference-mcp]"

# Alternative: Manual installation
pip install honeyhive openinference-instrumentation-mcp mcp>=1.0.0

from honeyhive import HoneyHiveTracer
from openinference.instrumentation.mcp import MCPInstrumentor
import os

# Step 1: Initialize HoneyHive tracer first
tracer = HoneyHiveTracer.init(api_key=os.getenv("HH_API_KEY"))

# Step 2: Initialize instrumentor with tracer_provider
instrumentor = MCPInstrumentor()
instrumentor.instrument(tracer_provider=tracer.provider)

# MCP calls are now traced (experimental)

from mcp.server import Server
from mcp.types import Tool, TextContent
from honeyhive import HoneyHiveTracer, trace, enrich_span
import os

# Initialize HoneyHive
tracer = HoneyHiveTracer.init(
    api_key=os.getenv("HH_API_KEY"),
    source="mcp-server"
)

# Create MCP server
server = Server("my-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """List available tools."""
    return [
        Tool(
            name="get_weather",
            description="Get weather for a location",
            inputSchema={
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        ),
        Tool(
            name="search_database",
            description="Search the database",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
@trace
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle tool calls with tracing."""
    
    # Add context about the tool call
    enrich_span({
        "tool_name": name,
        "arguments": arguments,
        "service_type": "mcp_server"
    })
    
    if name == "get_weather":
        result = await get_weather(arguments["location"])
    elif name == "search_database":
        result = await search_database(arguments["query"])
    else:
        result = f"Unknown tool: {name}"
    
    enrich_span({"result_length": len(result)})
    
    return [TextContent(type="text", text=result)]

@trace
async def get_weather(location: str) -> str:
    """Get weather for a location."""
    enrich_span({"location": location})
    # Your weather API call here
    return f"Weather in {location}: 72°F, Sunny"

@trace
async def search_database(query: str) -> str:
    """Search the database."""
    enrich_span({"query": query})
    # Your database query here
    return f"Found 5 results for: {query}"

from mcp.types import McpError, ErrorCode

@server.call_tool()
@trace
async def call_tool_safe(name: str, arguments: dict) -> list[TextContent]:
    """Handle tool calls with comprehensive error handling."""
    
    enrich_span({
        "tool_name": name,
        "arguments": arguments
    })
    
    try:
        # Validate tool exists
        if name not in ["get_weather", "search_database"]:
            enrich_span({"error": f"Unknown tool: {name}"})
            raise McpError(ErrorCode.MethodNotFound, f"Unknown tool: {name}")
        
        # Execute tool
        if name == "get_weather":
            if "location" not in arguments:
                enrich_span({"error": "Missing location"})
                raise McpError(ErrorCode.InvalidParams, "Missing location")
            result = await get_weather(arguments["location"])
        
        elif name == "search_database":
            if "query" not in arguments:
                enrich_span({"error": "Missing query"})
                raise McpError(ErrorCode.InvalidParams, "Missing query")
            result = await search_database(arguments["query"])
        
        enrich_span({"success": True, "result": result})
        return [TextContent(type="text", text=result)]
        
    except McpError:
        raise
    except Exception as e:
        enrich_span({
            "error": str(e),
            "error_type": type(e).__name__
        })
        raise McpError(ErrorCode.InternalError, str(e))

from mcp import ClientSession
from mcp.client.stdio import stdio_client
from honeyhive import HoneyHiveTracer, trace, enrich_span
import os

tracer = HoneyHiveTracer.init(
    api_key=os.getenv("HH_API_KEY"),
    source="mcp-client"
)

@trace
async def use_mcp_tools():
    """Use MCP tools with tracing."""
    
    enrich_span({"service_type": "mcp_client"})
    
    async with stdio_client("path/to/mcp-server") as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize session
            await session.initialize()
            
            # List available tools
            tools = await session.list_tools()
            enrich_span({"tools_available": len(tools.tools)})
            
            # Call a tool
            result = await session.call_tool(
                "get_weather",
                arguments={"location": "Paris"}
            )
            
            enrich_span({
                "tool_called": "get_weather",
                "result": result.content[0].text
            })
            
            return result

@trace
async def call_multiple_servers(query: str) -> dict:
    """Call tools across multiple MCP servers."""
    
    results = {}
    
    servers = {
        "weather": "path/to/weather-server",
        "database": "path/to/database-server",
        "calendar": "path/to/calendar-server"
    }
    
    enrich_span({
        "query": query,
        "servers_count": len(servers)
    })
    
    for server_name, server_path in servers.items():
        try:
            async with stdio_client(server_path) as (read, write):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    
                    # Get tools from this server
                    tools = await session.list_tools()
                    
                    # Call appropriate tool based on query
                    if server_name == "weather" and "weather" in query.lower():
                        result = await session.call_tool(
                            "get_weather",
                            {"location": "NYC"}
                        )
                        results[server_name] = result.content[0].text
                        
        except Exception as e:
            results[server_name] = f"Error: {e}"
    
    enrich_span({
        "servers_called": list(results.keys()),
        "success_count": len([r for r in results.values() if not r.startswith("Error")])
    })
    
    return results

import asyncio

@trace
async def call_tool_with_retry(
    session: ClientSession,
    tool_name: str,
    arguments: dict,
    max_retries: int = 3
) -> str:
    """Call MCP tool with retry logic."""
    
    enrich_span({
        "tool_name": tool_name,
        "max_retries": max_retries
    })
    
    last_error = None
    
    for attempt in range(max_retries):
        try:
            result = await session.call_tool(tool_name, arguments=arguments)
            
            enrich_span({
                "attempt": attempt + 1,
                "success": True
            })
            
            return result.content[0].text
            
        except Exception as e:
            last_error = e
            enrich_span({
                "attempt": attempt + 1,
                "error": str(e),
                "will_retry": attempt < max_retries - 1
            })
            
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                await asyncio.sleep(wait_time)
    
    raise Exception(f"Max retries exceeded: {last_error}")

from langchain.tools import BaseTool
from honeyhive import HoneyHiveTracer, trace, enrich_span

tracer = HoneyHiveTracer.init(api_key=os.getenv("HH_API_KEY"))

class MCPTool(BaseTool):
    """LangChain tool that calls MCP server."""
    
    name: str = "mcp_tool"
    description: str = "Call an MCP server tool"
    
    @trace
    def _run(self, query: str) -> str:
        """Execute MCP tool via LangChain."""
        enrich_span({
            "framework": "langchain",
            "tool_type": "mcp"
        })
        
        # Call MCP server (implement your client logic)
        return call_mcp_tool_sync(query)

from honeyhive import HoneyHiveTracer, trace, enrich_span
from openinference.instrumentation.openai import OpenAIInstrumentor
import openai
import json

tracer = HoneyHiveTracer.init(api_key=os.getenv("HH_API_KEY"))
instrumentor = OpenAIInstrumentor()
instrumentor.instrument(tracer_provider=tracer.provider)

@trace
async def chat_with_mcp_tools(user_message: str):
    """Use OpenAI with MCP tools."""
    
    # Get tools from MCP server
    mcp_tools = await get_mcp_tools()
    
    # Convert to OpenAI format
    openai_tools = convert_to_openai_format(mcp_tools)
    
    enrich_span({
        "tools_count": len(openai_tools),
        "user_message_length": len(user_message)
    })
    
    # Call OpenAI
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}],
        tools=openai_tools
    )
    
    # Execute any tool calls via MCP
    if response.choices[0].message.tool_calls:
        for tool_call in response.choices[0].message.tool_calls:
            result = await execute_mcp_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            enrich_span({
                f"tool_{tool_call.function.name}": "executed"
            })
    
    return response

# HoneyHive configuration
export HH_API_KEY="your-honeyhive-api-key"
export HH_SOURCE="mcp-server"  # or "mcp-client"

# MCP configuration
export MCP_SERVER_URL="http://localhost:8000"

# ✅ Good: Separate traces for each step
@trace
async def handle_tool(name: str, args: dict):
    enrich_span({"phase": "validation"})
    validate_arguments(name, args)
    
    enrich_span({"phase": "execution"})
    result = await execute_tool(name, args)
    
    enrich_span({"phase": "formatting"})
    return format_result(result)

# ✅ Good: Rich context for debugging
@server.call_tool()
@trace
async def call_tool(name: str, arguments: dict):
    enrich_span({
        "tool_name": name,
        "tool_version": TOOL_VERSIONS.get(name, "unknown"),
        "argument_count": len(arguments),
        "server_id": SERVER_ID
    })
    # ...

import time

@trace
async def call_tool_with_metrics(name: str, arguments: dict):
    """Track metrics for tool calls."""
    
    start_time = time.time()
    
    enrich_span({
        "input_size_bytes": len(str(arguments))
    })
    
    result = await execute_tool(name, arguments)
    
    duration_ms = (time.time() - start_time) * 1000
    enrich_span({
        "metrics": {
            "output_size_bytes": len(result),
            "execution_time_ms": duration_ms
        }
    })
    
    return result

# ✅ Correct - @trace after @server.call_tool()
@server.call_tool()
@trace
async def call_tool(name: str, arguments: dict):
    ...

# ❌ Wrong - decorator order matters
@trace
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    ...

# ✅ Correct
@trace
async def my_function():
    ...

# ❌ Wrong - missing tracer
@trace
async def my_function():
    ...

import time

@trace
async def diagnose_latency(name: str, arguments: dict):
    
    t1 = time.time()
    payload = serialize(arguments)
    
    t2 = time.time()
    response = await send_request(payload)
    
    t3 = time.time()
    result = deserialize(response)
    
    enrich_span({
        "timing": {
            "serialize_ms": (t2 - t1) * 1000,
            "network_ms": (t3 - t2) * 1000,
            "deserialize_ms": (time.time() - t3) * 1000
        }
    })
    
    return result