Add MCP Agent instrumentation support for AgentOps tracing

Co-authored-by: alex <alex@agentops.ai>

Author: cursoragent

agentops/instrumentation/__init__.py

@@ -107,6 +107,12 @@ class InstrumentorConfig(TypedDict):
         "class_name": "SmolagentsInstrumentor",
         "min_version": "1.0.0",
     },
+    "mcp_agent": {
+        "module_name": "agentops.instrumentation.agentic.mcp_agent",
+        "class_name": "MCPAgentInstrumentor",
+        "min_version": "0.1.0",
+        "package_name": "mcp-agent",
+    },
     "langgraph": {
         "module_name": "agentops.instrumentation.agentic.langgraph",
         "class_name": "LanggraphInstrumentor",

agentops/instrumentation/agentic/mcp_agent/__init__.py

@@ -0,0 +1,5 @@
+"""MCP Agent instrumentation for AgentOps."""
+
+from agentops.instrumentation.agentic.mcp_agent.instrumentor import MCPAgentInstrumentor
+
+__all__ = ["MCPAgentInstrumentor"]

agentops/instrumentation/agentic/mcp_agent/instrumentor.py

@@ -0,0 +1,50 @@
+"""MCP Agent Instrumentation for AgentOps
+
+This instrumentor hooks into mcp-agent's telemetry to ensure spans are created
+with the AgentOps tracer provider and to prevent duplicate or conflicting tracer usage.
+"""
+
+from typing import Dict, Any
+
+from opentelemetry.metrics import Meter
+
+from agentops.logging import logger
+from agentops.instrumentation.common import CommonInstrumentor, StandardMetrics, InstrumentorConfig
+from agentops.instrumentation.agentic.mcp_agent.patch import patch_mcp_agent, unpatch_mcp_agent
+
+# Library info for tracer/meter
+LIBRARY_NAME = "agentops.instrumentation.agentic.mcp_agent"
+LIBRARY_VERSION = "0.1.0"
+
+
+class MCPAgentInstrumentor(CommonInstrumentor):
+    """An instrumentor for Lastmile MCP Agent.
+
+    This instrumentor patches mcp-agent's telemetry get_tracer to return the
+    AgentOps tracer, ensuring spans flow through AgentOps' configured provider.
+    """
+
+    def __init__(self):
+        """Initialize the MCP Agent instrumentor."""
+        config = InstrumentorConfig(
+            library_name=LIBRARY_NAME,
+            library_version=LIBRARY_VERSION,
+            wrapped_methods=[],  # We use patching
+            metrics_enabled=True,
+            dependencies=["mcp-agent >= 0.1.0"],
+        )
+        super().__init__(config)
+
+    def _create_metrics(self, meter: Meter) -> Dict[str, Any]:
+        """Create metrics for the instrumentor."""
+        return StandardMetrics.create_standard_metrics(meter)
+
+    def _custom_wrap(self, **kwargs):
+        """Apply custom patching for MCP Agent."""
+        patch_mcp_agent(self._tracer)
+        logger.info("MCP Agent instrumentation enabled")
+
+    def _custom_unwrap(self, **kwargs):
+        """Remove custom patching from MCP Agent."""
+        unpatch_mcp_agent()
+        logger.info("MCP Agent instrumentation disabled")

agentops/instrumentation/agentic/mcp_agent/patch.py

@@ -0,0 +1,66 @@
+"""Patching utilities for mcp-agent telemetry integration.
+
+We replace mcp_agent.tracing.telemetry.get_tracer to return the AgentOps tracer,
+so spans created by mcp-agent flow through AgentOps' configured provider.
+"""
+
+from typing import Optional
+
+from opentelemetry.trace import Tracer
+
+from agentops.logging import logger
+
+
+_original_get_tracer = None  # type: ignore[var-annotated]
+
+
+def patch_mcp_agent(agentops_tracer: Optional[Tracer]) -> None:
+    """Patch mcp-agent telemetry to use the AgentOps tracer.
+
+    If the AgentOps tracer is None (unexpected), the patch is skipped.
+    """
+    if agentops_tracer is None:
+        logger.debug("patch_mcp_agent: AgentOps tracer is None; skipping patch")
+        return
+
+    global _original_get_tracer
+
+    try:
+        import mcp_agent.tracing.telemetry as mcp_telemetry  # type: ignore
+
+        get_tracer_func = getattr(mcp_telemetry, "get_tracer", None)
+        if get_tracer_func is None:
+            logger.debug("patch_mcp_agent: mcp_agent.tracing.telemetry.get_tracer not found; skipping patch")
+            return
+
+        if getattr(get_tracer_func, "__agentops_patched__", False):
+            logger.debug("patch_mcp_agent: already patched; skipping")
+            return
+
+        _original_get_tracer = get_tracer_func
+
+        def _agentops_get_tracer(_context):  # context is ignored; AgentOps manages provider
+            return agentops_tracer
+
+        # Tag function to avoid double patching
+        setattr(_agentops_get_tracer, "__agentops_patched__", True)
+
+        mcp_telemetry.get_tracer = _agentops_get_tracer  # type: ignore[attr-defined]
+        logger.debug("Patched mcp_agent.tracing.telemetry.get_tracer to use AgentOps tracer")
+    except Exception as e:
+        logger.debug(f"patch_mcp_agent: failed to patch mcp-agent telemetry: {e}")
+
+
+def unpatch_mcp_agent() -> None:
+    """Restore original mcp-agent telemetry.get_tracer if it was patched."""
+    global _original_get_tracer
+    try:
+        if _original_get_tracer is None:
+            return
+        import mcp_agent.tracing.telemetry as mcp_telemetry  # type: ignore
+
+        mcp_telemetry.get_tracer = _original_get_tracer  # type: ignore[attr-defined]
+        _original_get_tracer = None
+        logger.debug("Restored original mcp_agent.tracing.telemetry.get_tracer")
+    except Exception as e:
+        logger.debug(f"unpatch_mcp_agent: failed to restore mcp-agent telemetry: {e}")