diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py index 6a3ad12111..47be7b577e 100644 --- a/src/mcp/client/stdio.py +++ b/src/mcp/client/stdio.py @@ -34,6 +34,7 @@ terminate_windows_process_tree, ) from mcp.shared.message import SessionMessage +from mcp.shared.wire import serialize_outbound_message logger = logging.getLogger(__name__) @@ -172,7 +173,25 @@ async def stdin_writer() -> None: try: async with write_stream_reader: async for session_message in write_stream_reader: - json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True) + # Contain serialization failures so one bad outbound message + # cannot tear down the client session (#3707). + json = serialize_outbound_message(session_message.message) + if json is None: + message = session_message.message + if isinstance(message, types.JSONRPCRequest): + error = SessionMessage( + types.JSONRPCError( + jsonrpc="2.0", + id=message.id, + error=types.ErrorData( + code=types.INTERNAL_ERROR, + message="Failed to serialize request", + ), + ) + ) + with suppress(anyio.ClosedResourceError, anyio.BrokenResourceError): + await read_stream_writer.send(error) + continue data = (json + "\n").encode(encoding=server.encoding, errors=server.encoding_error_handler) await process.stdin.send(data) except (anyio.ClosedResourceError, anyio.BrokenResourceError, OSError): diff --git a/src/mcp/server/sse.py b/src/mcp/server/sse.py index d71ef25004..63976c957d 100644 --- a/src/mcp/server/sse.py +++ b/src/mcp/server/sse.py @@ -59,6 +59,7 @@ async def handle_sse(request): ) from mcp.shared._context_streams import ContextSendStream, create_context_streams from mcp.shared.message import ServerMessageMetadata, SessionMessage +from mcp.shared.wire import serialize_outbound_message logger = logging.getLogger(__name__) @@ -184,10 +185,13 @@ async def sse_writer(): async for session_message in write_stream_reader: logger.debug(f"Sending message via SSE: {session_message}") + data = serialize_outbound_message(session_message.message) + if data is None: + continue await sse_stream_writer.send( { "event": "message", - "data": session_message.message.model_dump_json(by_alias=True, exclude_unset=True), + "data": data, } ) diff --git a/src/mcp/server/stdio.py b/src/mcp/server/stdio.py index de8bbae5f1..9ea1aa2f37 100644 --- a/src/mcp/server/stdio.py +++ b/src/mcp/server/stdio.py @@ -27,6 +27,7 @@ async def run_server(): from mcp.os.win32.utilities import rebind_std_handle_to_fd from mcp.shared._context_streams import create_context_streams from mcp.shared.message import SessionMessage +from mcp.shared.wire import serialize_outbound_message if sys.platform != "win32": # pragma: no branch import fcntl # pragma: lax no cover - POSIX-only line, uncovered on Windows runners @@ -200,7 +201,12 @@ async def stdout_writer(): try: async with write_stream_reader: async for session_message in write_stream_reader: - json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True) + # One unserializable message (e.g. a lone surrogate) must not + # end the writer; other bad values already become Internal + # server error higher up (#3707). + json = serialize_outbound_message(session_message.message) + if json is None: + continue await stdout.write(json + "\n") await stdout.flush() except anyio.ClosedResourceError: # pragma: no cover diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 416dd9e2b4..04bffc1ab3 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -46,6 +46,7 @@ from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER from mcp.shared.message import CloseSSEStreamCallback, ServerMessageMetadata, SessionMessage +from mcp.shared.wire import serialize_jsonrpc_message, serialize_outbound_message logger = logging.getLogger(__name__) @@ -366,7 +367,10 @@ async def _run_sse_writer( if priming_event is not None: await sse_stream_writer.send(priming_event) async for event_message in request_stream_reader: - await sse_stream_writer.send(self._create_event_data(event_message)) + event_data = self._create_event_data(event_message) + if event_data is None: + continue + await sse_stream_writer.send(event_data) if isinstance(event_message.message, JSONRPCResponse | JSONRPCError): break except anyio.ClosedResourceError: # pragma: lax no cover @@ -401,7 +405,7 @@ def _create_error_response( ) return Response( - error_response.model_dump_json(by_alias=True, exclude_unset=True), + serialize_jsonrpc_message(error_response), status_code=status_code, headers=response_headers, ) @@ -420,8 +424,21 @@ def _create_json_response( if self.mcp_session_id: response_headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id + body = None + if response_message is not None: + body = serialize_outbound_message(response_message) + if body is None: + # Request/notification with no id fallback: still answer the HTTP + # request so the session is not left without a body. + body = serialize_jsonrpc_message( + JSONRPCError( + jsonrpc="2.0", + id=None, + error=ErrorData(code=INTERNAL_ERROR, message="Internal server error"), + ) + ) return Response( - response_message.model_dump_json(by_alias=True, exclude_unset=True) if response_message else None, + body, status_code=status_code, headers=response_headers, ) @@ -430,11 +447,19 @@ def _get_session_id(self, request: Request) -> str | None: """Extract the session ID from request headers.""" return request.headers.get(MCP_SESSION_ID_HEADER) - def _create_event_data(self, event_message: EventMessage) -> SSEEvent: - """Create event data dictionary from an EventMessage.""" - event_data = { + def _create_event_data(self, event_message: EventMessage) -> SSEEvent | None: + """Create event data dictionary from an EventMessage. + + Returns None when the message cannot be serialized and has no request id + to attach an Internal server error to, so the SSE loop can skip it. + """ + data = serialize_outbound_message(event_message.message) + if data is None: + return None + + event_data: SSEEvent = { "event": "message", - "data": event_message.message.model_dump_json(by_alias=True, exclude_unset=True), + "data": data, } # If an event ID was provided, include it @@ -795,6 +820,8 @@ async def standalone_sse_writer(): # Send the message via SSE event_data = self._create_event_data(event_message) + if event_data is None: + continue await sse_stream_writer.send(event_data) except anyio.ClosedResourceError: # Session teardown can close the stream while the writer is between dequeues. @@ -958,6 +985,8 @@ async def replay_sender(): # Define an async callback for sending events async def send_event(event_message: EventMessage) -> None: event_data = self._create_event_data(event_message) + if event_data is None: + return await sse_stream_writer.send(event_data) # Replay past events and get the stream ID @@ -986,7 +1015,8 @@ async def send_event(event_message: EventMessage) -> None: async with msg_reader: async for event_message in msg_reader: event_data = self._create_event_data(event_message) - + if event_data is None: + continue await sse_stream_writer.send(event_data) finally: self._sse_stream_writers.pop(stream_id, None) diff --git a/src/mcp/shared/wire.py b/src/mcp/shared/wire.py new file mode 100644 index 0000000000..30801ecca9 --- /dev/null +++ b/src/mcp/shared/wire.py @@ -0,0 +1,50 @@ +"""Outbound JSON-RPC wire serialization shared by transports. + +`model_dump_json` is the normal path. Lone surrogates (and any other value that +survives `model_dump(mode="json")` but fails the final UTF-8 encode) must not +tear down a writer task: one bad message costs only itself. +""" + +from __future__ import annotations + +import logging + +from mcp_types import INTERNAL_ERROR, ErrorData, JSONRPCError, JSONRPCMessage, JSONRPCResponse + +logger = logging.getLogger(__name__) + + +def serialize_jsonrpc_message(message: JSONRPCMessage) -> str: + """Serialize a JSON-RPC message the way transports always have.""" + return message.model_dump_json(by_alias=True, exclude_unset=True) + + +def serialize_outbound_message(message: JSONRPCMessage) -> str | None: + """Serialize for the wire; never raise for a single unserializable message. + + On failure: + - A response or error with an id becomes ``Internal server error`` (same + shape as other handler serialization failures higher in the stack). + - A request, notification, or id-less error returns ``None`` so the caller + can drop it or fail that request locally without closing the session. + """ + try: + return serialize_jsonrpc_message(message) + except Exception: + logger.exception("Failed to serialize outbound JSON-RPC message") + return _fallback_after_serialization_failure(message) + + +def _fallback_after_serialization_failure(message: JSONRPCMessage) -> str | None: + if isinstance(message, JSONRPCResponse | JSONRPCError) and message.id is not None: + replacement = JSONRPCError( + jsonrpc="2.0", + id=message.id, + error=ErrorData(code=INTERNAL_ERROR, message="Internal server error"), + ) + try: + return serialize_jsonrpc_message(replacement) + except Exception: # pragma: no cover + logger.exception("Failed to serialize serialization-error fallback") + return None + return None diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index b64eb01902..b40a5e5666 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -1456,3 +1456,40 @@ async def test_a_graceful_exit_with_a_surviving_child_leaks_no_pipe_fds( # prag # Subset, not equality: other machinery may close fds, but never open new # ones; a leaked pipe fd would show up as an extra entry. assert set(os.listdir("/proc/self/fd")) <= baseline + + +@pytest.mark.anyio +async def test_surrogate_in_outbound_request_does_not_tear_down_writer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A lone surrogate in one outbound request fails that request only; later writes continue (#3707).""" + import os + + from mcp_types import INTERNAL_ERROR, JSONRPCError + + surrogate = os.fsdecode(b"a\xffb") + bad = JSONRPCRequest( + jsonrpc="2.0", + id=1, + method="tools/call", + params={"name": "echo", "arguments": {"text": surrogate}}, + ) + good = JSONRPCRequest(jsonrpc="2.0", id=2, method="ping") + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS) as (read_stream, write_stream): + await write_stream.send(SessionMessage(bad)) + await anyio.wait_all_tasks_blocked() + failed = await read_stream.receive() + assert isinstance(failed, SessionMessage) + assert isinstance(failed.message, JSONRPCError) + assert failed.message.id == 1 + assert failed.message.error.code == INTERNAL_ERROR + assert failed.message.error.message == "Failed to serialize request" + assert process.written == [] + + await write_stream.send(SessionMessage(good)) + await anyio.wait_all_tasks_blocked() + assert process.written == [_line(good)] diff --git a/tests/server/test_stdio.py b/tests/server/test_stdio.py index eafd1fca59..43c0e3550e 100644 --- a/tests/server/test_stdio.py +++ b/tests/server/test_stdio.py @@ -692,3 +692,38 @@ def test_mcpserver_run_stdio_serves_a_modern_connection(monkeypatch: pytest.Monk # resultType is modern-only: proves the request was served at the discovered version. assert responses[1].result["tools"] == [] assert responses[1].result["resultType"] == "complete" + + +@pytest.mark.anyio +async def test_stdio_server_surrogate_in_response_does_not_kill_writer() -> None: + """A lone surrogate in one outbound response becomes Internal error; the writer continues (#3707).""" + import json + import os + + from mcp_types import INTERNAL_ERROR + + stdin = io.StringIO() + stdout = io.StringIO() + surrogate = os.fsdecode(b"report-\xff.txt") + + with anyio.fail_after(5): + async with stdio_server(stdin=anyio.AsyncFile(stdin), stdout=anyio.AsyncFile(stdout)) as ( + read_stream, + write_stream, + ): + await write_stream.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=1, result={"text": surrogate}))) + await write_stream.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=2, result={"ok": True}))) + await write_stream.aclose() + await read_stream.aclose() + + stdout.seek(0) + lines = [line.strip() for line in stdout.readlines() if line.strip()] + assert len(lines) == 2 + first = json.loads(lines[0]) + assert first == { + "jsonrpc": "2.0", + "id": 1, + "error": {"code": INTERNAL_ERROR, "message": "Internal server error"}, + } + second = jsonrpc_message_adapter.validate_json(lines[1]) + assert second == JSONRPCResponse(jsonrpc="2.0", id=2, result={"ok": True}) diff --git a/tests/shared/test_wire.py b/tests/shared/test_wire.py new file mode 100644 index 0000000000..c2ef08e7a2 --- /dev/null +++ b/tests/shared/test_wire.py @@ -0,0 +1,72 @@ +"""Outbound wire serialization must contain lone-surrogate failures (#3707).""" + +import json +import os + +import pytest +from mcp_types import ( + INTERNAL_ERROR, + ErrorData, + JSONRPCError, + JSONRPCNotification, + JSONRPCRequest, + JSONRPCResponse, +) + +from mcp.shared.wire import serialize_jsonrpc_message, serialize_outbound_message + + +def _surrogate() -> str: + return os.fsdecode(b"report-\xff.txt") + + +def test_serialize_jsonrpc_message_rejects_lone_surrogate() -> None: + message = JSONRPCResponse(jsonrpc="2.0", id=1, result={"text": _surrogate()}) + with pytest.raises(Exception): + serialize_jsonrpc_message(message) + + +def test_serialize_outbound_response_with_surrogate_becomes_internal_error() -> None: + message = JSONRPCResponse(jsonrpc="2.0", id=7, result={"text": _surrogate()}) + wire = serialize_outbound_message(message) + assert wire is not None + body = json.loads(wire) + assert body == { + "jsonrpc": "2.0", + "id": 7, + "error": {"code": INTERNAL_ERROR, "message": "Internal server error"}, + } + + +def test_serialize_outbound_request_with_surrogate_returns_none() -> None: + message = JSONRPCRequest( + jsonrpc="2.0", + id=3, + method="tools/call", + params={"name": "echo", "arguments": {"text": _surrogate()}}, + ) + assert serialize_outbound_message(message) is None + + +def test_serialize_outbound_notification_with_surrogate_returns_none() -> None: + message = JSONRPCNotification( + jsonrpc="2.0", + method="notifications/message", + params={"level": "info", "data": _surrogate()}, + ) + assert serialize_outbound_message(message) is None + + +def test_serialize_outbound_healthy_message_unchanged() -> None: + message = JSONRPCResponse(jsonrpc="2.0", id=1, result={"ok": True}) + assert serialize_outbound_message(message) == serialize_jsonrpc_message(message) + + +def test_serialize_outbound_error_with_surrogate_in_message_becomes_internal_error() -> None: + message = JSONRPCError(jsonrpc="2.0", id=9, error=ErrorData(code=-32000, message=_surrogate())) + wire = serialize_outbound_message(message) + assert wire is not None + body = json.loads(wire) + assert body["id"] == 9 + assert body["error"]["code"] == INTERNAL_ERROR + assert body["error"]["message"] == "Internal server error"