fix: ignore closed log streams during shutdown

Author: lavish0000

src/mcp/server/lowlevel/server.py

@@ -414,11 +414,14 @@ 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.BrokenResourceError, anyio.ClosedResourceError):
+                        logger.debug("Skipping exception log message because the session write stream is closed")
                     if raise_exceptions:
                         raise message
                 case _:

tests/server/test_lowlevel_exception_handling.py

@@ -1,5 +1,6 @@
 from unittest.mock import AsyncMock, Mock
 
+import anyio
 import pytest
 
 from mcp import types
@@ -52,6 +53,19 @@ async def test_exception_handling_with_raise_exceptions_false(exception_class: t
     assert call_args.kwargs["logger"] == "mcp.server.exception_handler"
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("stream_error", [anyio.ClosedResourceError(), anyio.BrokenResourceError()])
+async def test_exception_handling_ignores_closed_log_stream(stream_error: Exception):
+    """Logging an exception should not crash shutdown if the write stream is already gone."""
+    server = Server("test-server")
+    session = Mock(spec=ServerSession)
+    session.send_log_message = AsyncMock(side_effect=stream_error)
+
+    await server._handle_message(RuntimeError("Test error"), session, {}, raise_exceptions=False)
+
+    session.send_log_message.assert_called_once()
+
+
 @pytest.mark.anyio
 async def test_normal_message_handling_not_affected():
     """Test that normal messages still work correctly"""