From c93b63a57ccdb7e42c4736750d909ffe994c8376 Mon Sep 17 00:00:00 2001 From: Ido Gol Date: Wed, 9 Sep 2026 14:37:03 +0300 Subject: [PATCH 1/4] fix(mcp): honor TRACELOOP_TRACE_CONTENT on the client path The package documents TRACELOOP_TRACE_CONTENT as the switch that turns off content logging, but only the FastMCP server-side wrapper consulted it. The MCP client path recorded content regardless: tools/call arguments and results via _extract_clean_input/_extract_clean_output, whole request and response bodies via serialize() in _handle_mcp_method and _execute_and_handle_result, and the response value in InstrumentedStreamWriter.send. An operator who set the variable to false still got request and response payloads on their spans. Move should_send_prompts() into utils so one gate serves both wrappers, and apply it to every content-bearing attribute on the client path. Span names, entity names, span kind, request ids and error status are unaffected; only content is withheld. Tests drive the real client against a FastMCP server and assert a marker value is absent from every span attribute when the switch is off and still present when it is on, so neither the gate nor the capture can regress unnoticed. --- .../mcp/fastmcp_instrumentation.py | 6 +- .../instrumentation/mcp/instrumentation.py | 41 ++++-- .../instrumentation/mcp/utils.py | 12 ++ .../tests/test_content_capture_gate.py | 120 ++++++++++++++++++ 4 files changed, 164 insertions(+), 15 deletions(-) create mode 100644 packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py index 64fdb86024..4c428b2ac7 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py @@ -9,7 +9,7 @@ from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from wrapt import register_post_import_hook, wrap_function_wrapper -from .utils import dont_throw +from .utils import dont_throw, should_send_prompts class FastMCPInstrumentor: @@ -155,9 +155,7 @@ async def traced_method(wrapped, instance, args, kwargs): def _should_send_prompts(self): """Check if content tracing is enabled (matches traceloop SDK)""" - return ( - os.getenv("TRACELOOP_TRACE_CONTENT") or "true" - ).lower() == "true" + return should_send_prompts() def _get_json_encoder(self): """Get JSON encoder class (simplified - traceloop SDK uses custom JSONEncoder)""" diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py index 688cdb2c7f..b8cfdca5f1 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py @@ -15,7 +15,11 @@ from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from opentelemetry.instrumentation.mcp.version import __version__ -from opentelemetry.instrumentation.mcp.utils import dont_throw, Config +from opentelemetry.instrumentation.mcp.utils import ( + Config, + dont_throw, + should_send_prompts, +) from opentelemetry.instrumentation.mcp.fastmcp_instrumentation import ( FastMCPInstrumentor, ) @@ -289,8 +293,13 @@ async def _handle_tool_call(self, tracer, method, params, args, kwargs, wrapped) ) span.set_attribute(SpanAttributes.TRACELOOP_ENTITY_NAME, entity_name) - # Add input - clean_input = self._extract_clean_input(method, params) + # Add input. Tool arguments are request content, so they are + # recorded only when content capture is enabled. + clean_input = ( + self._extract_clean_input(method, params) + if should_send_prompts() + else None + ) if clean_input: try: span.set_attribute( @@ -308,9 +317,12 @@ async def _handle_tool_call(self, tracer, method, params, args, kwargs, wrapped) async def _handle_mcp_method(self, tracer, method, args, kwargs, wrapped): """Handle non-tool MCP methods with simple serialization""" with tracer.start_as_current_span(f"{method}.mcp") as span: - span.set_attribute( - SpanAttributes.TRACELOOP_ENTITY_INPUT, f"{serialize(args[0])}" - ) + # The serialized request is content: it carries caller-supplied + # params, so it is recorded only when content capture is enabled. + if should_send_prompts(): + span.set_attribute( + SpanAttributes.TRACELOOP_ENTITY_INPUT, f"{serialize(args[0])}" + ) return await self._execute_and_handle_result( span, method, args, kwargs, wrapped, clean_output=False ) @@ -321,8 +333,11 @@ async def _execute_and_handle_result( """Execute the wrapped function and handle the result""" try: result = await wrapped(*args, **kwargs) - # Add output - if clean_output: + # Add output. The response body is content, so it is recorded only + # when content capture is enabled. + if not should_send_prompts(): + pass + elif clean_output: clean_output_data = self._extract_clean_output(method, result) if clean_output_data: try: @@ -565,9 +580,13 @@ async def send(self, item: Any) -> Any: with self._tracer.start_as_current_span("ResponseStreamWriter") as span: if hasattr(request, "result"): - span.set_attribute( - SpanAttributes.MCP_RESPONSE_VALUE, f"{serialize(request.result)}" - ) + # The response body is content; the error status below is not, + # so only the value itself is gated. + if should_send_prompts(): + span.set_attribute( + SpanAttributes.MCP_RESPONSE_VALUE, + f"{serialize(request.result)}", + ) if "isError" in request.result: if request.result["isError"] is True: span.set_status( diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py index d4a80e58dc..3d8cedf23b 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py @@ -2,6 +2,7 @@ import asyncio import logging +import os import traceback @@ -9,6 +10,17 @@ class Config: exception_logger = None +def should_send_prompts() -> bool: + """Whether request/response content may be recorded on spans. + + Mirrors the traceloop SDK's ``TRACELOOP_TRACE_CONTENT`` switch: content + capture is on unless an operator explicitly turns it off. Shared by the + FastMCP server wrapper and the MCP client path so a single environment + variable governs both, which is what the package README documents. + """ + return (os.getenv("TRACELOOP_TRACE_CONTENT") or "true").lower() == "true" + + def dont_throw(func): """ A decorator that wraps the passed in function and logs exceptions instead of throwing them. diff --git a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py new file mode 100644 index 0000000000..d62c572768 --- /dev/null +++ b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py @@ -0,0 +1,120 @@ +"""TRACELOOP_TRACE_CONTENT must gate the MCP client path, not only FastMCP. + +The package documents TRACELOOP_TRACE_CONTENT as the switch that disables content +logging. Before this test, only the FastMCP server-side wrapper consulted it: the +client path (tools/call arguments, non-tool request bodies, response bodies) recorded +content regardless, so an operator who turned the switch off still got request and +response payloads on their spans. + +Each test drives the real client wrapper with a marker value and asserts the marker +is absent from every span attribute when content capture is off, and present when it +is on, so the test fails if either the gate or the capture itself regresses. +""" + +import json + +from fastmcp import Client, FastMCP + +MARKER = "content-capture-marker-9f3a" + + +def _all_attribute_text(span_exporter) -> str: + """Every attribute value across every exported span, as one string.""" + chunks = [] + for span in span_exporter.get_finished_spans(): + for value in (span.attributes or {}).values(): + if isinstance(value, (list, tuple)): + chunks.extend(str(item) for item in value) + else: + chunks.append(str(value)) + return "\n".join(chunks) + + +def _server() -> FastMCP: + server = FastMCP("content-gate-server") + + @server.tool() + async def echo_secret(token: str) -> str: + """Echo back a caller-supplied token.""" + return f"received {token}" + + return server + + +async def test_tool_arguments_suppressed_when_content_capture_off( + span_exporter, monkeypatch +) -> None: + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + async with Client(_server()) as client: + await client.call_tool("echo_secret", {"token": MARKER}) + + assert span_exporter.get_finished_spans(), "expected the tool call to be traced" + assert MARKER not in _all_attribute_text(span_exporter) + + +async def test_tool_arguments_captured_when_content_capture_on( + span_exporter, monkeypatch +) -> None: + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "true") + + async with Client(_server()) as client: + await client.call_tool("echo_secret", {"token": MARKER}) + + # The gate must not silently disable capture altogether: with the switch on, + # the argument is still recorded. + assert MARKER in _all_attribute_text(span_exporter) + + +async def test_non_tool_response_body_suppressed_when_content_capture_off( + span_exporter, monkeypatch +) -> None: + """list_tools goes through _handle_mcp_method, which serialized the whole response. + + The marker lives in the registered tool description, so it travels back in the + list_tools result and exercises the response-serialization path. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + server = FastMCP("content-gate-server") + + @server.tool(description=f"A tool whose description carries {MARKER}.") + async def documented(arg: str) -> str: + return arg + + async with Client(server) as client: + tools = await client.list_tools() + + assert any(MARKER in (t.description or "") for t in tools), ( + "the marker must reach the client, otherwise this test proves nothing" + ) + assert span_exporter.get_finished_spans(), "expected the request to be traced" + assert MARKER not in _all_attribute_text(span_exporter) + + +async def test_span_structure_survives_content_capture_off( + span_exporter, monkeypatch +) -> None: + """Turning content off must not remove spans or their non-content attributes.""" + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + async with Client(_server()) as client: + await client.call_tool("echo_secret", {"token": MARKER}) + + spans = span_exporter.get_finished_spans() + tool_spans = [s for s in spans if s.name.endswith(".tool")] + assert tool_spans, f"expected a tool span, got {[s.name for s in spans]}" + + entity_names = [ + (s.attributes or {}).get("traceloop.entity.name") for s in tool_spans + ] + assert "echo_secret" in entity_names + + # Structural attributes stay; only content is withheld. + for span in tool_spans: + attributes = span.attributes or {} + assert "traceloop.span.kind" in attributes + for key in ("traceloop.entity.input", "traceloop.entity.output"): + if key in attributes: + json.loads(attributes[key]) # if present it must still be valid JSON + assert MARKER not in attributes[key] From 4ff0dd45ef4ff114b627beb0bc6bb4ed9e8c0df7 Mon Sep 17 00:00:00 2001 From: Ido Gol Date: Wed, 9 Sep 2026 15:12:27 +0300 Subject: [PATCH 2/4] docs(mcp): docstring the functions this branch touches --- .../mcp/fastmcp_instrumentation.py | 3 ++ .../instrumentation/mcp/instrumentation.py | 35 +++++++++++++++++++ .../instrumentation/mcp/utils.py | 4 +++ .../tests/test_content_capture_gate.py | 4 +++ 4 files changed, 46 insertions(+) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py index 4c428b2ac7..0adaff1abd 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py @@ -16,6 +16,7 @@ class FastMCPInstrumentor: """Handles FastMCP-specific instrumentation logic.""" def __init__(self): + """Create the instrumentor with no tracer or server name bound yet.""" self._tracer = None self._server_name = None @@ -50,6 +51,7 @@ def _fastmcp_init_wrapper(self): @dont_throw def traced_method(wrapped, instance, args, kwargs): # Call the original __init__ first + """Record the server name from FastMCP's constructor arguments.""" result = wrapped(*args, **kwargs) if args and len(args) > 0: @@ -63,6 +65,7 @@ def traced_method(wrapped, instance, args, kwargs): def _fastmcp_tool_wrapper(self): """Create wrapper for FastMCP tool execution.""" async def traced_method(wrapped, instance, args, kwargs): + """Wrap a FastMCP tool call in server and tool spans.""" if not self._tracer: return await wrapped(*args, **kwargs) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py index b8cfdca5f1..3f4366d436 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py @@ -28,15 +28,19 @@ class McpInstrumentor(BaseInstrumentor): + """Instrument the MCP client, server and transports with OpenTelemetry spans.""" def __init__(self, exception_logger=None): + """Store the exception logger and build the FastMCP sub-instrumentor.""" super().__init__() Config.exception_logger = exception_logger self._fastmcp_instrumentor = FastMCPInstrumentor() def instrumentation_dependencies(self) -> Collection[str]: + """Return the package versions this instrumentation supports.""" return _instruments def _instrument(self, **kwargs): + """Wrap the MCP client, server sessions and every supported transport.""" tracer_provider = kwargs.get("tracer_provider") tracer = get_tracer(__name__, __version__, tracer_provider) @@ -118,11 +122,13 @@ def _instrument(self, **kwargs): ) def _uninstrument(self, **kwargs): + """Unwrap the transports this instrumentation replaced.""" unwrap("mcp.client.stdio", "stdio_client") unwrap("mcp.server.stdio", "stdio_server") self._fastmcp_instrumentor.uninstrument() def _transport_wrapper(self, tracer): + """Wrap a transport so its read and write streams are instrumented.""" @asynccontextmanager async def traced_method( wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any @@ -133,6 +139,7 @@ async def traced_method( ], None, ]: + """Yield the transport's streams wrapped in instrumented proxies.""" async with wrapped(*args, **kwargs) as result: try: read_stream, write_stream = result @@ -161,9 +168,11 @@ async def traced_method( return traced_method def _base_session_init_wrapper(self, tracer): + """Wrap a server session's incoming message streams to carry trace context.""" def traced_method( wrapped: Callable[..., None], instance: Any, args: Any, kwargs: Any ) -> None: + """Replace the session's incoming stream pair with context-propagating proxies.""" wrapped(*args, **kwargs) reader = getattr(instance, "_incoming_message_stream_reader", None) writer = getattr(instance, "_incoming_message_stream_writer", None) @@ -182,8 +191,10 @@ def traced_method( return traced_method def patch_mcp_client(self, tracer: Tracer): + """Wrap BaseSession.send_request so each MCP request becomes a span.""" @dont_throw async def traced_method(wrapped, instance, args, kwargs): + """Start a span for the outgoing request and inject trace context into its meta.""" meta = None method = None params = None @@ -220,6 +231,7 @@ def _fastmcp_client_enter_wrapper(self, tracer): @dont_throw async def traced_method(wrapped, instance, args, kwargs): # Start a root span for the MCP client session and make it current + """Wrap a FastMCP client session enter to open a session span.""" span_context_manager = tracer.start_as_current_span("mcp.client.session") span = span_context_manager.__enter__() span.set_attribute(SpanAttributes.TRACELOOP_SPAN_KIND, "session") @@ -247,6 +259,7 @@ def _fastmcp_client_exit_wrapper(self, tracer): @dont_throw async def traced_method(wrapped, instance, args, kwargs): + """Close the session span when the FastMCP client exits.""" try: # Call the original method first result = await wrapped(*args, **kwargs) @@ -472,6 +485,7 @@ def serialize(request, depth=0, max_depth=4): depth += 1 def is_serializable(request): + """Return whether a value can be JSON-encoded without a fallback.""" try: json.dumps(request) return True @@ -507,18 +521,23 @@ def is_serializable(request): class InstrumentedStreamReader(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 + """Stream reader proxy that extracts trace context from incoming messages.""" def __init__(self, wrapped, tracer): + """Wrap a stream reader and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: + """Enter the wrapped stream reader.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: + """Exit the wrapped stream reader.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def __aiter__(self) -> AsyncGenerator[Any, None]: + """Iterate the wrapped reader, restoring trace context from each message.""" from mcp.types import JSONRPCMessage, JSONRPCRequest async for item in self.__wrapped__: @@ -553,18 +572,23 @@ async def __aiter__(self) -> AsyncGenerator[Any, None]: class InstrumentedStreamWriter(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 + """Stream writer proxy that records outgoing responses on a span.""" def __init__(self, wrapped, tracer): + """Wrap a stream writer and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: + """Enter the wrapped stream writer.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: + """Exit the wrapped stream writer.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def send(self, item: Any) -> Any: + """Record the outgoing response on a span and forward it to the wrapped stream.""" from mcp.types import JSONRPCMessage, JSONRPCRequest # Handle different item types based on what's available @@ -611,42 +635,53 @@ async def send(self, item: Any) -> Any: @dataclass(slots=True, frozen=True) class ItemWithContext: + """A stream item paired with the OpenTelemetry context it was written under.""" item: Any ctx: context.Context class ContextSavingStreamWriter(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 + """Stream writer proxy that attaches the current context to each item.""" def __init__(self, wrapped, tracer): + """Wrap a stream writer and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: + """Enter the wrapped stream writer.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: + """Exit the wrapped stream writer.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def send(self, item: Any) -> Any: # Removed RequestStreamWriter span creation - we don't need low-level protocol spans + """Forward the item together with the context it was sent under.""" ctx = context.get_current() return await self.__wrapped__.send(ItemWithContext(item, ctx)) class ContextAttachingStreamReader(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 + """Stream reader proxy that restores each item's saved context while it is handled.""" def __init__(self, wrapped, tracer): + """Wrap a stream reader and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: + """Enter the wrapped stream reader.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: + """Exit the wrapped stream reader.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) async def __aiter__(self) -> AsyncGenerator[Any, None]: + """Yield each item with its saved context attached, detaching it afterwards.""" async for item in self.__wrapped__: item_with_context = cast(ItemWithContext, item) restore = context.attach(item_with_context.ctx) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py index 3d8cedf23b..8968296286 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py @@ -7,6 +7,7 @@ class Config: + """Module-level configuration for the MCP instrumentation.""" exception_logger = None @@ -29,18 +30,21 @@ def dont_throw(func): logger = logging.getLogger(func.__module__) async def async_wrapper(*args, **kwargs): + """Await the wrapped coroutine, logging instead of raising on failure.""" try: return await func(*args, **kwargs) except Exception as e: _handle_exception(e, func, logger) def sync_wrapper(*args, **kwargs): + """Call the wrapped function, logging instead of raising on failure.""" try: return func(*args, **kwargs) except Exception as e: _handle_exception(e, func, logger) def _handle_exception(e, func, logger): + """Log a tracing failure and hand it to the configured exception logger.""" logger.debug( "OpenLLMetry failed to trace in %s, error: %s", func.__name__, diff --git a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py index d62c572768..d8cbde5a9a 100644 --- a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py +++ b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py @@ -31,6 +31,7 @@ def _all_attribute_text(span_exporter) -> str: def _server() -> FastMCP: + """Build a server with one tool that echoes a caller-supplied token.""" server = FastMCP("content-gate-server") @server.tool() @@ -44,6 +45,7 @@ async def echo_secret(token: str) -> str: async def test_tool_arguments_suppressed_when_content_capture_off( span_exporter, monkeypatch ) -> None: + """With the switch off, tool arguments must not appear on any span.""" monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") async with Client(_server()) as client: @@ -56,6 +58,7 @@ async def test_tool_arguments_suppressed_when_content_capture_off( async def test_tool_arguments_captured_when_content_capture_on( span_exporter, monkeypatch ) -> None: + """With the switch on, tool arguments are still recorded as before.""" monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "true") async with Client(_server()) as client: @@ -80,6 +83,7 @@ async def test_non_tool_response_body_suppressed_when_content_capture_off( @server.tool(description=f"A tool whose description carries {MARKER}.") async def documented(arg: str) -> str: + """A tool that exists only to carry the marker in its description.""" return arg async with Client(server) as client: From 4bdbabda0cc511fb4ef9aa54e2c7a9786610ae10 Mon Sep 17 00:00:00 2001 From: Ido Gol Date: Thu, 17 Sep 2026 16:08:43 +0300 Subject: [PATCH 3/4] fix(mcp): gate error text on TRACELOOP_TRACE_CONTENT too The switch reached request and response bodies but not the three places a failure puts the same text on a span: record_exception, which writes the message and the whole stacktrace as event attributes; the status description, which on the client path is the server's own error text; and OTel's automatic exception recording, which re-added both on the way out of every start_as_current_span block -- the reason each failure carried two identical exception events. Route the descriptions through one error_status helper, gate record_exception beside it, and turn off the automatic recording where this code already reports the failure itself. The error type and StatusCode.ERROR are not content and are recorded either way, so a failure stays just as visible. Also revert the docstrings the previous commit added to untouched functions. --- .../mcp/fastmcp_instrumentation.py | 29 ++-- .../instrumentation/mcp/instrumentation.py | 79 ++++------ .../instrumentation/mcp/utils.py | 18 ++- .../tests/test_content_capture_gate.py | 149 ++++++++++++++++-- 4 files changed, 199 insertions(+), 76 deletions(-) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py index 0adaff1abd..4fe04f2c33 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py @@ -9,14 +9,13 @@ from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from wrapt import register_post_import_hook, wrap_function_wrapper -from .utils import dont_throw, should_send_prompts +from .utils import dont_throw, error_status, should_send_prompts class FastMCPInstrumentor: """Handles FastMCP-specific instrumentation logic.""" def __init__(self): - """Create the instrumentor with no tracer or server name bound yet.""" self._tracer = None self._server_name = None @@ -51,7 +50,6 @@ def _fastmcp_init_wrapper(self): @dont_throw def traced_method(wrapped, instance, args, kwargs): # Call the original __init__ first - """Record the server name from FastMCP's constructor arguments.""" result = wrapped(*args, **kwargs) if args and len(args) > 0: @@ -65,7 +63,6 @@ def traced_method(wrapped, instance, args, kwargs): def _fastmcp_tool_wrapper(self): """Create wrapper for FastMCP tool execution.""" async def traced_method(wrapped, instance, args, kwargs): - """Wrap a FastMCP tool call in server and tool spans.""" if not self._tracer: return await wrapped(*args, **kwargs) @@ -85,7 +82,12 @@ async def traced_method(wrapped, instance, args, kwargs): entity_name = tool_key if tool_key else "unknown_tool" # Create parent server.mcp span - with self._tracer.start_as_current_span("mcp.server") as mcp_span: + # The error paths below record the failure themselves, with the + # message gated on content capture. OTel's own exception recording + # would re-add that text ungated on the way out. + with self._tracer.start_as_current_span( + "mcp.server", record_exception=False, set_status_on_exception=False + ) as mcp_span: mcp_span.set_attribute(SpanAttributes.TRACELOOP_SPAN_KIND, "server") mcp_span.set_attribute(SpanAttributes.TRACELOOP_ENTITY_NAME, "mcp.server") if self._server_name: @@ -93,7 +95,9 @@ async def traced_method(wrapped, instance, args, kwargs): # Create nested tool span span_name = f"{entity_name}.tool" - with self._tracer.start_as_current_span(span_name) as tool_span: + with self._tracer.start_as_current_span( + span_name, record_exception=False, set_status_on_exception=False + ) as tool_span: tool_span.set_attribute(SpanAttributes.TRACELOOP_SPAN_KIND, TraceloopSpanKindValues.TOOL.value) tool_span.set_attribute(SpanAttributes.TRACELOOP_ENTITY_NAME, entity_name) if self._server_name: @@ -115,12 +119,15 @@ async def traced_method(wrapped, instance, args, kwargs): result = await wrapped(*args, **kwargs) except Exception as e: tool_span.set_attribute(ERROR_TYPE, type(e).__name__) - tool_span.record_exception(e) - tool_span.set_status(Status(StatusCode.ERROR, str(e))) - mcp_span.set_attribute(ERROR_TYPE, type(e).__name__) - mcp_span.record_exception(e) - mcp_span.set_status(Status(StatusCode.ERROR, str(e))) + # record_exception writes the message and the full + # stacktrace as event attributes, both of which carry + # the tool's own text. + if should_send_prompts(): + tool_span.record_exception(e) + mcp_span.record_exception(e) + tool_span.set_status(error_status(str(e))) + mcp_span.set_status(error_status(str(e))) raise try: diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py index 3f4366d436..48b5417652 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py @@ -18,6 +18,7 @@ from opentelemetry.instrumentation.mcp.utils import ( Config, dont_throw, + error_status, should_send_prompts, ) from opentelemetry.instrumentation.mcp.fastmcp_instrumentation import ( @@ -28,19 +29,15 @@ class McpInstrumentor(BaseInstrumentor): - """Instrument the MCP client, server and transports with OpenTelemetry spans.""" def __init__(self, exception_logger=None): - """Store the exception logger and build the FastMCP sub-instrumentor.""" super().__init__() Config.exception_logger = exception_logger self._fastmcp_instrumentor = FastMCPInstrumentor() def instrumentation_dependencies(self) -> Collection[str]: - """Return the package versions this instrumentation supports.""" return _instruments def _instrument(self, **kwargs): - """Wrap the MCP client, server sessions and every supported transport.""" tracer_provider = kwargs.get("tracer_provider") tracer = get_tracer(__name__, __version__, tracer_provider) @@ -122,13 +119,11 @@ def _instrument(self, **kwargs): ) def _uninstrument(self, **kwargs): - """Unwrap the transports this instrumentation replaced.""" unwrap("mcp.client.stdio", "stdio_client") unwrap("mcp.server.stdio", "stdio_server") self._fastmcp_instrumentor.uninstrument() def _transport_wrapper(self, tracer): - """Wrap a transport so its read and write streams are instrumented.""" @asynccontextmanager async def traced_method( wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any @@ -139,7 +134,6 @@ async def traced_method( ], None, ]: - """Yield the transport's streams wrapped in instrumented proxies.""" async with wrapped(*args, **kwargs) as result: try: read_stream, write_stream = result @@ -168,11 +162,9 @@ async def traced_method( return traced_method def _base_session_init_wrapper(self, tracer): - """Wrap a server session's incoming message streams to carry trace context.""" def traced_method( wrapped: Callable[..., None], instance: Any, args: Any, kwargs: Any ) -> None: - """Replace the session's incoming stream pair with context-propagating proxies.""" wrapped(*args, **kwargs) reader = getattr(instance, "_incoming_message_stream_reader", None) writer = getattr(instance, "_incoming_message_stream_writer", None) @@ -191,10 +183,8 @@ def traced_method( return traced_method def patch_mcp_client(self, tracer: Tracer): - """Wrap BaseSession.send_request so each MCP request becomes a span.""" @dont_throw async def traced_method(wrapped, instance, args, kwargs): - """Start a span for the outgoing request and inject trace context into its meta.""" meta = None method = None params = None @@ -231,8 +221,11 @@ def _fastmcp_client_enter_wrapper(self, tracer): @dont_throw async def traced_method(wrapped, instance, args, kwargs): # Start a root span for the MCP client session and make it current - """Wrap a FastMCP client session enter to open a session span.""" - span_context_manager = tracer.start_as_current_span("mcp.client.session") + span_context_manager = tracer.start_as_current_span( + "mcp.client.session", + record_exception=False, + set_status_on_exception=False, + ) span = span_context_manager.__enter__() span.set_attribute(SpanAttributes.TRACELOOP_SPAN_KIND, "session") span.set_attribute( @@ -248,8 +241,9 @@ async def traced_method(wrapped, instance, args, kwargs): return result except Exception as e: span.set_attribute(ERROR_TYPE, type(e).__name__) - span.record_exception(e) - span.set_status(Status(StatusCode.ERROR, str(e))) + if should_send_prompts(): + span.record_exception(e) + span.set_status(error_status(str(e))) raise return traced_method @@ -259,7 +253,6 @@ def _fastmcp_client_exit_wrapper(self, tracer): @dont_throw async def traced_method(wrapped, instance, args, kwargs): - """Close the session span when the FastMCP client exits.""" try: # Call the original method first result = await wrapped(*args, **kwargs) @@ -299,7 +292,12 @@ async def _handle_tool_call(self, tracer, method, params, args, kwargs, wrapped) except Exception: pass - with tracer.start_as_current_span(span_name) as span: + # _execute_and_handle_result records the failure itself, with the + # message gated on content capture; OTel's own exception recording + # would re-add that text ungated on the way out. + with tracer.start_as_current_span( + span_name, record_exception=False, set_status_on_exception=False + ) as span: # Set tool-specific attributes span.set_attribute( SpanAttributes.TRACELOOP_SPAN_KIND, TraceloopSpanKindValues.TOOL.value @@ -329,7 +327,9 @@ async def _handle_tool_call(self, tracer, method, params, args, kwargs, wrapped) async def _handle_mcp_method(self, tracer, method, args, kwargs, wrapped): """Handle non-tool MCP methods with simple serialization""" - with tracer.start_as_current_span(f"{method}.mcp") as span: + with tracer.start_as_current_span( + f"{method}.mcp", record_exception=False, set_status_on_exception=False + ) as span: # The serialized request is content: it carries caller-supplied # params, so it is recorded only when content capture is enabled. if should_send_prompts(): @@ -371,16 +371,18 @@ async def _execute_and_handle_result( if hasattr(result, "isError") and result.isError: span.set_attribute(ERROR_TYPE, "tool_error") if len(result.content) > 0: - span.set_status( - Status(StatusCode.ERROR, f"{result.content[0].text}") - ) + span.set_status(error_status(f"{result.content[0].text}")) else: span.set_status(Status(StatusCode.OK)) return result except Exception as e: span.set_attribute(ERROR_TYPE, type(e).__name__) - span.record_exception(e) - span.set_status(Status(StatusCode.ERROR, str(e))) + # record_exception writes the message and the full stacktrace as + # event attributes, and on the client path that text comes from the + # server, so it is content. + if should_send_prompts(): + span.record_exception(e) + span.set_status(error_status(str(e))) raise def _extract_clean_input(self, method: str, params: Any) -> dict: @@ -485,7 +487,6 @@ def serialize(request, depth=0, max_depth=4): depth += 1 def is_serializable(request): - """Return whether a value can be JSON-encoded without a fallback.""" try: json.dumps(request) return True @@ -521,23 +522,18 @@ def is_serializable(request): class InstrumentedStreamReader(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 - """Stream reader proxy that extracts trace context from incoming messages.""" def __init__(self, wrapped, tracer): - """Wrap a stream reader and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: - """Enter the wrapped stream reader.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: - """Exit the wrapped stream reader.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def __aiter__(self) -> AsyncGenerator[Any, None]: - """Iterate the wrapped reader, restoring trace context from each message.""" from mcp.types import JSONRPCMessage, JSONRPCRequest async for item in self.__wrapped__: @@ -572,23 +568,18 @@ async def __aiter__(self) -> AsyncGenerator[Any, None]: class InstrumentedStreamWriter(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 - """Stream writer proxy that records outgoing responses on a span.""" def __init__(self, wrapped, tracer): - """Wrap a stream writer and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: - """Enter the wrapped stream writer.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: - """Exit the wrapped stream writer.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def send(self, item: Any) -> Any: - """Record the outgoing response on a span and forward it to the wrapped stream.""" from mcp.types import JSONRPCMessage, JSONRPCRequest # Handle different item types based on what's available @@ -604,8 +595,8 @@ async def send(self, item: Any) -> Any: with self._tracer.start_as_current_span("ResponseStreamWriter") as span: if hasattr(request, "result"): - # The response body is content; the error status below is not, - # so only the value itself is gated. + # The response body is content, and so is the error text + # below -- it is the same payload. if should_send_prompts(): span.set_attribute( SpanAttributes.MCP_RESPONSE_VALUE, @@ -614,10 +605,7 @@ async def send(self, item: Any) -> Any: if "isError" in request.result: if request.result["isError"] is True: span.set_status( - Status( - StatusCode.ERROR, - f"{request.result['content'][0]['text']}", - ) + error_status(f"{request.result['content'][0]['text']}") ) if hasattr(request, "id"): span.set_attribute(SpanAttributes.MCP_REQUEST_ID, f"{request.id}") @@ -635,53 +623,42 @@ async def send(self, item: Any) -> Any: @dataclass(slots=True, frozen=True) class ItemWithContext: - """A stream item paired with the OpenTelemetry context it was written under.""" item: Any ctx: context.Context class ContextSavingStreamWriter(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 - """Stream writer proxy that attaches the current context to each item.""" def __init__(self, wrapped, tracer): - """Wrap a stream writer and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: - """Enter the wrapped stream writer.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: - """Exit the wrapped stream writer.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) @dont_throw async def send(self, item: Any) -> Any: # Removed RequestStreamWriter span creation - we don't need low-level protocol spans - """Forward the item together with the context it was sent under.""" ctx = context.get_current() return await self.__wrapped__.send(ItemWithContext(item, ctx)) class ContextAttachingStreamReader(ObjectProxy): # type: ignore # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73 - """Stream reader proxy that restores each item's saved context while it is handled.""" def __init__(self, wrapped, tracer): - """Wrap a stream reader and keep the tracer used for its spans.""" super().__init__(wrapped) self._tracer = tracer async def __aenter__(self) -> Any: - """Enter the wrapped stream reader.""" return await self.__wrapped__.__aenter__() async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: - """Exit the wrapped stream reader.""" return await self.__wrapped__.__aexit__(exc_type, exc_value, traceback) async def __aiter__(self) -> AsyncGenerator[Any, None]: - """Yield each item with its saved context attached, detaching it afterwards.""" async for item in self.__wrapped__: item_with_context = cast(ItemWithContext, item) restore = context.attach(item_with_context.ctx) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py index 8968296286..bfc850afa5 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py @@ -5,9 +5,10 @@ import os import traceback +from opentelemetry.trace import Status, StatusCode + class Config: - """Module-level configuration for the MCP instrumentation.""" exception_logger = None @@ -22,6 +23,18 @@ def should_send_prompts() -> bool: return (os.getenv("TRACELOOP_TRACE_CONTENT") or "true").lower() == "true" +def error_status(description: str) -> Status: + """An ERROR status, carrying `description` only when content capture is on. + + The status code is not content, but its description is: on the client path + it is the server's own error text. Callers record the exception type + separately, so the failure stays visible either way. + """ + if should_send_prompts(): + return Status(StatusCode.ERROR, description) + return Status(StatusCode.ERROR) + + def dont_throw(func): """ A decorator that wraps the passed in function and logs exceptions instead of throwing them. @@ -30,21 +43,18 @@ def dont_throw(func): logger = logging.getLogger(func.__module__) async def async_wrapper(*args, **kwargs): - """Await the wrapped coroutine, logging instead of raising on failure.""" try: return await func(*args, **kwargs) except Exception as e: _handle_exception(e, func, logger) def sync_wrapper(*args, **kwargs): - """Call the wrapped function, logging instead of raising on failure.""" try: return func(*args, **kwargs) except Exception as e: _handle_exception(e, func, logger) def _handle_exception(e, func, logger): - """Log a tracing failure and hand it to the configured exception logger.""" logger.debug( "OpenLLMetry failed to trace in %s, error: %s", func.__name__, diff --git a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py index d8cbde5a9a..ca27b8423e 100644 --- a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py +++ b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py @@ -9,24 +9,41 @@ Each test drives the real client wrapper with a marker value and asserts the marker is absent from every span attribute when content capture is off, and present when it is on, so the test fails if either the gate or the capture itself regresses. + +"Every span attribute" includes the two places content reaches a span without being +an attribute: a status description, and the message and stacktrace that +``record_exception`` writes as event attributes. """ import json +import pytest from fastmcp import Client, FastMCP +from mcp.types import JSONRPCMessage, JSONRPCResponse +from opentelemetry.instrumentation.mcp.instrumentation import InstrumentedStreamWriter +from opentelemetry.trace import StatusCode MARKER = "content-capture-marker-9f3a" -def _all_attribute_text(span_exporter) -> str: - """Every attribute value across every exported span, as one string.""" +def _all_recorded_text(span_exporter) -> str: + """Everything an exported span carries that could hold content, as one string. + + Status descriptions and event attributes count too: ``record_exception`` + puts the message and the whole stacktrace in the latter. + """ chunks = [] for span in span_exporter.get_finished_spans(): - for value in (span.attributes or {}).values(): - if isinstance(value, (list, tuple)): - chunks.extend(str(item) for item in value) - else: - chunks.append(str(value)) + sources = [span.attributes or {}] + sources.extend(event.attributes or {} for event in span.events) + for attributes in sources: + for value in attributes.values(): + if isinstance(value, (list, tuple)): + chunks.extend(str(item) for item in value) + else: + chunks.append(str(value)) + if span.status is not None and span.status.description: + chunks.append(span.status.description) return "\n".join(chunks) @@ -52,7 +69,7 @@ async def test_tool_arguments_suppressed_when_content_capture_off( await client.call_tool("echo_secret", {"token": MARKER}) assert span_exporter.get_finished_spans(), "expected the tool call to be traced" - assert MARKER not in _all_attribute_text(span_exporter) + assert MARKER not in _all_recorded_text(span_exporter) async def test_tool_arguments_captured_when_content_capture_on( @@ -66,7 +83,7 @@ async def test_tool_arguments_captured_when_content_capture_on( # The gate must not silently disable capture altogether: with the switch on, # the argument is still recorded. - assert MARKER in _all_attribute_text(span_exporter) + assert MARKER in _all_recorded_text(span_exporter) async def test_non_tool_response_body_suppressed_when_content_capture_off( @@ -93,7 +110,7 @@ async def documented(arg: str) -> str: "the marker must reach the client, otherwise this test proves nothing" ) assert span_exporter.get_finished_spans(), "expected the request to be traced" - assert MARKER not in _all_attribute_text(span_exporter) + assert MARKER not in _all_recorded_text(span_exporter) async def test_span_structure_survives_content_capture_off( @@ -122,3 +139,115 @@ async def test_span_structure_survives_content_capture_off( if key in attributes: json.loads(attributes[key]) # if present it must still be valid JSON assert MARKER not in attributes[key] + + +def _failing_server() -> FastMCP: + """Build a server whose tool fails with the marker in its message.""" + server = FastMCP("content-gate-server") + + @server.tool() + async def boom(token: str) -> str: + """Fail with the caller's token in the exception message.""" + raise ValueError(f"failure involving {token}") + + return server + + +async def test_error_text_suppressed_when_content_capture_off( + span_exporter, monkeypatch +) -> None: + """A failure's message and stacktrace are the server's text, so they are content.""" + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + async with Client(_failing_server()) as client: + with pytest.raises(Exception): + await client.call_tool("boom", {"token": MARKER}) + + spans = span_exporter.get_finished_spans() + assert spans, "expected the failed call to be traced" + assert MARKER not in _all_recorded_text(span_exporter) + + # The failure itself stays visible; only its text is withheld. + assert any(s.status.status_code is StatusCode.ERROR for s in spans) + assert any("error.type" in (s.attributes or {}) for s in spans) + + +async def test_error_text_captured_when_content_capture_on( + span_exporter, monkeypatch +) -> None: + """With the switch on, the failure text is still recorded as before.""" + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "true") + + async with Client(_failing_server()) as client: + with pytest.raises(Exception): + await client.call_tool("boom", {"token": MARKER}) + + assert MARKER in _all_recorded_text(span_exporter) + + +async def test_non_tool_response_body_captured_when_content_capture_on( + span_exporter, monkeypatch +) -> None: + """The paired case: list_tools still serializes its response when the switch is on.""" + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "true") + + server = FastMCP("content-gate-server") + + @server.tool(description=f"A tool whose description carries {MARKER}.") + async def documented(arg: str) -> str: + """A tool that exists only to carry the marker in its description.""" + return arg + + async with Client(server) as client: + await client.list_tools() + + assert MARKER in _all_recorded_text(span_exporter) + + +class _Sink: + """Stands in for the wrapped stream, recording what was forwarded to it.""" + + def __init__(self): + """Start with nothing sent.""" + self.sent = [] + + async def send(self, item): + """Record the forwarded item.""" + self.sent.append(item) + + +def _error_response() -> JSONRPCMessage: + """A tool-error response whose payload text carries the marker.""" + return JSONRPCMessage( + JSONRPCResponse( + jsonrpc="2.0", + id=1, + result={"isError": True, "content": [{"type": "text", "text": MARKER}]}, + ) + ) + + +@pytest.mark.parametrize( + ("switch", "expected"), [("false", False), ("true", True)] +) +async def test_stream_writer_error_status_honors_the_switch( + span_exporter, tracer_provider, monkeypatch, switch, expected +) -> None: + """The stdio/SSE proxy is not reachable through the in-memory client. + + _handle_mcp_tool_call covers the FastMCP path; InstrumentedStreamWriter is the + one used for stdio and SSE, and it sets the same payload text as a status + description. Drive it directly rather than leaving it uncovered. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", switch) + sink = _Sink() + + await InstrumentedStreamWriter(sink, tracer_provider.get_tracer(__name__)).send( + _error_response() + ) + + assert sink.sent, "the item must still reach the wrapped stream" + spans = span_exporter.get_finished_spans() + assert spans, "expected the response to be traced" + assert any(s.status.status_code is StatusCode.ERROR for s in spans) + assert (MARKER in _all_recorded_text(span_exporter)) is expected From fd35087285d60ef71b8d18c1afe26f114923e350 Mon Sep 17 00:00:00 2001 From: Ido Gol Date: Fri, 18 Sep 2026 14:39:24 +0300 Subject: [PATCH 4/4] fix(mcp): keep the stack frames, drop only the text that is content Round 1 gated record_exception wholesale, which threw out exception.type and exception.stacktrace along with the message -- the stacktrace being the part you actually debug from. It also left two spans reporting nothing at all: mcp.client.session, whose exit wrapper only forwarded to a context manager whose own recording had just been turned off, and ResponseStreamWriter, which still had OTel's defaults and recorded transport failures ungated. One record_error helper now handles every error path. The exception type and the frames are recorded either way; the message is withheld when content capture is off. The frames are rendered by hand rather than by traceback.format_tb, which reproduces the message twice over: once as the final line, and once inside the source line of the raise site -- a tool raising ToolError("...") would have printed its own message there. --- .../mcp/fastmcp_instrumentation.py | 15 +- .../instrumentation/mcp/instrumentation.py | 82 +++++----- .../instrumentation/mcp/utils.py | 32 ++++ .../tests/test_content_capture_gate.py | 143 +++++++++++++++++- 4 files changed, 222 insertions(+), 50 deletions(-) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py index 4fe04f2c33..9cd7226c6a 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/fastmcp_instrumentation.py @@ -6,10 +6,9 @@ from opentelemetry.trace import Tracer from opentelemetry.trace.status import Status, StatusCode from opentelemetry.semconv_ai import SpanAttributes, TraceloopSpanKindValues -from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from wrapt import register_post_import_hook, wrap_function_wrapper -from .utils import dont_throw, error_status, should_send_prompts +from .utils import dont_throw, record_error, should_send_prompts class FastMCPInstrumentor: @@ -118,16 +117,8 @@ async def traced_method(wrapped, instance, args, kwargs): try: result = await wrapped(*args, **kwargs) except Exception as e: - tool_span.set_attribute(ERROR_TYPE, type(e).__name__) - mcp_span.set_attribute(ERROR_TYPE, type(e).__name__) - # record_exception writes the message and the full - # stacktrace as event attributes, both of which carry - # the tool's own text. - if should_send_prompts(): - tool_span.record_exception(e) - mcp_span.record_exception(e) - tool_span.set_status(error_status(str(e))) - mcp_span.set_status(error_status(str(e))) + record_error(tool_span, e) + record_error(mcp_span, e) raise try: diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py index 48b5417652..7a5aea557a 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py @@ -19,6 +19,7 @@ Config, dont_throw, error_status, + record_error, should_send_prompts, ) from opentelemetry.instrumentation.mcp.fastmcp_instrumentation import ( @@ -232,18 +233,18 @@ async def traced_method(wrapped, instance, args, kwargs): SpanAttributes.TRACELOOP_ENTITY_NAME, "mcp.client.session" ) - # Store the span context manager on the instance to properly exit it later + # Store the span context manager on the instance to properly exit it + # later, and the span itself so the exit wrapper can report a + # teardown failure on it. setattr(instance, "_tracing_session_context_manager", span_context_manager) + setattr(instance, "_tracing_session_span", span) try: # Call the original method result = await wrapped(*args, **kwargs) return result except Exception as e: - span.set_attribute(ERROR_TYPE, type(e).__name__) - if should_send_prompts(): - span.record_exception(e) - span.set_status(error_status(str(e))) + record_error(span, e) raise return traced_method @@ -266,7 +267,12 @@ async def traced_method(wrapped, instance, args, kwargs): return result except Exception as e: - # End the session span context manager with exception info + # Record the teardown failure before __exit__ ends the span -- + # the span's own exception recording is off, so nothing else + # would report it. + span = getattr(instance, "_tracing_session_span", None) + if span is not None: + record_error(span, e) context_manager = getattr( instance, "_tracing_session_context_manager", None ) @@ -376,13 +382,7 @@ async def _execute_and_handle_result( span.set_status(Status(StatusCode.OK)) return result except Exception as e: - span.set_attribute(ERROR_TYPE, type(e).__name__) - # record_exception writes the message and the full stacktrace as - # event attributes, and on the client path that text comes from the - # server, so it is content. - if should_send_prompts(): - span.record_exception(e) - span.set_status(error_status(str(e))) + record_error(span, e) raise def _extract_clean_input(self, method: str, params: Any) -> dict: @@ -593,32 +593,44 @@ async def send(self, item: Any) -> Any: else: return await self.__wrapped__.send(item) - with self._tracer.start_as_current_span("ResponseStreamWriter") as span: - if hasattr(request, "result"): - # The response body is content, and so is the error text - # below -- it is the same payload. - if should_send_prompts(): - span.set_attribute( - SpanAttributes.MCP_RESPONSE_VALUE, - f"{serialize(request.result)}", - ) - if "isError" in request.result: - if request.result["isError"] is True: - span.set_status( - error_status(f"{request.result['content'][0]['text']}") + # The wrapped send runs inside this span, so a transport failure would + # be recorded by OTel with its message and stacktrace ungated. The + # except below reports it instead, with the message gated. + with self._tracer.start_as_current_span( + "ResponseStreamWriter", + record_exception=False, + set_status_on_exception=False, + ) as span: + try: + if hasattr(request, "result"): + # The response body is content, and so is the error text + # below -- it is the same payload. + if should_send_prompts(): + span.set_attribute( + SpanAttributes.MCP_RESPONSE_VALUE, + f"{serialize(request.result)}", ) - if hasattr(request, "id"): - span.set_attribute(SpanAttributes.MCP_REQUEST_ID, f"{request.id}") + if "isError" in request.result: + if request.result["isError"] is True: + span.set_status( + error_status(f"{request.result['content'][0]['text']}") + ) + if hasattr(request, "id"): + span.set_attribute(SpanAttributes.MCP_REQUEST_ID, f"{request.id}") - if not isinstance(request, JSONRPCRequest): + if not isinstance(request, JSONRPCRequest): + return await self.__wrapped__.send(item) + meta = None + if not request.params: + request.params = {} + meta = request.params.setdefault("_meta", {}) + + propagate.get_global_textmap().inject(meta) return await self.__wrapped__.send(item) - meta = None - if not request.params: - request.params = {} - meta = request.params.setdefault("_meta", {}) - propagate.get_global_textmap().inject(meta) - return await self.__wrapped__.send(item) + except Exception as e: + record_error(span, e) + raise @dataclass(slots=True, frozen=True) diff --git a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py index bfc850afa5..6293600442 100644 --- a/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py +++ b/packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/utils.py @@ -5,6 +5,7 @@ import os import traceback +from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from opentelemetry.trace import Status, StatusCode @@ -35,6 +36,37 @@ def error_status(description: str) -> Status: return Status(StatusCode.ERROR) +def record_error(span, exc) -> None: + """Mark `span` failed, withholding only the parts of `exc` that are content. + + The exception type and the stack frames are not content and are recorded + either way -- the stacktrace is the part you debug from. The message is, + and it reaches a traceback twice over: as the last line of a formatted + one, and inside the source line of the raise site's own frame. Hence the + frames are rendered by hand, without either. + """ + span.set_attribute(ERROR_TYPE, type(exc).__name__) + if should_send_prompts(): + span.record_exception(exc) + else: + span.add_event( + "exception", + { + "exception.type": f"{type(exc).__module__}.{type(exc).__qualname__}", + # File, line and function, but not the frame's source line: a + # raise site like ToolError("...") reproduces its own message + # there, and that message is the thing being withheld. + "exception.stacktrace": "\n".join( + f' File "{frame.filename}", line {frame.lineno},' + f" in {frame.name}" + for frame in traceback.extract_tb(exc.__traceback__) + ), + "exception.escaped": False, + }, + ) + span.set_status(error_status(str(exc))) + + def dont_throw(func): """ A decorator that wraps the passed in function and logs exceptions instead of throwing them. diff --git a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py index ca27b8423e..5a05d4ed05 100644 --- a/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py +++ b/packages/opentelemetry-instrumentation-mcp/tests/test_content_capture_gate.py @@ -19,7 +19,9 @@ import pytest from fastmcp import Client, FastMCP +from fastmcp.exceptions import ToolError from mcp.types import JSONRPCMessage, JSONRPCResponse +from opentelemetry.instrumentation.mcp import McpInstrumentor from opentelemetry.instrumentation.mcp.instrumentation import InstrumentedStreamWriter from opentelemetry.trace import StatusCode @@ -207,12 +209,15 @@ async def documented(arg: str) -> str: class _Sink: """Stands in for the wrapped stream, recording what was forwarded to it.""" - def __init__(self): - """Start with nothing sent.""" + def __init__(self, failure=None): + """Start with nothing sent, optionally failing every send.""" self.sent = [] + self.failure = failure async def send(self, item): - """Record the forwarded item.""" + """Record the forwarded item, or fail the way a dead transport would.""" + if self.failure is not None: + raise self.failure self.sent.append(item) @@ -251,3 +256,135 @@ async def test_stream_writer_error_status_honors_the_switch( assert spans, "expected the response to be traced" assert any(s.status.status_code is StatusCode.ERROR for s in spans) assert (MARKER in _all_recorded_text(span_exporter)) is expected + + +def _exception_events(span_exporter) -> list: + """Every exception event recorded across the exported spans.""" + return [ + event + for span in span_exporter.get_finished_spans() + for event in span.events + if event.name == "exception" + ] + + +@pytest.mark.parametrize(("switch", "expected"), [("false", False), ("true", True)]) +async def test_transport_failure_text_honors_the_switch( + span_exporter, tracer_provider, monkeypatch, switch, expected +) -> None: + """A failing wrapped send must not put its message on the span ungated. + + The send runs inside the ResponseStreamWriter span, so OTel would have + recorded the message and stacktrace on the way out. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", switch) + sink = _Sink(failure=RuntimeError(f"transport died: {MARKER}")) + + # send() is @dont_throw, so the failure is logged rather than raised. + await InstrumentedStreamWriter(sink, tracer_provider.get_tracer(__name__)).send( + _error_response() + ) + + spans = span_exporter.get_finished_spans() + assert spans, "expected the failed write to be traced" + assert any(s.status.status_code is StatusCode.ERROR for s in spans) + assert any("error.type" in (s.attributes or {}) for s in spans) + assert (MARKER in _all_recorded_text(span_exporter)) is expected + + +@pytest.mark.parametrize(("switch", "expected"), [("false", False), ("true", True)]) +async def test_session_teardown_failure_is_recorded( + span_exporter, tracer_provider, monkeypatch, switch, expected +) -> None: + """A teardown failure must stay visible on mcp.client.session. + + The exit wrapper hands the exception to the span's context manager, whose + own recording is off, so nothing would report the failure unless the + wrapper does it itself. Driven directly: Client.__aexit__ is already + wrapped, so patching it would replace the wrapper under test. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", switch) + instrumentor = McpInstrumentor() + tracer = tracer_provider.get_tracer(__name__) + + class _Client: + """Stands in for a FastMCP client carrying the wrapper's own state.""" + + client = _Client() + + async def _ok(*args, **kwargs): + """Enter successfully, the way a healthy client would.""" + return None + + async def _explode(*args, **kwargs): + """Fail on teardown, the way a dead stdio transport does.""" + raise RuntimeError(f"teardown exploded: {MARKER}") + + await instrumentor._fastmcp_client_enter_wrapper(tracer)(_ok, client, (), {}) + await instrumentor._fastmcp_client_exit_wrapper(tracer)(_explode, client, (), {}) + + session_spans = [ + s for s in span_exporter.get_finished_spans() if s.name == "mcp.client.session" + ] + assert session_spans, "expected the session span to be exported" + assert all(s.status.status_code is StatusCode.ERROR for s in session_spans) + assert (MARKER in _all_recorded_text(span_exporter)) is expected + + +async def test_stack_frames_survive_content_capture_off( + span_exporter, monkeypatch +) -> None: + """Withholding the message must not cost the stacktrace, which is not content. + + A fully formatted traceback would not do: its last line repeats the + message, so only the frames can be recorded. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + async with Client(_failing_server()) as client: + with pytest.raises(Exception): + await client.call_tool("boom", {"token": MARKER}) + + events = _exception_events(span_exporter) + assert events, "the failure must still be recorded as an exception event" + + for event in events: + attributes = event.attributes or {} + assert attributes.get("exception.type"), "the type is not content" + assert attributes.get("exception.stacktrace"), "the frames are not content" + assert MARKER not in str(attributes.get("exception.stacktrace")) + assert MARKER not in str(attributes.get("exception.message", "")) + + +def _literal_failing_server() -> FastMCP: + """A tool whose failure message is written literally at the raise site.""" + server = FastMCP("content-gate-server") + + @server.tool() + async def boom_literal() -> str: + """Fail the way a FastMCP tool usually does: ToolError, hardcoded. + + ToolError is not re-wrapped by fastmcp's tool manager, so this frame + stays in the traceback -- source line and all. + """ + raise ToolError("content-capture-marker-9f3a from a source literal") + + return server + + +async def test_source_literals_do_not_leak_through_the_stacktrace( + span_exporter, monkeypatch +) -> None: + """A raise site's message also lives in its frame's source line. + + The interpolated marker the other tests use never appears in source, so + only a literal one exercises this. + """ + monkeypatch.setenv("TRACELOOP_TRACE_CONTENT", "false") + + async with Client(_literal_failing_server()) as client: + with pytest.raises(Exception): + await client.call_tool("boom_literal", {}) + + assert _exception_events(span_exporter), "the failure must still be recorded" + assert MARKER not in _all_recorded_text(span_exporter)