fix(mcp): tolerate non-dict result shapes in InstrumentedStreamWriter.send (#4038)

packages/opentelemetry-instrumentation-mcp/opentelemetry/instrumentation/mcp/instrumentation.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping, Sequence
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from typing import Any, AsyncGenerator, Callable, Collection, Tuple, Union, cast
@@ -490,6 +491,24 @@ def is_serializable(request):
         return json.dumps(result)
 
 
+def _read_field(payload: Any, name: str) -> Any:
+    """Read a field from an MCP payload that may be dict- or object-shaped."""
+    if isinstance(payload, Mapping):
+        return payload.get(name)
+    return getattr(payload, name, None)
+
+
+def _error_description(result: Any) -> str:
+    """Best-effort text of an isError result, for both dict- and object-shaped content."""
+    content = _read_field(result, "content")
+    if isinstance(content, Sequence) and not isinstance(content, (str, bytes)):
+        for block in content:
+            text = _read_field(block, "text")
+            if text:
+                return f"{text}"
+    return ""
+
+
 class InstrumentedStreamReader(ObjectProxy):  # type: ignore
     # ObjectProxy missing context manager - https://github.com/GrahamDumpleton/wrapt/issues/73
     def __init__(self, wrapped, tracer):
@@ -568,14 +587,13 @@ async def send(self, item: Any) -> Any:
                 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(
-                            Status(
-                                StatusCode.ERROR,
-                                f"{request.result['content'][0]['text']}",
-                            )
+                if _read_field(request.result, "isError") is True:
+                    span.set_status(
+                        Status(
+                            StatusCode.ERROR,
+                            _error_description(request.result),
                         )
+                    )
             if hasattr(request, "id"):
                 span.set_attribute(SpanAttributes.MCP_REQUEST_ID, f"{request.id}")
 

packages/opentelemetry-instrumentation-mcp/tests/test_stream_writer_result_shapes.py

@@ -0,0 +1,100 @@
+"""Tests for InstrumentedStreamWriter.send() result handling (issue #4038).
+
+send() is wrapped in @dont_throw, so anything that raises while inspecting
+request.result is swallowed and the wrapped send() is never awaited -- the MCP
+message is dropped. These tests pin both halves: the message is always
+forwarded, and the ERROR status is still set for isError results.
+"""
+import pytest
+from opentelemetry.trace.status import StatusCode
+
+from opentelemetry.instrumentation.mcp.instrumentation import InstrumentedStreamWriter
+
+
+class FakeRoot:
+    """Stands in for JSONRPCMessage.root; only .result and .id are read here."""
+
+    def __init__(self, result):
+        self.result = result
+        self.id = 1
+
+
+class FakeItem:
+    def __init__(self, result):
+        self.root = FakeRoot(result)
+
+
+class FakeStream:
+    def __init__(self):
+        self.sent = []
+
+    async def send(self, item):
+        self.sent.append(item)
+        return "sent"
+
+
+class ObjectBlock:
+    type = "text"
+    text = "object-shaped failure"
+
+
+class ObjectResult:
+    """An MCP result exposing fields as attributes rather than dict keys."""
+
+    isError = True
+    content = [ObjectBlock()]
+
+
+@pytest.mark.parametrize(
+    "result,expected_description",
+    [
+        # first content block is an image, so there is no "text" key
+        (
+            {
+                "isError": True,
+                "content": [
+                    {"type": "image", "data": "iVBOR", "mimeType": "image/png"}
+                ],
+            },
+            "",
+        ),
+        # error result carrying no content blocks at all
+        ({"isError": True, "content": []}, ""),
+        # error result with no "content" key
+        ({"isError": True}, ""),
+        # object-shaped result: `"isError" in result` would raise TypeError
+        (ObjectResult(), "object-shaped failure"),
+        # the already-working shape, to pin that it is unchanged
+        (
+            {"isError": True, "content": [{"type": "text", "text": "tool blew up"}]},
+            "tool blew up",
+        ),
+    ],
+)
+async def test_error_results_are_forwarded_and_marked(
+    result, expected_description, span_exporter, tracer_provider
+):
+    stream = FakeStream()
+    writer = InstrumentedStreamWriter(stream, tracer_provider.get_tracer(__name__))
+
+    returned = await writer.send(FakeItem(result))
+
+    assert returned == "sent", "send() swallowed an error and dropped the message"
+    assert len(stream.sent) == 1
+
+    span = span_exporter.get_finished_spans()[-1]
+    assert span.status.status_code == StatusCode.ERROR
+    assert span.status.description == expected_description
+
+
+async def test_non_error_result_leaves_status_unset(span_exporter, tracer_provider):
+    stream = FakeStream()
+    writer = InstrumentedStreamWriter(stream, tracer_provider.get_tracer(__name__))
+
+    returned = await writer.send(
+        FakeItem({"content": [{"type": "text", "text": "fine"}]})
+    )
+
+    assert returned == "sent"
+    span = span_exporter.get_finished_spans()[-1]
+    assert span.status.status_code == StatusCode.UNSET