Stop answering cancelled requests (#3188)
This commit is contained in:
parent
e8ef138153
commit
923341c98a
11 changed files with 464 additions and 204 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -65,6 +65,13 @@ GET_STREAM_KEY = "_GET_stream"
|
|||
# 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 @@ class StreamableHTTPServerTransport:
|
|||
|
||||
def _create_session_message(
|
||||
self,
|
||||
message: JSONRPCMessage,
|
||||
message: JSONRPCRequest,
|
||||
request: Request,
|
||||
request_id: RequestId,
|
||||
protocol_version: str,
|
||||
|
|
@ -262,7 +269,10 @@ class StreamableHTTPServerTransport:
|
|||
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 @@ class StreamableHTTPServerTransport:
|
|||
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 @@ class StreamableHTTPServerTransport:
|
|||
|
||||
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 @@ class StreamableHTTPServerTransport:
|
|||
)
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -80,7 +80,9 @@ TransportT = TypeVar("TransportT", bound=TransportContext, default=TransportCont
|
|||
|
||||
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 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
|
|||
) -> 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 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
|
|||
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 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
|
|||
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 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
|
|||
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]],
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -537,14 +537,16 @@ REQUIREMENTS: dict[str, Requirement] = {
|
|||
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 @@ REQUIREMENTS: dict[str, Requirement] = {
|
|||
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=(
|
||||
|
|
|
|||
|
|
@ -28,9 +28,10 @@ from mcp_types import (
|
|||
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 @@ from tests.interaction._requirements import requirement
|
|||
|
||||
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 test_cancellation_stops_in_flight_handler(connect: Connect) -> None:
|
|||
|
||||
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 test_session_serves_requests_after_cancellation(connect: Connect) -> N
|
|||
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 test_session_serves_requests_after_cancellation(connect: Connect) -> N
|
|||
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 test_abandoning_a_call_stops_the_server_handler(connect: Connect, unst
|
|||
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 test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Con
|
|||
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(
|
||||
|
|
|
|||
|
|
@ -353,8 +353,8 @@ async def test_abandoning_a_call_on_a_modern_stream_wire_sends_one_cancelled_fra
|
|||
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]
|
||||
|
|
|
|||
|
|
@ -12,22 +12,37 @@ import pytest
|
|||
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 test_server_initiated_elicitation_round_trips_during_a_tool_call() ->
|
|||
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"),
|
||||
)
|
||||
]
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@ from mcp_types.version import LATEST_HANDSHAKE_VERSION
|
|||
|
||||
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 test_server_remains_functional_after_cancel():
|
|||
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)
|
||||
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()
|
||||
# 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"),
|
||||
# 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", {})
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from mcp_types import (
|
|||
CancelledNotificationParams,
|
||||
ErrorData,
|
||||
JSONRPCError,
|
||||
JSONRPCMessage,
|
||||
JSONRPCNotification,
|
||||
JSONRPCRequest,
|
||||
JSONRPCResponse,
|
||||
|
|
@ -34,10 +35,11 @@ from mcp import Client
|
|||
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 test_handler_raising_exception_sends_code_zero_with_str_message():
|
|||
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
|
||||
|
||||
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 await _drive_cancelled_request(on_request, done=handler_exited) == (_CONTROL_ONLY, [1])
|
||||
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_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."""
|
||||
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
|
||||
@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 test_cancelled_notification_for_in_flight_request_is_teed_to_on_notify
|
|||
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 test_cancelled_with_bool_request_id_does_not_cancel_request_one():
|
|||
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 test_cancelled_with_bool_request_id_does_not_cancel_request_one():
|
|||
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 test_request_with_bool_meta_progress_token_is_not_adopted():
|
|||
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 test_cancelled_correlates_across_string_and_int_request_id_forms(reque
|
|||
)
|
||||
)
|
||||
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 test_completed_handler_does_not_evict_reused_request_id_from_in_flight
|
|||
)
|
||||
)
|
||||
)
|
||||
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 test_duplicate_request_id_completion_of_first_handler_keeps_second_can
|
|||
)
|
||||
)
|
||||
)
|
||||
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 test_server_middleware_observes_cancelled_notification():
|
|||
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 test_server_middleware_observes_cancelled_notification():
|
|||
)
|
||||
)
|
||||
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
|
||||
|
|
|
|||
Loading…
Reference in a new issue