test(v1.x): fix closed-stream regression CI

Author: Ubuntu

tests/server/test_lowlevel_exception_handling.py

@@ -1,5 +1,6 @@
 from unittest.mock import AsyncMock, Mock
 
+import anyio
 import pytest
 
 import mcp.types as types
@@ -72,3 +73,24 @@ 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_handle_request_drops_response_when_transport_is_closed():
+    """Closed write streams during respond should be treated as expected drops."""
+    server = Server("test-server")
+    session = Mock(spec=ServerSession)
+
+    responder = Mock(spec=RequestResponder)
+    responder.request_id = 1
+    responder.respond = AsyncMock(side_effect=anyio.ClosedResourceError())
+
+    await server._handle_request(
+        responder,
+        types.PingRequest(method="ping"),
+        session,
+        {},
+        raise_exceptions=False,
+    )
+
+    responder.respond.assert_called_once()

tests/server/test_session.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, cast
 
 import anyio
 import pytest
@@ -221,64 +221,65 @@ async def mock_client():
 
 
 class _ClosedWriteStream:
-    async def send(self, item):
+    async def send(self, item: SessionMessage) -> None:
         raise anyio.ClosedResourceError
 
 
 class _OpenWriteStream:
     def __init__(self):
         self.items: list[SessionMessage] = []
 
-    async def send(self, item):
+    async def send(self, item: SessionMessage) -> None:
         self.items.append(item)
 
 
 class _FakeResult:
     def __init__(self, payload: dict[str, Any]):
         self._payload = payload
 
-    def model_dump(self, **kwargs):
+    def model_dump(self, **kwargs: Any) -> dict[str, Any]:
         return dict(self._payload)
 
 
 class _FakeNotification:
     def __init__(self, payload: dict[str, Any]):
         self._payload = payload
 
-    def model_dump(self, **kwargs):
+    def model_dump(self, **kwargs: Any) -> dict[str, Any]:
         return dict(self._payload)
 
 
 @pytest.mark.anyio
 async def test_base_session_send_response_ignores_closed_write_stream():
-    session = object.__new__(BaseSession)
+    session = cast(Any, object.__new__(BaseSession))
     session._write_stream = _ClosedWriteStream()
 
-    await BaseSession._send_response(session, 1, _FakeResult({"ok": True}))
+    await cast(Any, BaseSession)._send_response(session, 1, _FakeResult({"ok": True}))
 
 
 @pytest.mark.anyio
 async def test_base_session_send_notification_ignores_closed_write_stream():
-    session = object.__new__(BaseSession)
+    session = cast(Any, object.__new__(BaseSession))
     session._write_stream = _ClosedWriteStream()
 
-    await BaseSession.send_notification(
+    await cast(Any, BaseSession).send_notification(
         session,
         _FakeNotification({"method": "notifications/progress", "params": {"progress": 1}}),
     )
 
 
 @pytest.mark.anyio
 async def test_base_session_send_notification_still_writes_when_open():
-    session = object.__new__(BaseSession)
-    session._write_stream = _OpenWriteStream()
+    open_stream = _OpenWriteStream()
+    session = cast(Any, object.__new__(BaseSession))
+    session._write_stream = open_stream
 
-    await BaseSession.send_notification(
+    await cast(Any, BaseSession).send_notification(
         session,
         _FakeNotification({"method": "notifications/progress", "params": {"progress": 1}}),
     )
 
-    assert len(session._write_stream.items) == 1
+    assert len(open_stream.items) == 1
 
 
 @pytest.mark.anyio