Add courtesy-cancel controls and bound shielded dispatcher writes

- New CallOptions key cancel_on_abandon (default true): abandoning a request
  (timeout or caller cancellation) sends notifications/cancelled unless the
  caller opted out or the request carries resumption hints
- Bound the two shielded cancellation-path writes with a 5s deadline so a
  wedged transport write cannot hang shutdown or a cancelled caller
- Capitalize the connection-closed fan-out message ("Connection closed")
- Pin the server-seat timeout contract in the interaction suite: a timed-out
  server-initiated request is followed by notifications/cancelled

Author: maxisbey

src/mcp/shared/dispatcher.py

@@ -55,6 +55,16 @@ class CallOptions(TypedDict, total=False):
     timeout: float
     """Seconds to wait for a result before raising and sending `notifications/cancelled`."""
 
+    cancel_on_abandon: bool
+    """Whether abandoning this request sends `notifications/cancelled` to the peer.
+
+    A request is abandoned when its `timeout` elapses or the caller's scope is
+    cancelled while awaiting the response. Defaults to `True`. Set `False` for
+    requests the protocol forbids cancelling, such as `initialize`. The
+    notification is also suppressed when resumption hints are present: the
+    caller intends to resume the request, so the peer's work must keep running.
+    """
+
     on_progress: ProgressFnT
     """Receive `notifications/progress` updates for this request."""
 
@@ -97,8 +107,8 @@ async def send_raw_request(
     ) -> dict[str, Any]:
         """Send a request and await its raw result dict.
 
-        `opts` carries per-call `timeout` / `on_progress` / resumption hints;
-        see `CallOptions`.
+        `opts` carries per-call `timeout` / `on_progress` / abandon-cancellation
+        / resumption hints; see `CallOptions`.
 
         Raises:
             MCPError: If the peer responded with an error, or the handler

src/mcp/shared/jsonrpc_dispatcher.py

@@ -63,6 +63,13 @@
 
 logger = logging.getLogger(__name__)
 
+_SHIELDED_WRITE_TIMEOUT: float = 5
+"""Bound for the shielded courtesy writes on the cancellation paths.
+
+Those writes run inside a shield because the surrounding scope is already
+cancelled; without a bound, a wedged transport write would turn the shield
+into an uncancellable hang (and block shutdown indefinitely)."""
+
 TransportT = TypeVar("TransportT", bound=TransportContext)
 
 PeerCancelMode = Literal["interrupt", "signal"]
@@ -323,6 +330,18 @@ async def send_raw_request(
         pending = _Pending(send=send, receive=receive, on_progress=on_progress)
         self._pending[request_id] = pending
 
+        # An abandoned request (timeout elapsed, or the caller's scope was
+        # cancelled while awaiting the response) sends a courtesy
+        # `notifications/cancelled` so the peer can stop work - unless the
+        # caller opted out (`initialize`, which the spec forbids cancelling),
+        # or the request carries resumption hints (the caller intends to
+        # resume it, so the peer's work must keep running).
+        cancel_on_abandon = (
+            opts.get("cancel_on_abandon", True)
+            and opts.get("resumption_token") is None
+            and opts.get("on_resumption_token") is None
+        )
+
         metadata = _outbound_metadata(_related_request_id, opts)
         target = out_params.get("name")
         span_name = f"MCP send {method}{f' {target}' if isinstance(target, str) else ''}"
@@ -348,14 +367,16 @@ async def send_raw_request(
             # Spec-recommended courtesy: tell the peer we've given up so it can
             # stop work and free resources. v1's BaseSession.send_request does
             # NOT do this; it's new behaviour.
-            await self._cancel_outbound(request_id, f"timed out after {opts.get('timeout')}s", _related_request_id)
+            if cancel_on_abandon:
+                await self._cancel_outbound(request_id, f"timed out after {opts.get('timeout')}s", _related_request_id)
             raise MCPError(code=REQUEST_TIMEOUT, message=f"Request {method!r} timed out") from None
         except anyio.get_cancelled_exc_class():
             # Our caller's scope was cancelled. We're already inside a cancelled
-            # scope, so any bare `await` here re-raises immediately - shield to
-            # let the courtesy cancel notification go out before we propagate.
-            with anyio.CancelScope(shield=True):
-                await self._cancel_outbound(request_id, "caller cancelled", _related_request_id)
+            # scope, so any bare `await` here re-raises immediately - shield
+            # (bounded) to let the courtesy cancel go out before we propagate.
+            if cancel_on_abandon:
+                with anyio.move_on_after(_SHIELDED_WRITE_TIMEOUT, shield=True):
+                    await self._cancel_outbound(request_id, "caller cancelled", _related_request_id)
             raise
         finally:
             # Always remove the waiter, even on cancel/timeout, so a late
@@ -635,7 +656,7 @@ def _fan_out_closed(self) -> None:
         Synchronous (uses `send_nowait`) because it's called from `finally`
         which may be inside a cancelled scope. Idempotent.
         """
-        closed = ErrorData(code=CONNECTION_CLOSED, message="connection closed")
+        closed = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
         for pending in self._pending.values():
             try:
                 pending.send.send_nowait(closed)
@@ -681,8 +702,9 @@ async def _handle_request(
                 await self._write_error(req.id, ErrorData(code=0, message="Request cancelled"))
         except anyio.get_cancelled_exc_class():
             # Outer-cancel: run()'s task group is shutting down. Any bare
-            # `await` here re-raises immediately, so shield the courtesy write.
-            with anyio.CancelScope(shield=True):
+            # `await` here re-raises immediately, so shield (bounded) the
+            # courtesy write.
+            with anyio.move_on_after(_SHIELDED_WRITE_TIMEOUT, shield=True):
                 await self._write_error(req.id, ErrorData(code=REQUEST_CANCELLED, message="Request cancelled"))
             raise
         except MCPError as e:

tests/interaction/_requirements.py

@@ -468,7 +468,9 @@ def __post_init__(self) -> None:
         ),
         divergence=Divergence(
             note=(
-                "The client only raises locally and sends nothing on timeout, so the server keeps running the handler."
+                "Client seat only: the client raises locally and sends nothing on timeout, so the server keeps "
+                "running the handler. The server seat conforms: a timed-out server-initiated request is followed "
+                "by notifications/cancelled on the wire."
             ),
         ),
     ),

tests/interaction/lowlevel/test_timeouts.py

@@ -13,9 +13,13 @@
 from trio.testing import MockClock
 
 from mcp import MCPError, types
+from mcp.client import ClientRequestContext
+from mcp.client._memory import InMemoryTransport
 from mcp.client.client import Client
 from mcp.server import Server, ServerRequestContext
-from mcp.types import REQUEST_TIMEOUT, CallToolResult, ErrorData, TextContent
+from mcp.shared.message import SessionMessage
+from mcp.types import REQUEST_TIMEOUT, CallToolResult, ErrorData, JSONRPCNotification, TextContent
+from tests.interaction._helpers import RecordingTransport
 from tests.interaction._requirements import requirement
 
 pytestmark = pytest.mark.anyio
@@ -56,6 +60,68 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
     )
 
 
+@requirement("protocol:timeout:basic")
+@requirement("protocol:timeout:sends-cancellation")
+async def test_server_request_timeout_sends_cancellation_to_the_client() -> None:
+    """A server-initiated request that times out fails server-side and cancels the client's work.
+
+    The server seat conforms to the spec's timeout guidance: the handler's timed-out sampling
+    request is followed by notifications/cancelled on the wire. The client's sampling callback
+    blocks until the server has already given up, then answers; the late response is discarded
+    and the tool call still completes.
+    """
+    release = anyio.Event()
+    callback_started = anyio.Event()
+    errors: list[ErrorData] = []
+
+    async def list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=[types.Tool(name="impatient", input_schema={"type": "object"})])
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        assert params.name == "impatient"
+        request = types.CreateMessageRequest(
+            params=types.CreateMessageRequestParams(
+                messages=[types.SamplingMessage(role="user", content=TextContent(text="Say hello."))],
+                max_tokens=8,
+            )
+        )
+        with pytest.raises(MCPError) as exc_info:
+            await ctx.session.send_request(request, types.CreateMessageResult, request_read_timeout_seconds=0.000001)
+        errors.append(exc_info.value.error)
+        release.set()
+        return CallToolResult(content=[TextContent(text="gave up")])
+
+    server = Server("impatient", on_list_tools=list_tools, on_call_tool=call_tool)
+    recording = RecordingTransport(InMemoryTransport(server))
+
+    async def sampling_callback(
+        context: ClientRequestContext, params: types.CreateMessageRequestParams
+    ) -> types.CreateMessageResult:
+        callback_started.set()
+        await release.wait()
+        return types.CreateMessageResult(role="assistant", content=TextContent(text="too late"), model="test-model")
+
+    async with Client(recording, sampling_callback=sampling_callback) as client:
+        result = await client.call_tool("impatient", {})
+
+    assert result == snapshot(CallToolResult(content=[TextContent(text="gave up")]))
+    assert callback_started.is_set()
+    assert errors == snapshot([ErrorData(code=REQUEST_TIMEOUT, message="Request 'sampling/createMessage' timed out")])
+    cancellations = [
+        item.message
+        for item in recording.received
+        if isinstance(item, SessionMessage)
+        and isinstance(item.message, JSONRPCNotification)
+        and item.message.method == "notifications/cancelled"
+    ]
+    # The cancel names the sampling request (the server's first outbound request) and the reason.
+    assert [notification.params for notification in cancellations] == snapshot(
+        [{"requestId": 1, "reason": "timed out after 1e-06s"}]
+    )
+
+
 @requirement("protocol:timeout:session-survives")
 async def test_session_serves_requests_after_timeout() -> None:
     """A timed-out request does not poison the session: the next request succeeds."""