fix: send notifications/cancelled on request timeout and caller cancellation

ClientSession.send_request() never emitted a notifications/cancelled
message when a request was interrupted, either by the SDK's own timeout
or by the caller's cancellation. This violated the MCP spec and caused
server-side tool coroutines to leak — they remained suspended holding
resources (DB connections, locks, file handles) until the session ended.

Handle both cancellation paths:
- TimeoutError from anyio.fail_after: send notification before raising
- External CancelledError from caller: catch, send notification, re-raise

The notification is sent inside a shielded cancel scope to guarantee
delivery even when called from a cancelled task. The server's existing
CancelledNotification handler already cancels in-flight requests on
receipt, so no server-side changes are needed.

Fixes #2507

Author: dragogargo

src/mcp/shared/session.py

@@ -23,6 +23,7 @@
     INVALID_PARAMS,
     REQUEST_TIMEOUT,
     CancelledNotification,
+    CancelledNotificationParams,
     ClientNotification,
     ClientRequest,
     ClientResult,
@@ -292,9 +293,13 @@ async def send_request(
                     with anyio.fail_after(timeout):
                         response_or_error = await response_stream_reader.receive()
                 except TimeoutError:
+                    await self._send_cancelled_notification(request_id, "request timed out")
                     class_name = request.__class__.__name__
                     message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds."
                     raise MCPError(code=REQUEST_TIMEOUT, message=message)
+                except anyio.get_cancelled_exc_class():
+                    await self._send_cancelled_notification(request_id, "request cancelled by caller")
+                    raise
 
                 if isinstance(response_or_error, JSONRPCError):
                     raise MCPError.from_jsonrpc_error(response_or_error)
@@ -540,6 +545,26 @@ async def send_progress_notification(
     ) -> None:
         """Sends a progress notification for a request that is currently being processed."""
 
+    async def _send_cancelled_notification(self, request_id: RequestId, reason: str) -> None:
+        """Send a cancellation notification to the remote side (best-effort).
+
+        Uses a shielded cancel scope so the notification is delivered even
+        when called from inside a cancelled task.
+        """
+        try:
+            with anyio.CancelScope(shield=True):
+                notification = CancelledNotification(
+                    method="notifications/cancelled",
+                    params=CancelledNotificationParams(request_id=request_id, reason=reason),
+                )
+                jsonrpc_notification = JSONRPCNotification(
+                    jsonrpc="2.0",
+                    **notification.model_dump(by_alias=True, mode="json", exclude_none=True),
+                )
+                await self._write_stream.send(SessionMessage(message=jsonrpc_notification))
+        except Exception:
+            logging.warning("Failed to send cancellation notification for request %s", request_id)
+
     async def _handle_incoming(
         self, req: RequestResponder[ReceiveRequestT, SendResultT] | ReceiveNotificationT | Exception
     ) -> None:

tests/shared/test_session.py

@@ -416,3 +416,97 @@ async def make_request(client_session: ClientSession):
     # Pending request completed successfully
     assert len(result_holder) == 1
     assert isinstance(result_holder[0], EmptyResult)
+
+
+@pytest.mark.anyio
+async def test_send_request_sends_cancelled_notification_on_timeout():
+    """Client must send notifications/cancelled when a request times out.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/2507.
+    """
+    cancelled_notifications: list[SessionMessage] = []
+
+    async with create_client_server_memory_streams() as (client_streams, server_streams):
+        client_read, client_write = client_streams
+        server_read, server_write = server_streams
+
+        async def mock_server():
+            async for message in server_read:
+                assert isinstance(message, SessionMessage)
+                if isinstance(message.message, JSONRPCRequest):
+                    pass
+                elif isinstance(message.message, types.JSONRPCNotification):
+                    cancelled_notifications.append(message)
+
+        async with (
+            anyio.create_task_group() as tg,
+            ClientSession(
+                read_stream=client_read,
+                write_stream=client_write,
+                read_timeout_seconds=0.1,
+            ) as client_session,
+        ):
+            tg.start_soon(mock_server)
+
+            with pytest.raises(MCPError, match="Timed out"):
+                await client_session.send_ping()
+
+            await anyio.sleep(0.05)
+            tg.cancel_scope.cancel()
+
+    assert len(cancelled_notifications) == 1
+    notif = cancelled_notifications[0].message
+    assert isinstance(notif, types.JSONRPCNotification)
+    assert notif.method == "notifications/cancelled"
+    assert notif.params is not None
+    assert notif.params.get("reason") == "request timed out"
+
+
+@pytest.mark.anyio
+async def test_send_request_sends_cancelled_notification_on_caller_cancel():
+    """Client must send notifications/cancelled when caller cancels the request.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/2507.
+    """
+    cancelled_notifications: list[SessionMessage] = []
+    ev_request_sent = anyio.Event()
+
+    async with create_client_server_memory_streams() as (client_streams, server_streams):
+        client_read, client_write = client_streams
+        server_read, server_write = server_streams
+
+        async def mock_server():
+            async for message in server_read:
+                assert isinstance(message, SessionMessage)
+                if isinstance(message.message, JSONRPCRequest):
+                    ev_request_sent.set()
+                elif isinstance(message.message, types.JSONRPCNotification):
+                    cancelled_notifications.append(message)
+
+        async def make_request(client_session: ClientSession):
+            await client_session.send_ping()
+
+        async with (
+            anyio.create_task_group() as tg,
+            ClientSession(
+                read_stream=client_read,
+                write_stream=client_write,
+            ) as client_session,
+        ):
+            tg.start_soon(mock_server)
+
+            async with anyio.create_task_group() as request_tg:
+                request_tg.start_soon(make_request, client_session)
+                with anyio.fail_after(1):
+                    await ev_request_sent.wait()
+                request_tg.cancel_scope.cancel()
+
+            await anyio.sleep(0.05)
+            tg.cancel_scope.cancel()
+
+    assert len(cancelled_notifications) == 1
+    notif = cancelled_notifications[0].message
+    assert isinstance(notif, types.JSONRPCNotification)
+    assert notif.method == "notifications/cancelled"
+    assert notif.params is not None
+    assert notif.params.get("reason") == "request cancelled by caller"