fix: support mcp 2.x result shapes in MCP instrumentation

.github/workflows/ci.yml

@@ -38,6 +38,21 @@ jobs:
       - run: uv sync --locked --group dev
       - run: uv run pytest
 
+  # The mcp extra is unbounded (mcp>=1.0.0) and the two majors have different
+  # result shapes (snake_case + InputRequiredResult in 2.x, GLA2-300); the
+  # default suite exercises the locked 1.x, this job re-runs the MCP
+  # instrumentation tests with 2.x overlaid so a future rename fails a build
+  # instead of silently un-flagging error results.
+  mcp-v2:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7.0.1
+      - uses: astral-sh/setup-uv@v7
+        with:
+          enable-cache: true
+      - run: uv sync --locked --group dev
+      - run: uv run --with "mcp>=2,<3" pytest tests/test_mcp_instrumentation.py
+
   # Every instrumentation.REGISTRY entry must resolve against its real package —
   # enable_instrumentations() swallows ImportError, so a typo'd module/class name
   # would otherwise ship silently.

src/glassflow/instrumentation_mcp.py

@@ -27,16 +27,23 @@
 from .semconv import (
     GEN_AI_TOOL_NAME,
     INPUT_VALUE,
+    MCP_RESULT_TYPE,
     OUTPUT_VALUE,
     TRACER_NAME,
     SpanKind,
     set_span_kind,
 )
 
+# mcp 2.x (spec 2026-07-28) renamed CallToolResult's fields to snake_case and
+# dropped the camelCase attributes entirely; reads must try both spellings to
+# stay correct on whichever major the host application has installed.
+
 
 def _serialize_result(result: Any) -> str:
     """Best-effort serialization of a CallToolResult."""
-    structured = getattr(result, "structuredContent", None)
+    structured = getattr(result, "structured_content", None)
+    if structured is None:
+        structured = getattr(result, "structuredContent", None)  # mcp 1.x
     if structured is not None:
         return serialize(structured)
     content = getattr(result, "content", None)
@@ -47,6 +54,28 @@ def _serialize_result(result: Any) -> str:
     return serialize(result)
 
 
+def _result_is_error(result: Any) -> bool:
+    """The MCP error-result flag, on either major (never raises)."""
+    return bool(getattr(result, "is_error", getattr(result, "isError", False)))
+
+
+def _record_result(span: Any, result: Any) -> None:
+    """Record a tools/call result on the span, on either mcp major.
+
+    An interim ``InputRequiredResult`` (MRTR, mcp 2.x) is NOT the tool's
+    output: its ``input_requests`` carry elicitation/sampling content, so
+    recording them as ``output.value`` would leak conversation content into
+    a tool span. Interim rounds get only the ``mcp.result_type`` marker; the
+    final round of the retry loop records output as usual.
+    """
+    if getattr(result, "result_type", None) == "input_required":
+        span.set_attribute(MCP_RESULT_TYPE, "input_required")
+        return
+    span.set_attribute(OUTPUT_VALUE, _serialize_result(result))
+    if _result_is_error(result):
+        span.set_status(Status(StatusCode.ERROR, "tool returned an error result"))
+
+
 class MCPInstrumentor:
     """Duck-types the OTel instrumentor interface (instrument/uninstrument)."""
 
@@ -93,9 +122,7 @@ async def instrumented_call_tool(
                     span.record_exception(exc)
                     span.set_status(Status(StatusCode.ERROR, str(exc)))
                     raise
-                span.set_attribute(OUTPUT_VALUE, _serialize_result(result))
-                if getattr(result, "isError", False):
-                    span.set_status(Status(StatusCode.ERROR, "tool returned an error result"))
+                _record_result(span, result)
                 return result
 
         setattr(ClientSession, "call_tool", instrumented_call_tool)  # noqa: B010

src/glassflow/semconv.py

@@ -31,6 +31,10 @@
 GEN_AI_OUTPUT_MESSAGES = "gen_ai.output.messages"
 GEN_AI_RESPONSE_FINISH_REASONS = "gen_ai.response.finish_reasons"
 GEN_AI_TOOL_NAME = "gen_ai.tool.name"
+# MCP spec 2026-07-28: a tools/call round can end with an interim
+# "input_required" result (MRTR) instead of a final one. Set ONLY on interim
+# rounds; the key follows the mcp SDK's own `mcp.*` attribute namespace.
+MCP_RESULT_TYPE = "mcp.result_type"
 GEN_AI_REQUEST_PREFIX = "gen_ai.request."
 
 # --- Span event names ---

tests/test_mcp_instrumentation.py

@@ -1,9 +1,16 @@
-"""First-class MCP tool-call spans (client side: ClientSession.call_tool)."""
+"""First-class MCP tool-call spans (client side: ClientSession.call_tool).
+
+Runs against BOTH mcp majors: the default suite exercises the locked 1.x,
+and the ci `mcp-v2` job re-runs this module against ``mcp>=2`` (spec
+2026-07-28), whose ``CallToolResult`` renamed its fields to snake_case and
+whose tool calls can return interim ``InputRequiredResult``s (GLA2-300).
+"""
 
 from __future__ import annotations
 
 import asyncio
 import json
+from contextlib import asynccontextmanager
 from typing import Any
 
 import pytest
@@ -12,16 +19,22 @@
 
 pytest.importorskip("mcp")
 
-from mcp.server.fastmcp import FastMCP  # noqa: E402
-from mcp.shared.memory import create_connected_server_and_client_session  # noqa: E402
+try:  # mcp >= 2
+    from mcp.server import MCPServer
+
+    MCP_V2 = True
+except ImportError:  # mcp 1.x: FastMCP is the same decorator surface
+    from mcp.server.fastmcp import FastMCP as MCPServer  # type: ignore[assignment]
+
+    MCP_V2 = False
 
 from glassflow import init  # noqa: E402
 from glassflow.instrumentation import REGISTRY  # noqa: E402
 from glassflow.instrumentation_mcp import MCPInstrumentor  # noqa: E402
 
 
-def _make_server() -> FastMCP:
-    server = FastMCP("test-server")
+def _make_server() -> Any:
+    server = MCPServer("test-server")
 
     @server.tool()
     def add(a: int, b: int) -> int:
@@ -36,6 +49,26 @@ def boom() -> str:
     return server
 
 
+@asynccontextmanager
+async def _connected_session(server: Any) -> Any:
+    """Yield a live ClientSession against an in-memory server, on either major."""
+    if MCP_V2:
+        from mcp import Client
+
+        async with Client(server) as client:
+            yield client.session
+    else:
+        from mcp.shared.memory import create_connected_server_and_client_session
+
+        async with create_connected_server_and_client_session(server._mcp_server) as session:
+            yield session
+
+
+def _result_error_flag(result: Any) -> Any:
+    """Version-agnostic read of the result error flag, for assertions."""
+    return getattr(result, "is_error", getattr(result, "isError", None))
+
+
 @pytest.fixture(autouse=True)
 def _fresh_mcp_instrumentor() -> Any:
     instrumentor = MCPInstrumentor()
@@ -56,7 +89,7 @@ def _run_tool_call(
 
     async def scenario() -> Any:
         server = _make_server()
-        async with create_connected_server_and_client_session(server._mcp_server) as session:
+        async with _connected_session(server) as session:
             return await session.call_tool(tool, arguments)
 
     result = asyncio.run(scenario())
@@ -83,7 +116,8 @@ def test_call_tool_creates_tool_span() -> None:
 
 def test_tool_error_result_marks_span_error() -> None:
     spans, result = _run_tool_call("boom", None)
-    assert result.isError  # FastMCP converts tool exceptions into error results
+    # the server converts tool exceptions into error results on both majors
+    assert _result_error_flag(result)
     (tool_span,) = [s for s in spans if s.name == "execute_tool boom"]
     assert not tool_span.status.is_ok
 
@@ -95,7 +129,7 @@ def test_tool_span_nests_under_current_span() -> None:
     async def scenario() -> None:
         server = _make_server()
         with client.get_tracer().start_as_current_span("agent-step"):
-            async with create_connected_server_and_client_session(server._mcp_server) as session:
+            async with _connected_session(server) as session:
                 await session.call_tool("add", {"a": 1, "b": 1})
 
     asyncio.run(scenario())
@@ -127,9 +161,110 @@ def test_uninstrument_restores_call_tool() -> None:
 
     async def scenario() -> None:
         server = _make_server()
-        async with create_connected_server_and_client_session(server._mcp_server) as session:
+        async with _connected_session(server) as session:
             await session.call_tool("add", {"a": 1, "b": 2})
 
     asyncio.run(scenario())
     client.flush()
     assert not any(s.name.startswith("execute_tool") for s in inner.get_finished_spans())
+
+
+# --- Result-shape compatibility (GLA2-300) -------------------------------
+# mcp 2.x renamed CallToolResult's fields to snake_case (isError -> is_error,
+# structuredContent -> structured_content) and added result_type; tool calls
+# can return an interim InputRequiredResult (result_type "input_required")
+# whose input_requests must never be recorded as tool output. The fakes below
+# mimic each major's exact attribute surface, so these tests pin the compat
+# behavior regardless of which mcp is installed.
+
+
+class _V1Result:
+    """Attribute surface of mcp 1.x CallToolResult (camelCase)."""
+
+    def __init__(
+        self,
+        *,
+        isError: bool = False,
+        structuredContent: Any = None,
+        content: Any = None,
+    ) -> None:
+        self.isError = isError
+        self.structuredContent = structuredContent
+        self.content = content
+
+
+class _V2Result:
+    """Attribute surface of mcp 2.x CallToolResult (snake_case + result_type)."""
+
+    def __init__(
+        self,
+        *,
+        is_error: bool = False,
+        structured_content: Any = None,
+        content: Any = None,
+        result_type: str = "complete",
+    ) -> None:
+        self.is_error = is_error
+        self.structured_content = structured_content
+        self.content = content
+        self.result_type = result_type
+
+
+class _V2InputRequired:
+    """Attribute surface of mcp 2.x InputRequiredResult (no error/content fields)."""
+
+    def __init__(self) -> None:
+        self.result_type = "input_required"
+        self.input_requests = [{"type": "elicitation", "message": "which account?"}]
+        self.request_state = "opaque-token"
+
+
+def _record_on_fresh_span(result: Any) -> ReadableSpan:
+    from glassflow.instrumentation_mcp import _record_result
+
+    inner = InMemorySpanExporter()
+    client = init(span_exporter=inner, set_global=False)
+    span = client.get_tracer().start_span("execute_tool fake")
+    _record_result(span, result)
+    span.end()
+    client.flush()
+    (finished,) = inner.get_finished_spans()
+    return finished
+
+
+def test_v2_error_flag_marks_span_error() -> None:
+    span = _record_on_fresh_span(_V2Result(is_error=True))
+    assert not span.status.is_ok
+
+
+def test_v1_error_flag_still_marks_span_error() -> None:
+    span = _record_on_fresh_span(_V1Result(isError=True))
+    assert not span.status.is_ok
+
+
+def test_v2_structured_content_is_recorded_as_output() -> None:
+    span = _record_on_fresh_span(_V2Result(structured_content={"result": 5}))
+    assert span.attributes is not None
+    assert json.loads(span.attributes["output.value"]) == {"result": 5}
+
+
+def test_v1_structured_content_is_still_recorded_as_output() -> None:
+    span = _record_on_fresh_span(_V1Result(structuredContent={"result": 5}))
+    assert span.attributes is not None
+    assert json.loads(span.attributes["output.value"]) == {"result": 5}
+
+
+def test_input_required_round_is_marked_and_records_no_output() -> None:
+    span = _record_on_fresh_span(_V2InputRequired())
+    assert span.attributes is not None
+    # the interim payload (input_requests) is a content surface that is NOT
+    # the tool's output; it must never land in output.value
+    assert "output.value" not in span.attributes
+    assert span.attributes["mcp.result_type"] == "input_required"
+    assert span.status.is_ok
+
+
+def test_complete_result_carries_no_result_type_attribute() -> None:
+    span = _record_on_fresh_span(_V2Result(structured_content={"ok": True}))
+    assert span.attributes is not None
+    assert "mcp.result_type" not in span.attributes