fix: catch ClosedResourceError in _handle_message error recovery path

When a client disconnects during a request, _handle_message receives the
exception and tries to send_log_message() back to notify the client. Since
the write stream is already closed, this raises ClosedResourceError (or
BrokenResourceError), which is unhandled in the TaskGroup and crashes the
stateless session with an ExceptionGroup.

Wrap the send_log_message() call in _handle_message's Exception branch
with a try/except for ClosedResourceError and BrokenResourceError, since
failing to notify a disconnected client is expected and harmless.

Fixes #2064

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: g97iulio1609

src/mcp/server/lowlevel/server.py

@@ -417,11 +417,16 @@ async def _handle_message(
                         )
                 case Exception():
                     logger.error(f"Received exception from stream: {message}")
-                    await session.send_log_message(
-                        level="error",
-                        data="Internal Server Error",
-                        logger="mcp.server.exception_handler",
-                    )
+                    try:
+                        await session.send_log_message(
+                            level="error",
+                            data="Internal Server Error",
+                            logger="mcp.server.exception_handler",
+                        )
+                    except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+                        logger.debug(
+                            "Could not send error log to client (stream already closed)"
+                        )
                     if raise_exceptions:
                         raise message
                 case _:

tests/server/test_lowlevel_exception_handling.py

@@ -72,3 +72,61 @@ async def test_normal_message_handling_not_affected():
 
     # Verify _handle_request was called
     server._handle_request.assert_called_once()
+
+
+@pytest.mark.anyio
+async def test_exception_handling_survives_closed_write_stream():
+    """When the client disconnects, send_log_message may raise ClosedResourceError.
+
+    The server should catch it gracefully instead of crashing the session.
+    Regression test for GH-2064.
+    """
+    import anyio
+
+    server = Server("test-server")
+    session = Mock(spec=ServerSession)
+    session.send_log_message = AsyncMock(side_effect=anyio.ClosedResourceError)
+
+    test_exception = RuntimeError("client disconnected")
+
+    # Should NOT raise — the ClosedResourceError from send_log_message is caught
+    await server._handle_message(test_exception, session, {}, raise_exceptions=False)
+
+    session.send_log_message.assert_called_once()
+
+
+@pytest.mark.anyio
+async def test_exception_handling_survives_broken_write_stream():
+    """When the write stream is broken, send_log_message may raise BrokenResourceError.
+
+    The server should catch it gracefully.
+    Regression test for GH-2064.
+    """
+    import anyio
+
+    server = Server("test-server")
+    session = Mock(spec=ServerSession)
+    session.send_log_message = AsyncMock(side_effect=anyio.BrokenResourceError)
+
+    test_exception = RuntimeError("broken pipe")
+
+    await server._handle_message(test_exception, session, {}, raise_exceptions=False)
+
+    session.send_log_message.assert_called_once()
+
+
+@pytest.mark.anyio
+async def test_exception_handling_raises_when_configured_despite_closed_stream():
+    """With raise_exceptions=True, the original error is still re-raised
+    even when send_log_message fails with ClosedResourceError.
+    """
+    import anyio
+
+    server = Server("test-server")
+    session = Mock(spec=ServerSession)
+    session.send_log_message = AsyncMock(side_effect=anyio.ClosedResourceError)
+
+    test_exception = RuntimeError("original error")
+
+    with pytest.raises(RuntimeError, match="original error"):
+        await server._handle_message(test_exception, session, {}, raise_exceptions=True)