diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index ab333912d2..1ba061b4e2 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/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() diff --git a/tests/shared/test_otel.py b/tests/shared/test_otel.py index 51bece1fd6..a26d2fd40c 100644 --- a/tests/shared/test_otel.py +++ b/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)