Skip to main content
Trace MCP tool calls, server responses, and client interactions in HoneyHive so you can debug agent workflows that rely on the Model Context Protocol. Use OpenInference for experimental auto-instrumentation or @trace decorators for production MCP servers and clients. See custom spans for decorator patterns and tracer initialization for runtime setup.
MCP Integration: Trace tool calls, server responses, and client interactions in your MCP-based AI applications.

​
What is the Model Context Protocol?

The Model Context Protocol (MCP) is an open standard for connecting AI assistants to external tools and data sources. HoneyHive can trace both MCP servers (tool providers) and MCP clients (AI applications).

​
What MCP versions does HoneyHive support?

​
Python version support

Support LevelPython Versions
Fully Supported3.11, 3.12, 3.13
Not Supported3.10 and below

​
MCP SDK Requirements

  • Minimum: mcp >= 1.0.0
  • Recommended: mcp >= 1.0.0
  • Protocol: MCP 1.0

​
Instrumentor Status

InstrumentorStatusNotes
OpenInferenceExperimentalBasic protocol tracing available
TraceloopNot SupportedUse OpenInference or manual tracing
MCP instrumentation is experimental. For production use, we recommend manual tracing with the @trace decorator.

​
How do you auto-trace MCP with OpenInference?

​
Installation

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

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

​
Basic Setup

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)

​
How do you manually trace MCP in production?

For production MCP applications, use manual tracing with the @trace decorator.

​
MCP server instrumentation

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}"

​
Server with Error Handling

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))

​
MCP Client Instrumentation

Trace your MCP client to monitor how your AI application uses external tools.

​
Basic Client Setup

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

​
Client with Multiple Servers

@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

​
Client with Retry Logic

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}")

​
Integration with AI Frameworks

​
With LangChain

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)

​
With OpenAI Function Calling

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

​
Environment Configuration

# 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"

​
Best Practices

​
1. Trace Tool Boundaries Clearly

# ✅ 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)

​
2. Include Rich Context

# ✅ 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
    })
    # ...

​
3. Track Metrics

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

​
Troubleshooting

​
Tool Calls Not Appearing in Traces

Ensure the @trace decorator is correctly applied: 
# ✅ 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):
    ...

​
Missing Tracer Reference

Always pass the tracer to the decorator: 
# ✅ Correct
@trace
async def my_function():
    ...

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

​
High Latency Investigation

Add timing metadata to identify bottlenecks: 
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

LangChain Integration

Use MCP with LangChain

Distributed Tracing

Link client and server traces

Span Enrichment

Add metadata to traces

Multi-Provider

Use MCP with multiple providers