fix: contain lone-surrogate serialization failures in transport writers

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):

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,
                         }
                     )
 

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

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)

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

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)]

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})

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"