fix(mcp-instrumentation) Suppress MCP /ping spans when agent observability is enabled (#748)

- Suppress noisy MCP `/ping` spans when
`AGENT_OBSERVABILITY_ENABLED=true`
- Ping spans still emitted when agent observability is disabled 
- Added `test_ping_span_suppression` test with subtests for both
enabled/disabled cases

Author: liustve

CHANGELOG.md

@@ -12,6 +12,8 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ## Unreleased
 
+- fix(mcp-instrumentation): suppress MCP `/ping` spans when agent observability is enabled
+  ([#748](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/748))
 - fix(agent-observability): fall back to OTEL_EXPORTER_OTLP_ENDPOINT for unsampled spans; also export unsampled spans to non-AWS endpoints
   ([#738](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/738))
 - feat: auto-detect and mutually exclude AWS native vs third-party agentic instrumentors; add `AWS_AGENTIC_INSTRUMENTATION_OPT_IN` env var to override auto-detection

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

@@ -8,6 +8,7 @@
 from typing import Any, Callable, Coroutine, Dict, Optional, Tuple
 from urllib.parse import urlparse
 
+from amazon.opentelemetry.distro._utils import is_agent_observability_enabled
 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
@@ -60,17 +61,22 @@ def __init__(self, tracer: trace.Tracer, **kwargs: Any) -> None:
         self._should_suppress_http_spans = (
             os.environ.get(OTEL_MCP_SUPPRESS_HTTP_INSTRUMENTATION, "true").lower() == "true"
         )
+        self._agent_observability_enabled = is_agent_observability_enabled()
 
-    @staticmethod
-    def _should_suppress_mcp_span(message: Any) -> bool:
+    def _should_suppress_mcp_span(self, 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))
+        if isinstance(message, (types.InitializeRequest, types.InitializedNotification)):
+            return True
+        # MCP servers hosted on AgentCore get frequent /ping health checks
+        if self._agent_observability_enabled and isinstance(message, types.PingRequest):
+            return True
+        return False
 
     @staticmethod
     def _set_mcp_attributes(span: trace.Span, message: Any, request_id: Optional[int]) -> None:

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

@@ -473,6 +473,27 @@ async def run(session):
                     finally:
                         HTTPXClientInstrumentor().uninstrument()
 
+    def test_ping_span_suppression(self):
+        for agent_obs_enabled, expect_suppressed in [("true", True), ("false", False)]:
+            with self.subTest(agent_observability=agent_obs_enabled):
+                self.instrumentor.uninstrument()
+                self.span_exporter.clear()
+                with unittest.mock.patch.dict(os.environ, {"AGENT_OBSERVABILITY_ENABLED": agent_obs_enabled}):
+                    self.instrumentor.instrument(tracer_provider=self.tracer_provider, propagators=self.propagator)
+                    self.server = self._create_server()
+
+                    async def run(session):
+                        await session.send_ping()
+
+                    asyncio.run(self._run_inprocess(run))
+                    spans = self.span_exporter.get_finished_spans()
+
+                    ping_spans = [s for s in spans if "ping" in s.name.lower()]
+                    if expect_suppressed:
+                        self.assertEqual(len(ping_spans), 0, "Ping spans should be suppressed")
+                    else:
+                        self.assertGreater(len(ping_spans), 0, "Ping spans should not be suppressed")
+
     def test_mcp_respects_active_parent_span(self):
         tracer = get_tracer("test", tracer_provider=self.tracer_provider)