fix: don't send log notification on transport error

When a transport yields an Exception into the server's read stream
(e.g., POST handler catches an error, or JSON parsing fails), the
server was attempting to send a notifications/message back to the
client. This raced with the write stream being closed during session
teardown, raising ClosedResourceError inside the handler's TaskGroup
and crashing the session.

The transport that just reported an error is typically broken or
about to close. In the streamable HTTP case, the client has already
received an HTTP 500 before the Exception is even yielded. The
TypeScript, Go, and C# SDKs all handle transport errors by logging
locally only, never writing back through the transport.

This removes the send_log_message call added in #786, keeping the
local logger.error and the raise_exceptions re-raise. A new
integration test with real memory streams (no mocks, no sleeps)
deterministically reproduces the exact #1967 stack trace on the old
code and passes on the new code.

Fixes #1967
Fixes #2064

Reported-by: jonfinerty
Reported-by: einarfd
Github-Issue: #1967
Github-Issue: #2064

Author: maxisbey

src/mcp/server/lowlevel/server.py

@@ -414,11 +414,6 @@ 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",
-                    )
                     if raise_exceptions:
                         raise message
                 case _:

tests/server/test_lowlevel_exception_handling.py

@@ -1,55 +1,42 @@
 from unittest.mock import AsyncMock, Mock
 
+import anyio
 import pytest
 
 from mcp import types
 from mcp.server.lowlevel.server import Server
 from mcp.server.session import ServerSession
+from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
 
 
 @pytest.mark.anyio
 async def test_exception_handling_with_raise_exceptions_true():
-    """Test that exceptions are re-raised when raise_exceptions=True"""
+    """Transport exceptions are re-raised when raise_exceptions=True."""
     server = Server("test-server")
     session = Mock(spec=ServerSession)
-    session.send_log_message = AsyncMock()
 
     test_exception = RuntimeError("Test error")
 
     with pytest.raises(RuntimeError, match="Test error"):
         await server._handle_message(test_exception, session, {}, raise_exceptions=True)
 
-    session.send_log_message.assert_called_once()
-
 
 @pytest.mark.anyio
-@pytest.mark.parametrize(
-    "exception_class,message",
-    [
-        (ValueError, "Test validation error"),
-        (RuntimeError, "Test runtime error"),
-        (KeyError, "Test key error"),
-        (Exception, "Basic error"),
-    ],
-)
-async def test_exception_handling_with_raise_exceptions_false(exception_class: type[Exception], message: str):
-    """Test that exceptions are logged when raise_exceptions=False"""
+async def test_exception_handling_with_raise_exceptions_false():
+    """Transport exceptions are logged locally but not sent to the client.
+
+    The transport that reported the error is likely broken; writing back
+    through it races with stream closure (#1967, #2064). The TypeScript,
+    Go, and C# SDKs all log locally only.
+    """
     server = Server("test-server")
     session = Mock(spec=ServerSession)
     session.send_log_message = AsyncMock()
 
-    test_exception = exception_class(message)
-
-    await server._handle_message(test_exception, session, {}, raise_exceptions=False)
-
-    # Should send log message
-    session.send_log_message.assert_called_once()
-    call_args = session.send_log_message.call_args
+    await server._handle_message(RuntimeError("Test error"), session, {}, raise_exceptions=False)
 
-    assert call_args.kwargs["level"] == "error"
-    assert call_args.kwargs["data"] == "Internal Server Error"
-    assert call_args.kwargs["logger"] == "mcp.server.exception_handler"
+    session.send_log_message.assert_not_called()
 
 
 @pytest.mark.anyio
@@ -72,3 +59,43 @@ 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_server_run_exits_cleanly_when_transport_yields_exception_then_closes():
+    """Regression test for #1967 / #2064.
+
+    Exercises the real Server.run() path with real memory streams, reproducing
+    what happens in stateless streamable HTTP when a POST handler throws:
+
+    1. Transport yields an Exception into the read stream
+       (streamable_http.py does this in its broad POST-handler except).
+    2. Transport closes the read stream (terminate() in stateless mode).
+    3. _receive_loop exits its `async with read_stream, write_stream:` block,
+       closing the write stream.
+    4. Meanwhile _handle_message(exc) was spawned via tg.start_soon and runs
+       after the write stream is closed.
+
+    Before the fix, _handle_message tried to send_log_message through the
+    closed write stream, raising ClosedResourceError inside the TaskGroup
+    and crashing server.run(). After the fix, it only logs locally.
+    """
+    server = Server("test-server")
+
+    read_send, read_recv = anyio.create_memory_object_stream[SessionMessage | Exception](10)
+    write_send, write_recv = anyio.create_memory_object_stream[SessionMessage](10)
+
+    async def transport_side() -> None:
+        # What the streamable HTTP transport does: push the exception, then close.
+        await read_send.send(RuntimeError("simulated transport error"))
+        await read_send.aclose()
+        # Drain the write side so it doesn't back-pressure; it must see no messages.
+        async with write_recv:
+            async for _ in write_recv:  # pragma: no cover
+                pytest.fail("server must not write back when the transport reported an error")
+
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(transport_side)
+            # stateless=True so server.run doesn't wait for initialize handshake
+            await server.run(read_recv, write_send, server.create_initialization_options(), stateless=True)