feat(mcp-instrumentation): suppress HTTP and initialization spans (#695)

*Description of changes:*

- Suppress redundant HTTP/ASGI spans on MCP client and server by default

- Patch `OpenTelemetryMiddleware.__call__` to respect
`suppress_http_instrumentation` context
note: SHOULD remove this patch once
open-telemetry/opentelemetry-python-contrib#4375
is merged

- Skip creating MCP spans for initialize and notifications/initialized
handshake messages

- Introduce `OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION` env var, defaults
to true


<img width="1038" height="546" alt="image"
src="https://github.com/user-attachments/assets/d6f0254a-3b4f-4d46-bb32-424d6fee12b3"
/>

 


By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

Author: liustve

CHANGELOG.md

@@ -12,6 +12,8 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ## Unreleased
 
+- feat: suppress redundant HTTP/ASGI and initialization spans in MCP instrumentation
+  ([#695](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/695))
 - feat: [BREAKING CHANGE] introduce AWS_AGENTIC_OBSERVABILITY_OPT_IN and refactor agent observability config
   ([#691](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/691))
 - feat: propagate HTTP context for MCP requests and prefix all span names with mcp

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/mcp/__init__.py

@@ -17,6 +17,7 @@
 _STDIO_MODULE = "mcp.client.stdio"
 _HTTP_MODULE = "mcp.client.streamable_http"
 _SSE_MODULE = "mcp.client.sse"
+_FASTMCP_MODULE = "mcp.server.fastmcp.server"
 
 
 class McpInstrumentor(BaseInstrumentor):
@@ -76,10 +77,24 @@ def _instrument(self, **kwargs: Any) -> None:
             self._client_wrapper.wrap_handle_post_request,
         )
 
+        _LOG.debug("Instrument MCP server ASGI apps to suppress redundant HTTP spans.")
+
+        try_wrap(
+            _FASTMCP_MODULE,
+            "FastMCP.streamable_http_app",
+            self._server_wrapper.wrap_mcp_http_sse_app_factory,
+        )
+        try_wrap(
+            _FASTMCP_MODULE,
+            "FastMCP.sse_app",
+            self._server_wrapper.wrap_mcp_http_sse_app_factory,
+        )
+
     def _uninstrument(self, **kwargs: Any) -> None:  # pylint: disable=no-self-use
         try:
             # pylint: disable=import-outside-toplevel
             from mcp.client import sse, stdio, streamable_http
+            from mcp.server.fastmcp import server as fastmcp_server
             from mcp.server.lowlevel import server
             from mcp.shared import session
 
@@ -100,5 +115,7 @@ def _uninstrument(self, **kwargs: Any) -> None:  # pylint: disable=no-self-use
                     streamable_http.StreamableHTTPTransport,
                     "_handle_post_request",
                 )
-        except ImportError:
-            _LOG.debug("MCP SDK not available, nothing to uninstrument")
+            try_unwrap(fastmcp_server.FastMCP, "streamable_http_app")
+            try_unwrap(fastmcp_server.FastMCP, "sse_app")
+        except (ImportError, AttributeError) as exc:
+            _LOG.debug("MCP SDK not fully available, nothing to uninstrument: %s", exc)

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/mcp/_wrappers.py

@@ -2,13 +2,15 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
+import os
 from contextlib import asynccontextmanager
 from contextvars import Token
 from typing import Any, Callable, Coroutine, Dict, Optional, Tuple
 from urllib.parse import urlparse
 
 from amazon.opentelemetry.distro.instrumentation.common.instrumentation_utils import serialize_to_json_string
 from opentelemetry import context, trace
+from opentelemetry.instrumentation.utils import suppress_http_instrumentation
 from opentelemetry.propagate import get_global_textmap
 from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import (
     GEN_AI_OPERATION_NAME,
@@ -35,6 +37,8 @@
 
 _LOG = logging.getLogger(__name__)
 
+OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION = "OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION"
+
 # Context key for storing client transport metadata alongside the session span.
 _TRANSPORT_KEY = context.create_key("mcp_client_transport")
 
@@ -53,6 +57,20 @@ class McpWrapper:
     def __init__(self, tracer: trace.Tracer, **kwargs: Any) -> None:
         self._tracer = tracer
         self._propagators = kwargs.get("propagators") or get_global_textmap()
+        self._should_suppress_http_spans = (
+            os.environ.get(OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION, "true").lower() == "true"
+        )
+
+    @staticmethod
+    def _should_suppress_mcp_span(message: Any) -> bool:
+        from mcp import types  # pylint: disable=import-outside-toplevel
+
+        if isinstance(
+            message, (types.ClientRequest, types.ClientNotification, types.ServerRequest, types.ServerNotification)
+        ):
+            message = message.root
+        # noisy spans most of the time
+        return isinstance(message, (types.InitializeRequest, types.InitializedNotification))
 
     @staticmethod
     def _set_mcp_attributes(span: trace.Span, message: Any, request_id: Optional[int]) -> None:
@@ -179,6 +197,9 @@ async def async_wrapper() -> Any:
             if not message:
                 return await wrapped(*args, **kwargs)
 
+            if self._should_suppress_http_spans and self._should_suppress_mcp_span(message):
+                return await wrapped(*args, **kwargs)
+
             span = self._tracer.start_span(name=self._CLIENT_SPAN_NAME, kind=SpanKind.CLIENT)
             ctx = trace.set_span_in_context(span)
             token = context.attach(ctx)
@@ -252,8 +273,13 @@ async def wrapper():
                 }
             )
             try:
-                async with wrapped(*args, **kwargs) as streams:
-                    yield streams
+                if self._should_suppress_http_spans:
+                    with suppress_http_instrumentation():
+                        async with wrapped(*args, **kwargs) as streams:
+                            yield streams
+                else:
+                    async with wrapped(*args, **kwargs) as streams:
+                        yield streams
             finally:
                 session_span.end()
                 context.detach(token)
@@ -292,7 +318,11 @@ async def async_wrapper():
             if ctx:
                 token = context.attach(ctx)
                 try:
-                    return await wrapped(*args, **kwargs)
+                    if self._should_suppress_http_spans:
+                        with suppress_http_instrumentation():
+                            return await wrapped(*args, **kwargs)
+                    else:
+                        return await wrapped(*args, **kwargs)
                 finally:
                     context.detach(token)
             else:
@@ -332,6 +362,25 @@ class ServerWrapper(McpWrapper):
     # Though we can opt out of doing so as well as believe those spans provide minimum value with added
     # complexity.
 
+    def wrap_mcp_http_sse_app_factory(self, wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any) -> Any:
+        if not self._should_suppress_http_spans:
+            return wrapped(*args, **kwargs)
+
+        class _HttpSuppressionMiddleware:
+            def __init__(self, inner_app: Any) -> None:
+                self.app = inner_app
+
+            async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
+                with suppress_http_instrumentation():
+                    await self.app(scope, receive, send)
+
+        app = wrapped(*args, **kwargs)
+        try:
+            app.add_middleware(_HttpSuppressionMiddleware)
+        except Exception:  # pylint: disable=broad-exception-caught
+            _LOG.debug("Failed to apply ASGI suppression to MCP server app", exc_info=True)
+        return app
+
     async def _wrap_server_handle_request(
         self,
         wrapped: Callable[..., Coroutine[Any, Any, Any]],
@@ -396,6 +445,9 @@ async def _wrap_server_message_handler(
         if not incoming_msg:
             return await wrapped(*args, **kwargs)
 
+        if self._should_suppress_http_spans and self._should_suppress_mcp_span(incoming_msg):
+            return await wrapped(*args, **kwargs)
+
         request_id = getattr(incoming_msg, "id", None)
 
         carrier = self._extract_trace_context(incoming_msg)

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/patches/_starlette_patches.py

@@ -16,6 +16,7 @@ def _apply_starlette_instrumentation_patches() -> None:
     try:
         # pylint: disable=import-outside-toplevel
         from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
+        from opentelemetry.instrumentation.utils import is_http_instrumentation_enabled
 
         # pylint: disable=line-too-long
         # Patch to exclude http receive/send ASGI event spans from Bedrock AgentCore,
@@ -36,6 +37,17 @@ def patched_init(self, app, **kwargs):
 
             OpenTelemetryMiddleware.__init__ = patched_init
 
+            # TODO: Remove this patch once upstream PR is merged:
+            # https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4375
+            original_call = OpenTelemetryMiddleware.__call__
+
+            async def patched_call(self, scope, receive, send):
+                if not is_http_instrumentation_enabled():
+                    return await self.app(scope, receive, send)
+                return await original_call(self, scope, receive, send)
+
+            OpenTelemetryMiddleware.__call__ = patched_call
+
         _logger.debug("Successfully patched Starlette ASGI middleware")
     except Exception as exc:  # pylint: disable=broad-except
         _logger.warning("Failed to apply Starlette instrumentation patches: %s", exc)

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/instrumentation/mcp/test_mcp_instrumentor.py

@@ -3,19 +3,22 @@
 
 import asyncio
 import json
+import os
 import signal
 import socket
 import subprocess
 import sys
 import time
 import unittest
+import unittest.mock
 from pathlib import Path
 from threading import Thread
 from unittest import TestCase
 
 from collector import OTLPServer, Telemetry
 
 from amazon.opentelemetry.distro.instrumentation.mcp import McpInstrumentor
+from amazon.opentelemetry.distro.instrumentation.mcp._wrappers import OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION
 from opentelemetry.baggage.propagation import W3CBaggagePropagator
 
 try:
@@ -38,7 +41,6 @@
 )
 from opentelemetry.semconv._incubating.attributes.mcp_attributes import (
     MCP_METHOD_NAME,
-    MCP_PROTOCOL_VERSION,
     MCP_RESOURCE_URI,
     MCP_SESSION_ID,
     McpMethodNameValues,
@@ -300,36 +302,38 @@ def _assert_client_and_server_spans(
         session_span = self._get_span(client_spans, "mcp.session")
         self.assertEqual(session_span.kind, SpanKind.INTERNAL)
 
-        init_span = self._get_span(client_spans, f"mcp {McpMethodNameValues.INITIALIZE.value}")
-        self.assertIsNotNone(init_span.attributes.get(MCP_PROTOCOL_VERSION))
-
-        client_notif_init_span = self._get_span(
-            client_spans, f"mcp {McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}"
-        )
-
-        self.assertEqual(init_span.kind, SpanKind.CLIENT)
-        self._assert_span_attrs(
-            init_span, {MCP_METHOD_NAME: McpMethodNameValues.INITIALIZE.value, NETWORK_TRANSPORT: expected_transport}
-        )
-
-        self.assertEqual(client_notif_init_span.kind, SpanKind.CLIENT)
-        self._assert_span_attrs(
-            client_notif_init_span,
-            {
-                MCP_METHOD_NAME: McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value,
-                NETWORK_TRANSPORT: expected_transport,
-            },
-        )
-
-        server_init_span = self._get_span(server_spans, f"mcp {McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}")
-        self.assertEqual(server_init_span.kind, ProtoSpan.SpanKind.SPAN_KIND_SERVER)
-        self._assert_span_attrs(
-            server_init_span, {MCP_METHOD_NAME: McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}
-        )
-        self._assert_no_attr(server_init_span, NETWORK_TRANSPORT)
-        self._assert_context_propagation(client_notif_init_span, server_init_span)
-
-        client_op_spans = [init_span, client_notif_init_span]
+        # TODO: Uncomment once we get user requirement to unsuppress initialize spans.
+
+        # init_span = self._get_span(client_spans, f"mcp {McpMethodNameValues.INITIALIZE.value}")
+        # self.assertIsNotNone(init_span.attributes.get(MCP_PROTOCOL_VERSION))
+        #
+        # client_notif_init_span = self._get_span(
+        #     client_spans, f"mcp {McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}"
+        # )
+        #
+        # self.assertEqual(init_span.kind, SpanKind.CLIENT)
+        # self._assert_span_attrs(
+        #     init_span, {MCP_METHOD_NAME: McpMethodNameValues.INITIALIZE.value, NETWORK_TRANSPORT: expected_transport}
+        # )
+        #
+        # self.assertEqual(client_notif_init_span.kind, SpanKind.CLIENT)
+        # self._assert_span_attrs(
+        #     client_notif_init_span,
+        #     {
+        #         MCP_METHOD_NAME: McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value,
+        #         NETWORK_TRANSPORT: expected_transport,
+        #     },
+        # )
+        #
+        # server_init_span = self._get_span(server_spans, f"mcp {McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}")
+        # self.assertEqual(server_init_span.kind, ProtoSpan.SpanKind.SPAN_KIND_SERVER)
+        # self._assert_span_attrs(
+        #     server_init_span, {MCP_METHOD_NAME: McpMethodNameValues.NOTIFICATIONS_INITIALIZED.value}
+        # )
+        # self._assert_no_attr(server_init_span, NETWORK_TRANSPORT)
+        # self._assert_context_propagation(client_notif_init_span, server_init_span)
+
+        client_op_spans = []
 
         if operation_span_name:
             client_op_span = self._get_span(client_spans, operation_span_name)
@@ -428,30 +432,46 @@ async def run(session):
         self.assertIsNotNone(server_span.attributes.get(CLIENT_ADDRESS))
         self.assertIsNotNone(server_span.attributes.get(CLIENT_PORT))
 
-    def test_http_span_parented_under_mcp_request(self):
-
-        HTTPXClientInstrumentor().instrument(tracer_provider=self.tracer_provider)
-        try:
-
-            async def run(session):
-                await session.call_tool("hello", {"name": "World"})
-
-            asyncio.run(self._run_http_inprocess(run))
-            spans = self.span_exporter.get_finished_spans()
-
-            tool_span = next(s for s in spans if s.name == "mcp tools/call hello" and s.kind == SpanKind.CLIENT)
-            post_spans = [s for s in spans if s.name == "POST" and s.kind == SpanKind.CLIENT]
-
-            self.assertTrue(len(post_spans) > 0, "Expected at least one httpx POST span")
-
-            tool_span_id = format(tool_span.context.span_id, "016x")
-            parented_posts = [s for s in post_spans if format(s.parent.span_id, "016x") == tool_span_id]
-            self.assertTrue(
-                len(parented_posts) > 0,
-                "httpx POST span should be parented under MCP tool call span",
-            )
-        finally:
-            HTTPXClientInstrumentor().uninstrument()
+    def test_http_span_suppression(self):
+        for suppress_value, expect_post_spans in [("false", True), ("true", False), (None, False)]:
+            label = f"suppress={suppress_value}" if suppress_value else "suppress=default"
+            with self.subTest(label):
+                self.instrumentor.uninstrument()
+                self.span_exporter.clear()
+
+                patch_env = {}
+                if suppress_value is not None:
+                    patch_env[OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION] = suppress_value
+
+                with unittest.mock.patch.dict(os.environ, patch_env):
+                    if suppress_value is None:
+                        os.environ.pop(OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION, None)
+
+                    self.instrumentor.instrument(tracer_provider=self.tracer_provider, propagators=self.propagator)
+                    self.server = self._create_server()
+                    HTTPXClientInstrumentor().instrument(tracer_provider=self.tracer_provider)
+                    try:
+
+                        async def run(session):
+                            await session.call_tool("hello", {"name": "World"})
+
+                        asyncio.run(self._run_http_inprocess(run))
+                        spans = self.span_exporter.get_finished_spans()
+
+                        tool_span = next(
+                            s for s in spans if s.name == "mcp tools/call hello" and s.kind == SpanKind.CLIENT
+                        )
+                        post_spans = [s for s in spans if s.name == "POST" and s.kind == SpanKind.CLIENT]
+
+                        if expect_post_spans:
+                            self.assertTrue(len(post_spans) > 0, "Expected httpx POST spans when unsuppressed")
+                            tool_span_id = format(tool_span.context.span_id, "016x")
+                            parented_posts = [s for s in post_spans if format(s.parent.span_id, "016x") == tool_span_id]
+                            self.assertTrue(len(parented_posts) > 0, "POST span should parent under MCP tool call")
+                        else:
+                            self.assertEqual(len(post_spans), 0, "Expected no httpx POST spans when suppressed")
+                    finally:
+                        HTTPXClientInstrumentor().uninstrument()
 
     def test_mcp_respects_active_parent_span(self):
         tracer = get_tracer("test", tracer_provider=self.tracer_provider)