diff --git a/docs/migration.md b/docs/migration.md index 501fd8aba1..7b5950e7cf 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -1584,6 +1584,14 @@ The method and the raw inbound params are `ctx.method` and `ctx.params` (`params Previously it also re-raised exceptions yielded by the transport onto the read stream (e.g. JSON parse errors). Those are now debug-logged and dropped regardless of `raise_exceptions`. If you relied on `run()` exiting on a transport-level parse error, that no longer happens. +### Cancelled requests are no longer answered + +In v1, when the peer sent `notifications/cancelled` for an in-flight request, the receiving side interrupted the handler and answered the request anyway with a JSON-RPC error, `{"code": 0, "message": "Request cancelled"}` - and `0` is not a defined JSON-RPC error code. The 2026-07-28 transport specifications (stdio, streamable HTTP) say a server **MUST NOT** send any further messages for a cancelled request; the older cancellation pattern already said it **SHOULD NOT**. The sender is expected to stop waiting once it cancels, so that error response has been removed: a cancelled request now produces no response at all - no result, and no error - even if the handler runs to completion or fails afterwards. This applies to both seats (the server for cancelled client requests, and the client for cancelled server-initiated requests such as sampling or elicitation). + +The one deliberate exception is the 2025-era streamable HTTP transport (`StreamableHTTPServerTransport`), whose wire can end a request's stream only with a response for that id (and stores that response so a resuming client's replay terminates too). Under the 2025 rule, a SHOULD NOT, that transport now terminates a cancelled request with a valid `-32800` error (`mcp.server.streamable_http.REQUEST_CANCELLED`, mirroring LSP's `RequestCancelled`) in place of the old `0`. Nothing else answers. + +Nothing changes for callers of the built-in client: abandoning a call (cancelling the awaiting task, or a per-request timeout) never waited for that response. If you send `notifications/cancelled` by hand while still awaiting the call, the call now receives nothing on most transports (over 2025-era streamable HTTP it fails with `REQUEST_CANCELLED`); stop awaiting it yourself, or use a per-request timeout. + ### `Server.run()` no longer takes a `stateless` flag The `stateless: bool` parameter on the lowlevel `Server.run()` has been removed. Stateless serving is now a property of how the connection is constructed (the streamable-HTTP manager builds a born-ready `Connection` per request), not a flag the loop driver inspects. @@ -1933,9 +1941,9 @@ except MCPError as e: Behavior changes: -- **Callbacks and notifications now run concurrently.** In v1 the receive loop processed one inbound message at a time, so callbacks ran inline and in order. Now each delivery starts in arrival order but runs as its own task. Server-initiated request callbacks (`sampling`, `elicitation`, `roots`) no longer block other traffic, may themselves send requests without deadlocking, and are interrupted if the server sends `notifications/cancelled` (the request is then answered with an error). Notification callbacks (`logging_callback`, `progress_callback`, `message_handler`) may interleave, and a `progress_callback` may run after the request it reports on has returned; there is no built-in bound on concurrent deliveries. Transport-level errors reach `message_handler` the same way, and a `message_handler` that raises is logged rather than fatal to the session. Callbacks that need strict sequencing must coordinate themselves. +- **Callbacks and notifications now run concurrently.** In v1 the receive loop processed one inbound message at a time, so callbacks ran inline and in order. Now each delivery starts in arrival order but runs as its own task. Server-initiated request callbacks (`sampling`, `elicitation`, `roots`) no longer block other traffic, may themselves send requests without deadlocking, and are interrupted if the server sends `notifications/cancelled` (no response is sent for the cancelled request). Notification callbacks (`logging_callback`, `progress_callback`, `message_handler`) may interleave, and a `progress_callback` may run after the request it reports on has returned; there is no built-in bound on concurrent deliveries. Transport-level errors reach `message_handler` the same way, and a `message_handler` that raises is logged rather than fatal to the session. Callbacks that need strict sequencing must coordinate themselves. - **Notification routing is unchanged.** Each server notification is still delivered to its typed callback first — `logging_callback` for log messages, the per-request `progress_callback` whose `progressToken` matches a request you issued (`progress_callback=` still stamps `params._meta.progressToken` with the outbound request id) — and then teed to `message_handler`. `notifications/cancelled` is applied by the dispatcher and never surfaced, also as in v1. -- **Cancellation now reaches the server.** Cancelling the task or cancel scope awaiting a request (e.g. `anyio.move_on_after()` around `session.call_tool(...)`), or a request hitting its read timeout, now sends `notifications/cancelled` for that request, so the server interrupts the handler instead of leaving it running; v1 sent nothing ([#2507](https://github.com/modelcontextprotocol/python-sdk/issues/2507)). A test that pinned the v1 gap with a strict `xfail` now passes — drop the marker. The cancelled peer still answers with `ErrorData(code=0, message="Request cancelled")` as in v1 (discarded, since the caller's waiter is already gone). There is no public request-id or cancel handle (v1's private `session._request_id` went with `BaseSession`): cancel the awaiting task or scope and the dispatcher sends the cancel for you. +- **Cancellation now reaches the server.** Cancelling the task or cancel scope awaiting a request (e.g. `anyio.move_on_after()` around `session.call_tool(...)`), or a request hitting its read timeout, now sends `notifications/cancelled` for that request, so the server interrupts the handler instead of leaving it running; v1 sent nothing ([#2507](https://github.com/modelcontextprotocol/python-sdk/issues/2507)). A test that pinned the v1 gap with a strict `xfail` now passes — drop the marker. The cancelled peer no longer answers at all (v1 sent `ErrorData(code=0, message="Request cancelled")`); the one exception, the 2025-era streamable HTTP transport's `-32800` terminator, is discarded like the v1 error since the caller's waiter is already gone — see [Cancelled requests are no longer answered](#cancelled-requests-are-no-longer-answered). There is no public request-id or cancel handle (v1's private `session._request_id` went with `BaseSession`): cancel the awaiting task or scope and the dispatcher sends the cancel for you. - **A raising request callback** is answered with `code=0` and the exception text; v1 flattened every callback exception to `INVALID_PARAMS`. For a specific error response, return `ErrorData` (unchanged) or raise `MCPError`. One carve-out: pydantic's `ValidationError` is still answered with `INVALID_PARAMS`, as in v1. - **`send_request` before entering the context manager** raises `RuntimeError` immediately; v1 wrote to the transport and hung until the timeout. After the connection has closed it raises `MCPError` (`CONNECTION_CLOSED`) instead. `send_notification` before entry still works. - **`send_notification` after the connection has closed is dropped with a debug log instead of raising.** In v1 the send raised `anyio.BrokenResourceError` (peer gone) or `anyio.ClosedResourceError` (session torn down), and this applied to the typed helpers (`send_roots_list_changed`, `send_progress_notification`) too. Code that used the exception as its disconnect signal should probe with a request instead (`send_request` still raises `MCPError` after close, see above) or scope the sending task to the session's lifetime. diff --git a/examples/stories/streaming/README.md b/examples/stories/streaming/README.md index e6bedb915a..c485399b71 100644 --- a/examples/stories/streaming/README.md +++ b/examples/stories/streaming/README.md @@ -56,10 +56,12 @@ uv run python -m stories.streaming.client --http --server server_lowlevel OpenTelemetry instead of `notifications/message`. It is shown here because servers still need to support 2025-era clients during that window. Progress and cancellation are **not** deprecated. TODO(maxisbey): revisit before beta. -- When a request is cancelled the server currently replies with - `ErrorData(code=0, message="Request cancelled")`; the spec says it should not - reply at all. The client never observes it (its awaiting task is already - cancelled), so this story does not assert on the reply. +- A cancelled request is not answered: no response follows + `notifications/cancelled`. (The 2025-era streamable HTTP transport is the one + exception - its wire ends a request only with a response, so it terminates + with a `-32800` `REQUEST_CANCELLED` error.) The client never observes any of + this - its awaiting task is already cancelled - so this story does not assert + on it. ## Spec @@ -69,5 +71,6 @@ uv run python -m stories.streaming.client --http --server server_lowlevel ## See also -`parallel_calls/` (concurrent in-flight calls), `error_handling/` (the -cancellation error path), `tools/` (the basics this builds on). +`parallel_calls/` (concurrent in-flight calls), `error_handling/` (error +surfaces: `is_error` results vs protocol errors), `tools/` (the basics this +builds on). diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index d316345c7e..324fc3e04b 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -65,6 +65,13 @@ # whole session on a lazily-started `sse_writer`. See #1764. REQUEST_STREAM_BUFFER_SIZE: Final = 16 +# Error code answering a request that settled without a response (e.g. it was +# cancelled) on this 2025-era wire, which ends a request's stream only with a +# response. Mirrors LSP's RequestCancelled; not sent by the 2026 transports, where +# the spec forbids answering a cancelled request. See +# `StreamableHTTPServerTransport._terminate_unanswered_request`. +REQUEST_CANCELLED: Final = -32800 + # Session ID validation pattern (visible ASCII characters ranging from 0x21 to 0x7E) # Pattern ensures entire string contains only valid characters by using ^ and $ anchors SESSION_ID_PATTERN = re.compile(r"^[\x21-\x7E]+$") @@ -252,7 +259,7 @@ def close_standalone_sse_stream(self) -> None: def _create_session_message( self, - message: JSONRPCMessage, + message: JSONRPCRequest, request: Request, request_id: RequestId, protocol_version: str, @@ -262,7 +269,10 @@ def _create_session_message( The close_sse_stream callbacks are only provided when the client supports resumability (protocol version >= 2025-11-25). Old clients can't resume if the stream is closed early because they didn't receive a priming event. + Every request carries `on_request_unanswered`, so a request that settles + without a response is still terminated on this era's wire. """ + end_stream = partial(self._terminate_unanswered_request, message.id) # Only provide close callbacks when client supports resumability if self._event_store and is_version_at_least(protocol_version, "2025-11-25"): @@ -276,9 +286,10 @@ async def close_standalone_stream_callback() -> None: request_context=request, close_sse_stream=close_stream_callback, close_standalone_sse_stream=close_standalone_stream_callback, + on_request_unanswered=end_stream, ) else: - metadata = ServerMessageMetadata(request_context=request) + metadata = ServerMessageMetadata(request_context=request, on_request_unanswered=end_stream) return SessionMessage(message, metadata=metadata) @@ -390,6 +401,20 @@ def _create_event_data(self, event_message: EventMessage) -> SSEEvent: return event_data + async def _terminate_unanswered_request(self, request_id: RequestId) -> None: + """Terminate a request that settled without a response (e.g. cancelled). + + The 2025-era wire ends a request's stream only with a response for its + id - and stores that response so a resuming client's replay terminates + too - so this era answers a cancelled request with `REQUEST_CANCELLED` + where the dispatcher itself stays silent (the 2026 transports MUST NOT + answer). It is written through the same ordered channel as the request's + other messages, so it cannot overtake anything already queued for it. + """ + assert self._write_stream is not None # a dispatched request implies connect() ran + error = ErrorData(code=REQUEST_CANCELLED, message="Request cancelled") + await self._write_stream.send(SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error))) + async def _clean_up_memory_streams(self, request_id: RequestId) -> None: """Clean up memory streams for a given request ID.""" if request_id in self._request_streams: # pragma: no branch @@ -555,7 +580,10 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re ) request_stream_reader = self._request_streams[request_id][1] # Process the message - metadata = ServerMessageMetadata(request_context=request) + metadata = ServerMessageMetadata( + request_context=request, + on_request_unanswered=partial(self._terminate_unanswered_request, message.id), + ) session_message = SessionMessage(message, metadata=metadata) await writer.send(session_message) try: diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 42798fdc54..0d9467ffeb 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -80,7 +80,9 @@ PeerCancelMode = Literal["interrupt", "signal"] """How `notifications/cancelled` is applied: `"interrupt"` (default) cancels -the handler's scope; `"signal"` only sets `ctx.cancel_requested`.""" +the handler's scope; `"signal"` only sets `ctx.cancel_requested` and lets the +handler run to completion. Either way the cancelled request is never +answered - the handler's eventual result or error is dropped, not written.""" def handler_exception_to_error_data(exc: BaseException) -> ErrorData | None: @@ -696,9 +698,13 @@ async def _handle_request( ) -> None: """Run `on_request` for one inbound request and write its response. - The single exception-to-wire boundary: handler exceptions become `JSONRPCError` here. + The single exception-to-wire boundary: handler exceptions become + `JSONRPCError` here. A request the peer cancelled is never answered + (spec: MUST NOT send further messages for it) - it settles unanswered + instead, and `_settle_unanswered` tells the transport. """ answer_write_started = False + handler_failure: BaseException | None = None # re-raised once the request settles try: with scope: try: @@ -711,27 +717,21 @@ async def _handle_request( key = coerce_request_id(req.id) if (entry := self._in_flight.get(key)) is not None and entry.dctx is dctx: del self._in_flight[key] - # A write interrupted by cancellation may still have delivered - # (a memory-stream send can hand its item to the receiver and - # still raise), so a started answer write counts as sent below: - # peers drop late responses, while a second answer for one id - # would break JSON-RPC. - answer_write_started = True - await self._write_result(req.id, result) - if scope.cancelled_caught: - # anyio absorbs the scope's own cancel at __exit__, and - # `cancelled_caught` (unlike `cancel_called`) guarantees the - # result write above did not happen - no double response. - # TODO(L38): spec says SHOULD NOT respond after cancel; - # the existing server always has, so match that for now. - answer_write_started = True - await self._write_error(req.id, ErrorData(code=0, message="Request cancelled")) + if not dctx.cancel_requested.is_set(): + # A write interrupted by cancellation may still have delivered + # (a memory-stream send can hand its item to the receiver and + # still raise), so a started answer write counts as sent below: + # peers drop late responses, while a second answer for one id + # would break JSON-RPC. + answer_write_started = True + await self._write_result(req.id, result) except anyio.get_cancelled_exc_class(): # Shutdown: answer the request so the peer isn't left waiting - unless # an answer write already started (it may have reached the transport; - # prefer possibly-zero answers over possibly-two). The shielded helper - # is needed because bare awaits re-raise here. - if not answer_write_started: + # prefer possibly-zero answers over possibly-two), or the peer already + # cancelled it and stopped waiting. The shielded helper is needed + # because bare awaits re-raise here. + if not answer_write_started and not dctx.cancel_requested.is_set(): await self._final_write( partial(self._write_error, req.id, ErrorData(code=CONNECTION_CLOSED, message="Connection closed")), shield=True, @@ -741,15 +741,24 @@ async def _handle_request( raise except Exception as e: error = handler_exception_to_error_data(e) - if error is not None: - await self._write_error(req.id, error) - else: + if error is None: logger.exception("handler for %r raised", req.method) # TODO(L58): code=0 pins existing-server compat; JSON-RPC says # INTERNAL_ERROR. Revisit per the suite's divergence entry. - await self._write_error(req.id, ErrorData(code=0, message=str(e))) + error = ErrorData(code=0, message=str(e)) if self._raise_handler_exceptions: - raise + handler_failure = e + # A cancel silences only the wire; the failure stays as visible as before. + if not dctx.cancel_requested.is_set(): + answer_write_started = True + await self._write_error(req.id, error) + # The one place a cancelled request settles: the handler is done (any + # mode) with nothing written. A peer-interrupt cancel is absorbed at + # scope __exit__ and lands here too. + if not answer_write_started: + await self._settle_unanswered(dctx) + if handler_failure is not None: + raise handler_failure # No `_in_flight` pop here: the inner finally covers every path, and a late pop could evict a reused id. def _allocate_id(self) -> int: @@ -771,6 +780,23 @@ async def _write_error(self, request_id: RequestId, error: ErrorData) -> None: except (anyio.BrokenResourceError, anyio.ClosedResourceError): logger.debug("dropped error for %r: write stream closed", request_id) + async def _settle_unanswered(self, dctx: _JSONRPCDispatchContext[TransportT]) -> None: + """Run the transport's `on_request_unanswered` hook: this request settled with no response. + + The dispatcher writes nothing for it; a transport whose wire must still + end the request (2025-era streamable HTTP) does so from this hook. A + raising hook is contained here, like the other callback boundaries. + """ + metadata = dctx.message_metadata + if not isinstance(metadata, ServerMessageMetadata) or metadata.on_request_unanswered is None: + return + try: + await metadata.on_request_unanswered() + except (anyio.BrokenResourceError, anyio.ClosedResourceError): + logger.debug("on_request_unanswered dropped: connection closing") + except Exception: + logger.exception("on_request_unanswered hook raised") + async def _final_write( self, write: Callable[[], Awaitable[None]], diff --git a/src/mcp/shared/message.py b/src/mcp/shared/message.py index 236569fac2..5ce255d15e 100644 --- a/src/mcp/shared/message.py +++ b/src/mcp/shared/message.py @@ -41,6 +41,10 @@ class ServerMessageMetadata: close_sse_stream: CloseSSEStreamCallback | None = None # Callback to close the standalone GET SSE stream (for unsolicited notifications) close_standalone_sse_stream: CloseSSEStreamCallback | None = None + # Callback the dispatcher runs when this request settles without a response + # (e.g. it was cancelled), for a transport whose wire must still end the + # request even though no response is written. + on_request_unanswered: Callable[[], Awaitable[None]] | None = None MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py index fbc7c13846..a26f7d9983 100644 --- a/tests/interaction/_requirements.py +++ b/tests/interaction/_requirements.py @@ -537,14 +537,16 @@ def __post_init__(self) -> None: source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#behavior-requirements", behavior=( "A cancellation notification for an in-flight request stops the server-side handler, and the " - "receiver does not send a response for the cancelled request." + "receiver does not send a response for the cancelled request - no result and no error." ), divergence=Divergence( note=( - "The spec says receivers of a cancellation SHOULD NOT send a response for the cancelled " - "request; both seats send an error response (code 0, 'Request cancelled') instead — the " - "server for cancelled client requests, and the client for cancelled server-initiated " - "requests — which is what unblocks the sender's pending call." + "The 2025-era streamable HTTP transport still answers a cancelled request, with a " + "REQUEST_CANCELLED (-32800) error - deliberate and era-scoped: that wire ends a request's " + "stream only with a response for its id, so silence would leave the POST (and any " + "resuming client's replay) open. Every other transport sends nothing, and the " + "2026-07-28 MUST NOT applies only there. Retires with the legacy transport; see " + "transport:streamable-http:cancelled-request-terminated." ), ), arm_exclusions=( @@ -2559,6 +2561,22 @@ def __post_init__(self) -> None: transports=("streamable-http",), note="Only observable over streamable HTTP: JSON-response mode is an HTTP framing option.", ), + "transport:streamable-http:cancelled-request-terminated": Requirement( + source="sdk", + behavior=( + "A request cancelled through notifications/cancelled is terminated with a REQUEST_CANCELLED " + "(-32800) error response, completing its POST - the JSON body in JSON-response mode, the " + "final event of its stream in SSE mode." + ), + transports=("streamable-http",), + note=( + "An SDK choice, not spec-mandated (the spec-side gap is the Divergence on " + "protocol:cancel:in-flight): this era's wire ends a request's stream only with a response " + "for its id, and stores it so a resuming client's replay terminates too. The terminator is " + "written through the same ordered channel as the request's other messages, so it cannot " + "overtake anything already queued for the request." + ), + ), "transport:streamable-http:stateless": Requirement( source=f"{SPEC_BASE_URL}/basic/transports#streamable-http", behavior=( diff --git a/tests/interaction/lowlevel/test_cancellation.py b/tests/interaction/lowlevel/test_cancellation.py index ecbf1a088d..74c32e304b 100644 --- a/tests/interaction/lowlevel/test_cancellation.py +++ b/tests/interaction/lowlevel/test_cancellation.py @@ -28,9 +28,10 @@ Tool, ) -from mcp import MCPError +from mcp import Client, MCPError from mcp.client import ClientRequestContext, ClientSession, IncomingMessage from mcp.server import Server, ServerRequestContext +from mcp.server.streamable_http import REQUEST_CANCELLED from mcp.shared.memory import MessageStream, create_client_server_memory_streams from mcp.shared.message import SessionMessage from tests._stamp import Unstamp @@ -39,20 +40,39 @@ pytestmark = pytest.mark.anyio +_LEGACY_HTTP_TERMINATOR = ErrorData(code=REQUEST_CANCELLED, message="Request cancelled") +"""The one wire where a cancelled request is still answered: the 2025-era streamable HTTP +transport ends a request only with a response, so it terminates the settled request with +`REQUEST_CANCELLED`. Every other transport sends nothing at all.""" + + +async def _await_doomed_call(client: Client, outcomes: list[object]) -> None: + """Await the doomed `block` call and record whatever, if anything, the caller receives. + + On the stream transports nothing ever arrives, so this parks until the task is abandoned; + over legacy streamable HTTP the transport's terminator arrives as an MCPError. + """ + try: + outcomes.append(await client.call_tool("block", {})) + except MCPError as exc: + outcomes.append(exc.error) + @requirement("protocol:cancel:in-flight") @requirement("protocol:cancel:handler-abort-propagates") async def test_cancellation_stops_in_flight_handler(connect: Connect) -> None: - """Cancelling an in-flight request interrupts its handler and fails the pending call. + """Cancelling an in-flight request interrupts its handler, and the server sends no response for it. - The server answers the cancelled request with an error response (the spec says it should - not respond at all; see the divergence note on the requirement), so the caller's pending - request raises rather than hanging. + The cancellation is scripted by hand while a sibling task still awaits the call, which is + something a well-behaved sender never does (per spec it stops waiting once it cancels). That + lets the test prove the negative: after the handler is interrupted and the connection has + quiesced, no server response has reached the still-parked call - except the legacy + streamable HTTP terminator (`_LEGACY_HTTP_TERMINATOR`). """ started = anyio.Event() handler_cancelled = anyio.Event() request_ids: list[types.RequestId] = [] - errors: list[ErrorData] = [] + outcomes: list[object] = [] async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult: assert params.name == "block" @@ -70,30 +90,26 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara async with connect(server) as client: with anyio.fail_after(5): - async with anyio.create_task_group() as task_group: - - async def call_and_capture_error() -> None: - with pytest.raises(MCPError) as exc_info: - await client.call_tool("block", {}) - errors.append(exc_info.value.error) - - task_group.start_soon(call_and_capture_error) + async with anyio.create_task_group() as task_group: # pragma: no branch + task_group.start_soon(_await_doomed_call, client, outcomes) await started.wait() await client.session.send_notification( types.CancelledNotification( params=types.CancelledNotificationParams(request_id=request_ids[0], reason="user aborted") ) ) - - await handler_cancelled.wait() - - assert errors == snapshot([ErrorData(code=0, message="Request cancelled")]) + await handler_cancelled.wait() + # Let anything the server was going to send be delivered before checking. + await anyio.wait_all_tasks_blocked() + assert outcomes in ([], [_LEGACY_HTTP_TERMINATOR]) + task_group.cancel_scope.cancel() # abandon the call if it is still parked @requirement("protocol:cancel:server-survives") async def test_session_serves_requests_after_cancellation(connect: Connect) -> None: """A request cancelled mid-flight does not poison the session: the next request succeeds.""" started = anyio.Event() + handler_cancelled = anyio.Event() request_ids: list[types.RequestId] = [] async def list_tools( @@ -112,7 +128,11 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara assert ctx.request_id is not None request_ids.append(ctx.request_id) started.set() - await anyio.Event().wait() # blocks until cancelled + try: + await anyio.Event().wait() # blocks until cancelled + except anyio.get_cancelled_exc_class(): + handler_cancelled.set() + raise raise NotImplementedError # unreachable server = Server("blocker", on_list_tools=list_tools, on_call_tool=call_tool) @@ -120,16 +140,13 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara async with connect(server) as client: with anyio.fail_after(5): async with anyio.create_task_group() as task_group: - - async def call_and_swallow_cancellation_error() -> None: - with pytest.raises(MCPError): - await client.call_tool("block", {}) - - task_group.start_soon(call_and_swallow_cancellation_error) + task_group.start_soon(_await_doomed_call, client, list[object]()) await started.wait() await client.session.send_notification( types.CancelledNotification(params=types.CancelledNotificationParams(request_id=request_ids[0])) ) + await handler_cancelled.wait() + task_group.cancel_scope.cancel() # abandon the parked call result = await client.call_tool("echo", {}) @@ -393,8 +410,9 @@ async def call_and_abandon() -> None: with anyio.fail_after(5): await handler_cancelled.wait() - # Let the abandoned call's late error response (sent on the legacy arms) arrive and be - # dropped while the client is still open, so teardown never races its delivery. + # Let anything still owed the abandoned call (the REQUEST_CANCELLED terminator over + # legacy streamable HTTP; nothing elsewhere) arrive and be dropped while the client is + # still open, so teardown never races its delivery. await anyio.wait_all_tasks_blocked() result = await client.call_tool("echo", {}) assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="ok")])) @@ -460,8 +478,9 @@ async def survivor_call() -> None: await doomed_cancelled.wait() release_survivor.set() - # Let the abandoned call's late error response (sent on the legacy arms) arrive and be - # dropped while the client is still open, so teardown never races its delivery. + # Let anything still owed the abandoned call (the REQUEST_CANCELLED terminator over + # legacy streamable HTTP; nothing elsewhere) arrive and be dropped while the client is + # still open, so teardown never races its delivery. await anyio.wait_all_tasks_blocked() assert [unstamped(result) for result in results] == snapshot( diff --git a/tests/interaction/lowlevel/test_wire.py b/tests/interaction/lowlevel/test_wire.py index b3d286ca1d..8ee05fec38 100644 --- a/tests/interaction/lowlevel/test_wire.py +++ b/tests/interaction/lowlevel/test_wire.py @@ -353,8 +353,8 @@ async def call_and_abandon() -> None: with anyio.fail_after(5): await handler_cancelled.wait() - # Let the cancelled call's late error response arrive and be dropped while the client - # is still open, so teardown never races its delivery. + # Let any in-flight delivery for the abandoned call settle while the client is still + # open, so teardown never races it (nothing arrives here: a cancelled request is not answered). await anyio.wait_all_tasks_blocked() call, cancel = [message.message for message in recording.sent] diff --git a/tests/interaction/transports/test_streamable_http.py b/tests/interaction/transports/test_streamable_http.py index cb22e7ab87..e1f16dc5f0 100644 --- a/tests/interaction/transports/test_streamable_http.py +++ b/tests/interaction/transports/test_streamable_http.py @@ -12,22 +12,37 @@ from inline_snapshot import snapshot from mcp_types import ( INVALID_REQUEST, + CallToolRequestParams, CallToolResult, ElicitRequestParams, ElicitResult, + ErrorData, + JSONRPCError, + JSONRPCMessage, + JSONRPCRequest, LoggingMessageNotification, LoggingMessageNotificationParams, ResourceUpdatedNotification, ResourceUpdatedNotificationParams, TextContent, + jsonrpc_message_adapter, ) from pydantic import BaseModel from mcp.client import ClientRequestContext, IncomingMessage +from mcp.server import Server, ServerRequestContext from mcp.server.elicitation import AcceptedElicitation from mcp.server.mcpserver import Context, MCPServer +from mcp.server.streamable_http import REQUEST_CANCELLED from mcp.shared.exceptions import MCPError -from tests.interaction._connect import connect_over_streamable_http +from tests.interaction._connect import ( + base_headers, + connect_over_streamable_http, + initialize_body, + initialize_via_http, + mounted_app, + post_jsonrpc, +) from tests.interaction._requirements import requirement pytestmark = pytest.mark.anyio @@ -168,3 +183,80 @@ async def answer(context: ClientRequestContext, params: ElicitRequestParams) -> CallToolResult(content=[TextContent(text="confirmed=True")], structured_content={"result": "confirmed=True"}) ) assert [params.message for params in asked] == snapshot(["Proceed?"]) + + +@requirement("transport:streamable-http:cancelled-request-terminated") +@pytest.mark.parametrize("json_response", [True, False], ids=["json-response", "sse-response"]) +async def test_cancelled_request_is_terminated_with_request_cancelled(json_response: bool) -> None: + """A cancelled request's POST completes carrying the `REQUEST_CANCELLED` terminal error. + + The 2025-era wire ends a request only with a response, so this transport answers the + settled-unanswered request with `REQUEST_CANCELLED` (the dispatcher writes nothing on the + other transports). Driven with raw httpx2 because the observable is the HTTP exchange + itself, which a Client abandoning its own call would tear down first. + """ + handler_started = anyio.Event() + handler_cancelled = anyio.Event() + call_request_id = 2 + + async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult: + handler_started.set() + try: + await anyio.sleep_forever() + except anyio.get_cancelled_exc_class(): + handler_cancelled.set() + raise + raise NotImplementedError # unreachable: only cancellation ends the sleep + + server = Server("blocker", on_call_tool=call_tool) + call_body = JSONRPCRequest( + jsonrpc="2.0", + id=call_request_id, + method="tools/call", + params=CallToolRequestParams(name="block", arguments={}).model_dump(by_alias=True, mode="json"), + ).model_dump(by_alias=True, exclude_none=True) + cancel_body = { + "jsonrpc": "2.0", + "method": "notifications/cancelled", + "params": {"requestId": call_request_id}, + } + call_answers: list[JSONRPCMessage] = [] + + async with mounted_app(server, json_response=json_response) as (http, _manager): + if json_response: + # The SSE-reading handshake helper does not apply: JSON mode answers initialize with JSON. + initialized = await http.post("/mcp", json=initialize_body(), headers=base_headers()) + session_id = initialized.headers["mcp-session-id"] + ready = await http.post( + "/mcp", + json={"jsonrpc": "2.0", "method": "notifications/initialized"}, + headers=base_headers(session_id=session_id), + ) + assert ready.status_code == 202 + else: + session_id = await initialize_via_http(http) + + async def post_call() -> None: + if json_response: + response = await http.post("/mcp", json=call_body, headers=base_headers(session_id=session_id)) + call_answers.append(jsonrpc_message_adapter.validate_json(response.content)) + else: + _, messages = await post_jsonrpc(http, call_body, session_id=session_id) + call_answers.extend(messages) + + with anyio.fail_after(5): + async with anyio.create_task_group() as task_group: # pragma: no branch + task_group.start_soon(post_call) + await handler_started.wait() + cancelled = await http.post("/mcp", json=cancel_body, headers=base_headers(session_id=session_id)) + assert cancelled.status_code == 202 + await handler_cancelled.wait() + # The call's POST must now complete on its own; the task group waits for it. + + assert call_answers == [ + JSONRPCError( + jsonrpc="2.0", + id=call_request_id, + error=ErrorData(code=REQUEST_CANCELLED, message="Request cancelled"), + ) + ] diff --git a/tests/server/test_cancel_handling.py b/tests/server/test_cancel_handling.py index 3d32adb3c8..fd0e4b28af 100644 --- a/tests/server/test_cancel_handling.py +++ b/tests/server/test_cancel_handling.py @@ -22,7 +22,6 @@ from mcp import Client from mcp.server import Server, ServerRequestContext -from mcp.shared.exceptions import MCPError from mcp.shared.message import SessionMessage @@ -33,6 +32,7 @@ async def test_server_remains_functional_after_cancel(): # Track tool calls call_count = 0 ev_first_call = anyio.Event() + ev_first_call_cancelled = anyio.Event() first_request_id = None async def handle_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult: @@ -53,38 +53,45 @@ async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestPar if call_count == 1: first_request_id = ctx.request_id ev_first_call.set() - await anyio.sleep(5) # First call is slow + try: + await anyio.sleep_forever() # First call blocks until cancelled + except anyio.get_cancelled_exc_class(): + ev_first_call_cancelled.set() + raise return CallToolResult(content=[TextContent(type="text", text=f"Call number: {call_count}")]) raise ValueError(f"Unknown tool: {params.name}") # pragma: no cover server = Server("test-server", on_list_tools=handle_list_tools, on_call_tool=handle_call_tool) async with Client(server, mode="legacy") as client: - # First request (will be cancelled) + # First request (will be cancelled server-side, then abandoned here: a + # cancelled request is never answered, so nothing would wake this call) async def first_request(): - try: - await client.session.send_request( - CallToolRequest(params=CallToolRequestParams(name="test_tool", arguments={})), - CallToolResult, - ) - pytest.fail("First request should have been cancelled") # pragma: no cover - except MCPError: - pass # Expected + await client.session.send_request( + CallToolRequest(params=CallToolRequestParams(name="test_tool", arguments={})), + CallToolResult, + ) + raise NotImplementedError # unreachable: the task is cancelled before any answer # Start first request - async with anyio.create_task_group() as tg: - tg.start_soon(first_request) - - # Wait for it to start - await ev_first_call.wait() - - # Cancel it - assert first_request_id is not None - await client.session.send_notification( - CancelledNotification( - params=CancelledNotificationParams(request_id=first_request_id, reason="Testing server recovery"), + with anyio.fail_after(5): + async with anyio.create_task_group() as tg: + tg.start_soon(first_request) + + # Wait for it to start + await ev_first_call.wait() + + # Cancel it + assert first_request_id is not None + await client.session.send_notification( + CancelledNotification( + params=CancelledNotificationParams( + request_id=first_request_id, reason="Testing server recovery" + ), + ) ) - ) + await ev_first_call_cancelled.wait() + tg.cancel_scope.cancel() # abandon the parked call # Second request (should work normally) result = await client.call_tool("test_tool", {}) diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 5c29c7e3ce..51f7f7c820 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -22,6 +22,7 @@ CancelledNotificationParams, ErrorData, JSONRPCError, + JSONRPCMessage, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse, @@ -34,10 +35,11 @@ from mcp.server import Server, ServerRequestContext from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream -from mcp.shared.dispatcher import CallOptions, DispatchContext, coerce_request_id +from mcp.shared.dispatcher import CallOptions, DispatchContext, OnRequest, coerce_request_id from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import ( # pyright: ignore[reportPrivateUsage] JSONRPCDispatcher, + PeerCancelMode, _OutboundPlan, _Pending, _plan_outbound, @@ -122,113 +124,171 @@ async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | assert exc.value.__cause__ is None # cause does not survive the wire -@pytest.mark.anyio -async def test_peer_cancel_interrupt_mode_writes_cancelled_error_response(): - """Matches the existing server: a peer-cancelled request is answered with code=0.""" +async def _drive_cancelled_request( + on_request: OnRequest, *, peer_cancel_mode: PeerCancelMode = "interrupt", done: anyio.Event +) -> tuple[list[JSONRPCMessage], list[RequestId]]: + """Send request 1, cancel it, then send an uncancelled control request 2. + + Returns (answers written, ids settled unanswered). The control request proves the write + path is live, so an empty answer for id 1 is the suppression under test. Both requests + carry an `on_request_unanswered` callback; only cancelled request 1 should fire it. + Server-only, since the cancelled request is never answered. `done` is set by the + handler once request 1 reaches the state under test. + """ + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4) + recording = RecordingWriteStream() + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( + c2s_recv, recording, peer_cancel_mode=peer_cancel_mode + ) handler_started = anyio.Event() + unanswered: list[RequestId] = [] + + async def on_request_with_control(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + if method == "control": + return {"control": True} + handler_started.set() + return await on_request(ctx, method, params) + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + pass # the cancelled notification is teed here; nothing to observe + + def request(request_id: RequestId, method: str) -> SessionMessage: + async def on_unanswered() -> None: + unanswered.append(request_id) + + return SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=request_id, method=method, params=None), + metadata=ServerMessageMetadata(on_request_unanswered=on_unanswered), + ) + + cancel = JSONRPCNotification(jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 1}) + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request_with_control, on_notify) + await c2s_send.send(request(1, "t")) + with anyio.fail_after(5): + await handler_started.wait() + await c2s_send.send(SessionMessage(message=cancel)) + await done.wait() + await c2s_send.send(request(2, "control")) + # Quiesce: let both handler tasks run to the end of `_handle_request`, + # so every write they were going to make has been recorded. + await anyio.wait_all_tasks_blocked() + tg.cancel_scope.cancel() + finally: + c2s_send.close() + c2s_recv.close() + return [m.message for m in recording.sent], unanswered + + +_CONTROL_ONLY: list[JSONRPCMessage] = [JSONRPCResponse(jsonrpc="2.0", id=2, result={"control": True})] +"""Everything on the wire after a cancelled request 1: only the uncancelled control request 2 is answered.""" + + +@pytest.mark.anyio +async def test_peer_cancel_interrupt_mode_interrupts_handler_and_writes_no_response(): + """Spec MUST NOT: a cancelled request is never answered - not even with an error - and settles unanswered.""" handler_exited = anyio.Event() seen_ctx: list[DCtx] = [] - async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: seen_ctx.append(ctx) - handler_started.set() try: await anyio.sleep_forever() finally: handler_exited.set() raise NotImplementedError - seen_error: list[ErrorData] = [] - async with running_pair(jsonrpc_pair, server_on_request=server_on_request) as (client, *_): - with anyio.fail_after(5): - async with anyio.create_task_group() as tg: # pragma: no branch + assert await _drive_cancelled_request(on_request, done=handler_exited) == (_CONTROL_ONLY, [1]) + assert seen_ctx[0].cancel_requested.is_set() - async def call_then_record() -> None: - with pytest.raises(MCPError) as exc: - await client.send_raw_request("slow", None) - seen_error.append(exc.value.error) - tg.start_soon(call_then_record) - await handler_started.wait() - await client.notify("notifications/cancelled", {"requestId": 1}) - await handler_exited.wait() - assert seen_ctx[0].cancel_requested.is_set() - assert seen_error == [ErrorData(code=0, message="Request cancelled")] +@pytest.mark.anyio +async def test_peer_cancel_signal_mode_sets_event_and_drops_the_completed_handlers_result(): + """`"signal"` mode lets the handler run to completion, but the cancelled request is still not answered.""" + cancel_seen = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + await ctx.cancel_requested.wait() + cancel_seen.set() + return {"finished": True} + + assert await _drive_cancelled_request(on_request, peer_cancel_mode="signal", done=cancel_seen) == ( + _CONTROL_ONLY, + [1], + ) @pytest.mark.anyio -async def test_peer_cancel_landing_after_handlers_last_checkpoint_writes_only_the_result(): - """A peer cancel that fails to interrupt the handler writes only the result: one answer per - id goes on the wire (SDK-defined). The recording stream is needed because a memory stream's - `send` checkpoints, letting the deferred cancellation land mid-write and hide a double answer.""" +@pytest.mark.parametrize( + "handler_failure", + [RuntimeError("cleanup failed"), MCPError(code=INTERNAL_ERROR, message="cleanup failed")], + ids=["unmapped-exception", "mcp-error"], +) +async def test_peer_cancel_drops_the_error_of_a_handler_that_fails_after_cancel(handler_failure: Exception): + """A handler that turns its cancellation into an exception - mapped or not - writes no error response either.""" + handler_failed = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + try: + await anyio.sleep_forever() + except anyio.get_cancelled_exc_class(): + handler_failed.set() + raise handler_failure from None + raise NotImplementedError + + assert await _drive_cancelled_request(on_request, done=handler_failed) == (_CONTROL_ONLY, [1]) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "hook_error", + [RuntimeError("hook failed"), anyio.ClosedResourceError()], + ids=["hook-bug", "connection-closing"], +) +async def test_a_raising_unanswered_hook_is_contained(hook_error: Exception): + """A transport `on_request_unanswered` hook that raises is contained, not fatal: the + dispatcher keeps serving (the control request written after it is still answered).""" c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4) recording = RecordingWriteStream() server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, recording) - handler_started = anyio.Event() + handler_exited = anyio.Event() + + async def failing_hook() -> None: + raise hook_error async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: - handler_started.set() - await ctx.cancel_requested.wait() - return {"completed": "after-cancel"} + if method == "control": + return {"control": True} + try: + await anyio.sleep_forever() + finally: + handler_exited.set() + raise NotImplementedError async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: - pass # the cancelled notification is teed here; nothing to observe + pass + request_1 = SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=1, method="t", params=None), + metadata=ServerMessageMetadata(on_request_unanswered=failing_hook), + ) + cancel = JSONRPCNotification(jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 1}) + control = SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=2, method="control", params=None)) try: async with anyio.create_task_group() as tg: await tg.start(server.run, on_request, on_notify) - await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=1, method="t", params=None))) with anyio.fail_after(5): - await handler_started.wait() - # The cancel is also the handler's wakeup, so anyio defers it and the handler completes. - await c2s_send.send( - SessionMessage( - message=JSONRPCNotification( - jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 1} - ) - ) - ) - # Quiesce: the handler has resumed, completed, and exited its scope. + await c2s_send.send(request_1) + await c2s_send.send(SessionMessage(message=cancel)) + await handler_exited.wait() + await c2s_send.send(control) await anyio.wait_all_tasks_blocked() tg.cancel_scope.cancel() finally: c2s_send.close() c2s_recv.close() - assert [m.message for m in recording.sent] == [ - JSONRPCResponse(jsonrpc="2.0", id=1, result={"completed": "after-cancel"}) - ] - - -@pytest.mark.anyio -async def test_peer_cancel_signal_mode_sets_event_but_handler_runs_to_completion(): - handler_started = anyio.Event() - cancel_seen = anyio.Event() - - async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: - handler_started.set() - await ctx.cancel_requested.wait() - cancel_seen.set() - return {"finished": True} - - def factory(*, can_send_request: bool = True): - client, server, close = jsonrpc_pair(can_send_request=can_send_request) - assert isinstance(server, JSONRPCDispatcher) - server._peer_cancel_mode = "signal" # pyright: ignore[reportPrivateUsage] - return client, server, close - - result_box: list[dict[str, Any]] = [] - async with running_pair(factory, server_on_request=server_on_request) as (client, *_): - with anyio.fail_after(5): - async with anyio.create_task_group() as tg: # pragma: no branch - - async def call() -> None: - result_box.append(await client.send_raw_request("slow", None)) - - tg.start_soon(call) - await handler_started.wait() - await client.notify("notifications/cancelled", {"requestId": 1}) - await cancel_seen.wait() - assert result_box == [{"finished": True}] + assert [m.message for m in recording.sent] == _CONTROL_ONLY @pytest.mark.anyio @@ -1592,14 +1652,16 @@ async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | async with anyio.create_task_group() as tg: # pragma: no branch async def call() -> None: - with pytest.raises(MCPError): - await client.send_raw_request("slow", None) + # Never answered; abandoned below without a courtesy cancel so `srec` sees only the peer's. + await client.send_raw_request("slow", None, {"cancel_on_abandon": False}) + raise NotImplementedError # unreachable: the task is cancelled first tg.start_soon(call) await handler_started.wait() await client.notify("notifications/cancelled", {"requestId": 1}) await handler_exited.wait() await srec.notified.wait() + tg.cancel_scope.cancel() # abandon the parked call assert srec.notifications == [("notifications/cancelled", {"requestId": 1})] @@ -2057,8 +2119,8 @@ async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | async with anyio.create_task_group() as tg: # pragma: no branch async def call() -> None: - with pytest.raises(MCPError): - await client.send_raw_request("slow", None) + await client.send_raw_request("slow", None) # never answered; abandoned below + raise NotImplementedError # unreachable: the task is cancelled first tg.start_soon(call) await handler_started.wait() @@ -2068,6 +2130,7 @@ async def call() -> None: assert not handler_exited.is_set() await client.notify("notifications/cancelled", {"requestId": 1}) await handler_exited.wait() + tg.cancel_scope.cancel() # abandon the parked call @pytest.mark.anyio @@ -2160,11 +2223,15 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> async def test_cancelled_correlates_across_string_and_int_request_id_forms(request_id: RequestId, cancel_id: object): """A peer that stringifies the id between request and cancel still cancels (same `coerce_request_id` path).""" c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) - s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) - server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + recording = RecordingWriteStream() + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, recording) + handler_interrupted = anyio.Event() async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: - await anyio.sleep_forever() + try: + await anyio.sleep_forever() + finally: + handler_interrupted.set() raise NotImplementedError async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: @@ -2184,15 +2251,12 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> ) ) with anyio.fail_after(5): - resp = await s2c_recv.receive() - assert isinstance(resp, SessionMessage) - assert isinstance(resp.message, JSONRPCError) - assert resp.message.id == request_id # response echoes the peer's id form verbatim - assert resp.message.error == ErrorData(code=0, message="Request cancelled") + await handler_interrupted.wait() # the cancel reached the handler despite the id form tg.cancel_scope.cancel() finally: - for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): - s.close() + c2s_send.close() + c2s_recv.close() + assert recording.sent == [] # cancelled: no response, in either id form @pytest.mark.anyio @@ -2245,11 +2309,7 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> ) ) ) - resp2 = await s2c_recv.receive() - assert isinstance(resp2, SessionMessage) - assert isinstance(resp2.message, JSONRPCError) - assert resp2.message.error == ErrorData(code=0, message="Request cancelled") - assert second_exited.is_set() + await second_exited.wait() # the cancel reached the surviving entry tg.cancel_scope.cancel() finally: for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): @@ -2308,11 +2368,7 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> ) ) ) - resp2 = await s2c_recv.receive() - assert isinstance(resp2, SessionMessage) - assert isinstance(resp2.message, JSONRPCError) - assert resp2.message.error == ErrorData(code=0, message="Request cancelled") - assert second_exited.is_set() + await second_exited.wait() # the cancel reached the surviving entry tg.cancel_scope.cancel() finally: for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): @@ -2366,11 +2422,12 @@ async def observe(ctx: Any, call_next: Any) -> Any: async with anyio.create_task_group() as tg: # pragma: no branch async def call() -> None: - with pytest.raises(MCPError): - await client.session.send_request( - CallToolRequest(params=CallToolRequestParams(name="t", arguments={})), - CallToolResult, - ) + # Never answered once cancelled; abandoned below. + await client.session.send_request( + CallToolRequest(params=CallToolRequestParams(name="t", arguments={})), + CallToolResult, + ) + raise NotImplementedError # unreachable: the task is cancelled first tg.start_soon(call) await handler_started.wait() @@ -2381,10 +2438,8 @@ async def call() -> None: ) ) await cancel_observed.wait() - assert len(observed) == 1 - assert observed[0][0] == "notifications/cancelled" - assert observed[0][1]["requestId"] == request_id - assert observed[0][1]["reason"] == "user clicked stop" + tg.cancel_scope.cancel() # abandon the parked call (sends its own courtesy cancel) + assert observed[0] == ("notifications/cancelled", {"requestId": request_id, "reason": "user clicked stop"}) @pytest.mark.anyio