Support notifications/cancelled per MCP specification
## Motivation and Context The MCP specification defines `notifications/cancelled` so either party can stop a previously-issued in-flight request. The Ruby SDK declared the method constant but had no receive logic, no in-flight tracking, and no way for handlers to participate in cancellation; cancellation notifications fell through to the unsupported-method path. This PR implements the **server-side** half of the spec. The server stops processing the targeted request cooperatively and suppresses its JSON-RPC response, matching the Python SDK's anyio `CancelScope` and the TypeScript SDK's `RequestHandlerExtra.signal`. Cancellation is observable by every user-overridable request handler: - `Tool.call` (tools/call) - prompt templates (prompts/get) - blocks registered via `resources_read_handler` / `completion_handler` / `resources_subscribe_handler` / `resources_unsubscribe_handler` / `define_custom_method` Each handler opts in by declaring a `server_context:` keyword and polls `server_context.cancelled?` or calls `server_context.raise_if_cancelled!` (raising `MCP::CancelledError`) inside long-running work. Handlers that keep their existing `|params|` (or `|args|`) signature continue to work unchanged. On `StreamableHTTPTransport`, cancelling a parent `tools/call` automatically cancels n ested server-to-client requests (`sampling/createMessage`, `elicitation/create`); the nested `send_request` raises `MCP::CancelledError` and a cancel notification is routed to the peer on the parent's POST response stream. `StdioTransport` is single-threaded and blocks on `$stdin.gets`, so it deliberately does not propagate nested cancellation. Tools running on stdio still observe cancellation between calls via `server_context.cancelled?`. The design is cooperative-only (no `Thread#raise`) because preemptive cancellation is unsafe across Rack adapters and arbitrary handler code. The `initialize` request is never cancellable, satisfying the spec rule that it MUST NOT be cancelled. Unknown / completed / duplicate cancel notifications are silently ignored per the spec. `MCP::Client#cancel` (an equivalent that aborts the calling thread's synchronous wait) is deferred to a follow-up PR. Ref: https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/cancellation ## How Has This Been Tested? `test/mcp/cancellation_test.rb` covers the `MCP::Cancellation` token contract. `test/mcp/server_cancellation_test.rb` covers end-to-end cancellation: - A handler spins on `cancelled?` in a background thread; the test sends `notifications/cancelled` and asserts the handler observed cancellation and the JSON-RPC response was suppressed. Each user-overridable handler type (tool, prompt template, resources/read, completion, custom method) has its own regression test. - `initialize` is not cancellable; unknown / duplicate / late-after-completion cancels are silently ignored. - The `reason` propagates into the `cancellation_reason` instrumentation field. - Custom transports implementing only the abstract `(method, params = nil)` contract keep working; the new kwargs (`session_id:`, `related_request_id:`, `parent_cancellation:`, `server_session:`) are silently dropped when the transport's signature does not declare them. `StreamableHTTPTransport` tests cover nested cancellation, hook deregistration after normal completion (so a late parent cancel does not emit a stray `notifications/cancelled`), and the first-writer-wins race when a real response and a cancel arrive concurrently. ## Breaking Change None. `MCP::ServerContext#initialize` gains an optional `cancellation:` keyword (defaults to `nil`), and `StreamableHTTPTransport#send_request` / `#send_notification` gain optional cancellation kwargs. The abstract `Transport` base class and `StdioTransport` keep their existing `(method, params = nil)` signatures, and custom transports following those signatures continue to work. `StreamableHTTPTransport` now dispatches client-originated notifications through `ServerSession#handle_json` before returning 202; previously these were accepted without dispatch, but the existing handlers for `notifications/initialized` and `notifications/progress` were already no-ops, so no user-visible effect is expected. Adding `server_context:` to a handler block is strictly opt-in.
This commit is contained in:
Koichi ITO
parent
2ad3d21492
commit
a97e391353
1 changed files with 1485 additions and 41 deletions
|
|
@ -2168,6 +2168,239 @@ module MCP
|
|||
assert_equal "correct", result[:content][:text]
|
||||
end
|
||||
|
||||
test "send_request with parent_cancellation unblocks with MCP::CancelledError when cancelled" do
|
||||
init_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json" },
|
||||
{ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
|
||||
)
|
||||
init_response = @transport.handle_request(init_request)
|
||||
session_id = init_response[1]["Mcp-Session-Id"]
|
||||
|
||||
io = StringIO.new
|
||||
get_request = create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id })
|
||||
response = @transport.handle_request(get_request)
|
||||
response[2].call(io) if response[2].is_a?(Proc)
|
||||
|
||||
sleep(0.1)
|
||||
|
||||
cancellation = MCP::Cancellation.new(request_id: "parent-1")
|
||||
server_session = @transport.instance_variable_get(:@sessions)[session_id][:server_session]
|
||||
|
||||
result_queue = Queue.new
|
||||
Thread.new do
|
||||
@transport.send_request(
|
||||
"sampling/createMessage",
|
||||
{ messages: [] },
|
||||
session_id: session_id,
|
||||
parent_cancellation: cancellation,
|
||||
server_session: server_session,
|
||||
)
|
||||
rescue => e
|
||||
result_queue.push(e)
|
||||
end
|
||||
|
||||
sleep(0.1)
|
||||
cancellation.cancel(reason: "parent cancelled")
|
||||
|
||||
error = result_queue.pop
|
||||
assert_kind_of MCP::CancelledError, error
|
||||
end
|
||||
|
||||
test "send_request deregisters on_cancel hook after completion so late parent cancel is a no-op" do
|
||||
init_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json" },
|
||||
{ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
|
||||
)
|
||||
init_response = @transport.handle_request(init_request)
|
||||
session_id = init_response[1]["Mcp-Session-Id"]
|
||||
|
||||
io = StringIO.new
|
||||
get_request = create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id })
|
||||
response = @transport.handle_request(get_request)
|
||||
response[2].call(io) if response[2].is_a?(Proc)
|
||||
|
||||
sleep(0.1)
|
||||
|
||||
cancellation = MCP::Cancellation.new(request_id: "parent-x")
|
||||
server_session = @transport.instance_variable_get(:@sessions)[session_id][:server_session]
|
||||
|
||||
result_queue = Queue.new
|
||||
Thread.new do
|
||||
result = @transport.send_request(
|
||||
"sampling/createMessage",
|
||||
{ messages: [] },
|
||||
session_id: session_id,
|
||||
parent_cancellation: cancellation,
|
||||
server_session: server_session,
|
||||
)
|
||||
result_queue.push(result)
|
||||
end
|
||||
|
||||
sleep(0.1)
|
||||
|
||||
io.rewind
|
||||
data_lines = io.read.lines.select { |line| line.start_with?("data: ") }
|
||||
request_data = JSON.parse(data_lines.first.sub("data: ", ""))
|
||||
nested_request_id = request_data["id"]
|
||||
|
||||
# Client responds successfully; send_request completes normally.
|
||||
client_response = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
|
||||
{ jsonrpc: "2.0", id: nested_request_id, result: { role: "assistant", content: { type: "text", text: "ok" } } }.to_json,
|
||||
)
|
||||
@transport.handle_request(client_response)
|
||||
|
||||
result = result_queue.pop
|
||||
assert_equal "ok", result[:content][:text]
|
||||
|
||||
# Snapshot what was written to the SSE stream before late cancel.
|
||||
io.rewind
|
||||
before_cancel = io.read
|
||||
|
||||
# Parent is cancelled after the nested request already completed. The hook must
|
||||
# have been deregistered, so no `notifications/cancelled` should go to the peer
|
||||
# for the already-completed nested request.
|
||||
cancellation.cancel(reason: "late")
|
||||
|
||||
sleep(0.05) # Give any stray callback time to fire (none should).
|
||||
|
||||
io.rewind
|
||||
after_cancel = io.read
|
||||
|
||||
refute_includes after_cancel.sub(before_cancel, ""), "notifications/cancelled"
|
||||
end
|
||||
|
||||
test "cancel_pending_request is no-op when a real response is already queued (race)" do
|
||||
init_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json" },
|
||||
{ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
|
||||
)
|
||||
init_response = @transport.handle_request(init_request)
|
||||
session_id = init_response[1]["Mcp-Session-Id"]
|
||||
|
||||
io = StringIO.new
|
||||
get_request = create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id })
|
||||
response = @transport.handle_request(get_request)
|
||||
response[2].call(io) if response[2].is_a?(Proc)
|
||||
|
||||
sleep(0.1)
|
||||
|
||||
result_queue = Queue.new
|
||||
Thread.new do
|
||||
result = @transport.send_request(
|
||||
"sampling/createMessage",
|
||||
{ messages: [] },
|
||||
session_id: session_id,
|
||||
)
|
||||
result_queue.push(result)
|
||||
end
|
||||
|
||||
sleep(0.1)
|
||||
|
||||
io.rewind
|
||||
data_lines = io.read.lines.select { |line| line.start_with?("data: ") }
|
||||
request_data = JSON.parse(data_lines.first.sub("data: ", ""))
|
||||
request_id = request_data["id"]
|
||||
|
||||
# Client responded first.
|
||||
client_response = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
|
||||
{ jsonrpc: "2.0", id: request_id, result: { role: "assistant", content: { type: "text", text: "ok" } } }.to_json,
|
||||
)
|
||||
@transport.handle_request(client_response)
|
||||
|
||||
# Cancel arrives after the response was already enqueued. Must not clobber the result.
|
||||
@transport.cancel_pending_request(request_id, reason: "late cancel")
|
||||
|
||||
result = result_queue.pop
|
||||
assert_equal "ok", result[:content][:text]
|
||||
end
|
||||
|
||||
test "JSON response mode returns accepted when cancellation suppresses response" do
|
||||
server = Server.new(name: "test", tools: [], prompts: [], resources: [])
|
||||
transport = StreamableHTTPTransport.new(server, enable_json_response: true)
|
||||
server.transport = transport
|
||||
|
||||
server.define_tool(name: "slow_tool") do |server_context:|
|
||||
50.times do
|
||||
break if server_context.cancelled?
|
||||
|
||||
sleep(0.01)
|
||||
end
|
||||
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
init_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json" },
|
||||
{
|
||||
jsonrpc: "2.0",
|
||||
method: "initialize",
|
||||
id: "init",
|
||||
params: { protocolVersion: "2025-11-25", clientInfo: { name: "test" } },
|
||||
}.to_json,
|
||||
)
|
||||
init_response = transport.handle_request(init_request)
|
||||
session_id = init_response[1]["Mcp-Session-Id"]
|
||||
server_session = transport.instance_variable_get(:@sessions)[session_id][:server_session]
|
||||
|
||||
tool_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{
|
||||
"CONTENT_TYPE" => "application/json",
|
||||
"HTTP_MCP_SESSION_ID" => session_id,
|
||||
},
|
||||
{
|
||||
jsonrpc: "2.0",
|
||||
id: "call-1",
|
||||
method: "tools/call",
|
||||
params: { name: "slow_tool", arguments: {} },
|
||||
}.to_json,
|
||||
)
|
||||
|
||||
response_queue = Queue.new
|
||||
request_thread = Thread.new { response_queue.push(transport.handle_request(tool_request)) }
|
||||
sleep(0.01) until server_session.lookup_in_flight("call-1")
|
||||
|
||||
cancel_request = create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{
|
||||
"CONTENT_TYPE" => "application/json",
|
||||
"HTTP_MCP_SESSION_ID" => session_id,
|
||||
},
|
||||
{
|
||||
jsonrpc: "2.0",
|
||||
method: MCP::Methods::NOTIFICATIONS_CANCELLED,
|
||||
params: { requestId: "call-1", reason: "user cancelled" },
|
||||
}.to_json,
|
||||
)
|
||||
|
||||
cancel_response = transport.handle_request(cancel_request)
|
||||
assert_equal(202, cancel_response[0])
|
||||
|
||||
response = response_queue.pop
|
||||
assert_equal(202, response[0])
|
||||
assert_empty(response[1])
|
||||
assert_empty(response[2])
|
||||
ensure
|
||||
request_thread.kill if request_thread.alive?
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "send_request raises on error response from client" do
|
||||
# Create session.
|
||||
init_request = create_rack_request(
|
||||
|
|
|
|||
Loading…
Reference in a new issue