Record JSON-RPC error responses on the client OpenTelemetry span

src/mcp/shared/jsonrpc_dispatcher.py

@@ -32,7 +32,7 @@
     ProgressToken,
     RequestId,
 )
-from opentelemetry.trace import SpanKind
+from opentelemetry.trace import SpanKind, StatusCode
 from pydantic import ValidationError
 from typing_extensions import TypeVar
 
@@ -76,6 +76,9 @@
 _SHUTDOWN_WRITE_TIMEOUT: float = 1
 """Tighter bound for the shutdown-arm error write so a wedged transport can't hold session close."""
 
+_CLOSED_OUTCOME = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
+"""What a waiter receives when the connection closes; matched by identity, since a peer can send the same code."""
+
 TransportT = TypeVar("TransportT", bound=TransportContext, default=TransportContext)
 
 PeerCancelMode = Literal["interrupt", "signal"]
@@ -384,7 +387,7 @@ async def send_raw_request(
                 span_name,
                 kind=SpanKind.CLIENT,
                 attributes={"mcp.method.name": method, "jsonrpc.request.id": str(request_id)},
-            ):
+            ) as span:
                 # SEP-414: inject W3C trace context.
                 inject_trace_context(out_meta)
                 if out_meta:
@@ -409,6 +412,10 @@ async def send_raw_request(
                 with anyio.fail_after(opts.get("timeout")):
                     timeout_armed = True
                     outcome = await receive.receive()
+                if isinstance(outcome, ErrorData) and outcome is not _CLOSED_OUTCOME:
+                    code = str(outcome.code)
+                    span.set_attributes({"error.type": code, "rpc.response.status_code": code})
+                    span.set_status(StatusCode.ERROR, outcome.message)
         except TimeoutError:
             if not timeout_armed:
                 # `fail_after` arms only after the write, so this TimeoutError is the
@@ -702,10 +709,9 @@ def _fan_out_closed(self) -> None:
 
         Synchronous: callers may be inside a cancelled scope. Idempotent.
         """
-        closed = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
         for pending in self._pending.values():
             try:
-                pending.send.send_nowait(closed)
+                pending.send.send_nowait(_CLOSED_OUTCOME)
             except (anyio.WouldBlock, anyio.BrokenResourceError, anyio.ClosedResourceError):
                 pass
         self._pending.clear()

tests/shared/test_otel.py

@@ -3,10 +3,13 @@
 import mcp_types as types
 import pytest
 from logfire.testing import CaptureLogfire
+from opentelemetry.trace import StatusCode
 
 from mcp.client.client import Client
+from mcp.server import Server, ServerRequestContext
 from mcp.server.mcpserver import MCPServer
 from mcp.shared._otel import extract_trace_context
+from mcp.shared.exceptions import MCPError
 
 pytestmark = pytest.mark.anyio
 
@@ -45,3 +48,46 @@ def greet(name: str) -> str:
 
     # Server span should be in the same trace as the client span (context propagation).
     assert server_span["context"]["trace_id"] == client_span["context"]["trace_id"]
+
+
+async def test_client_span_records_error_status_when_peer_answers_with_jsonrpc_error(capfire: CaptureLogfire):
+    """A request the peer answers with a JSON-RPC error ends its client span with status ERROR, the peer's
+    message as the description, and the code in `error.type` and `rpc.response.status_code` (OpenTelemetry
+    MCP semantic conventions, client span). A request that succeeds stays UNSET.
+    """
+    message = "unknown cursor"
+
+    async def list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=[])
+
+    async def list_prompts(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListPromptsResult:
+        raise MCPError(types.INVALID_PARAMS, message)
+
+    server = Server("test", on_list_tools=list_tools, on_list_prompts=list_prompts)
+
+    async with Client(server, mode="legacy") as client:
+        await client.list_tools()
+        with pytest.raises(MCPError) as exc_info:
+            await client.list_prompts()
+
+    assert exc_info.value.error.code == types.INVALID_PARAMS
+
+    # logfire also exports a `pending_span` marker under each span's name when it starts.
+    spans = {
+        span.name: span
+        for span in capfire.exporter.exported_spans
+        if (span.attributes or {}).get("logfire.span_type") != "pending_span"
+    }
+    succeeded = spans["MCP send tools/list"]
+    failed = spans["MCP send prompts/list"]
+
+    assert succeeded.status.status_code == StatusCode.UNSET
+    assert failed.status.status_code == StatusCode.ERROR
+    assert failed.status.description == message
+    assert failed.attributes is not None
+    assert failed.attributes["error.type"] == str(types.INVALID_PARAMS)
+    assert failed.attributes["rpc.response.status_code"] == str(types.INVALID_PARAMS)