Merge pull request #502 from koic/add_server_request_timeout
Bound server-to-client requests with a timeout
This commit is contained in:
Koichi ITO
GitHub
commit
0575dc0de8
9 changed files with 449 additions and 26 deletions
16
README.md
16
README.md
|
|
@ -1409,6 +1409,22 @@ Roots define the boundaries of where a server can operate, providing a list of d
|
|||
> automatically and routes the request onto the originating POST stream on the Streamable HTTP transport. Calling the corresponding
|
||||
> `ServerSession` methods without `related_request_id:` still works but emits a deprecation warning.
|
||||
|
||||
**Timeouts:** every server-to-client request is bounded, so a client that never answers cannot park the handler's thread indefinitely.
|
||||
`MCP::Server::Transports::StreamableHTTPTransport` waits `server_to_client_request_timeout:` seconds (600 by default), then tells
|
||||
the client the request was abandoned and raises `MCP::Server::RequestTimeoutError`. Individual calls override the deadline with `timeout:`,
|
||||
which is the knob to reach for when a prompt legitimately waits on a person:
|
||||
|
||||
```ruby
|
||||
server_context.create_form_elicitation(
|
||||
message: "Approve this deployment?",
|
||||
requested_schema: { type: "object", properties: { approved: { type: "boolean" } } },
|
||||
timeout: 3600, # This one waits up to an hour.
|
||||
)
|
||||
```
|
||||
|
||||
`StdioTransport` is not bounded and ignores `timeout:`: it owns the client process, so a client that stops answering
|
||||
surfaces as end-of-file rather than as a wait that never ends.
|
||||
|
||||
**Using Roots in Tools:**
|
||||
|
||||
Tools that accept a `server_context:` parameter can call `list_roots` on it.
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ require_relative "server_context"
|
|||
require_relative "server/capabilities"
|
||||
require_relative "server/input_required_result"
|
||||
require_relative "server/pagination"
|
||||
require_relative "server/pending_response"
|
||||
require_relative "server/request_state_security"
|
||||
require_relative "server/transports"
|
||||
|
||||
|
|
@ -125,6 +126,27 @@ module MCP
|
|||
end
|
||||
end
|
||||
|
||||
# Raised when a server-to-client request (sampling, elicitation, `roots/list`, `ping`) goes unanswered past its timeout.
|
||||
# The spec asks implementations to bound every sent request so a peer that never answers cannot exhaust the sender's resources,
|
||||
# and to cancel the request on expiry; the transport sends `notifications/cancelled` before raising this.
|
||||
# These requests exist only on connections speaking 2025-11-25 or earlier, since the modern lifecycle forbids them.
|
||||
#
|
||||
# Left uncaught in a handler, this answers the client's originating request with `-32001` rather than a generic
|
||||
# internal error, so the peer that failed to answer can tell a timeout apart from a server fault. The code is not
|
||||
# spec-allocated: it sits in the implementation-defined server range and is the value the Python SDK reports for
|
||||
# this condition, so a client that already recognizes it there reads the same meaning here.
|
||||
#
|
||||
# https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#timeouts
|
||||
class RequestTimeoutError < RequestHandlerError
|
||||
attr_reader :request_id, :timeout
|
||||
|
||||
def initialize(message, request_id:, timeout:)
|
||||
super(message, nil, error_type: :request_timeout, error_code: -32001)
|
||||
@request_id = request_id
|
||||
@timeout = timeout
|
||||
end
|
||||
end
|
||||
|
||||
class MethodAlreadyDefinedError < StandardError
|
||||
attr_reader :method_name
|
||||
|
||||
|
|
|
|||
62
lib/mcp/server/pending_response.rb
Normal file
62
lib/mcp/server/pending_response.rb
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
# frozen_string_literal: true
|
||||
|
||||
module MCP
|
||||
class Server
|
||||
# A one-shot, timeout-aware handoff between the thread awaiting a server-to-client response
|
||||
# and whichever thread resolves it (the client's response, a cancellation, or session teardown).
|
||||
#
|
||||
# `Queue#pop` only accepts a `timeout:` on Ruby 3.2 and later, and this gem supports 2.7,
|
||||
# so the wait is expressed with a `ConditionVariable`. The `push`/`pop` names mirror the `Queue`
|
||||
# this replaces, keeping the resolving call sites unchanged.
|
||||
#
|
||||
# First writer wins: a second `push` is ignored, so a cancellation that races a real response
|
||||
# cannot overwrite it. `pop` returns the pushed value, or the `on_timeout` result when
|
||||
# the deadline passes with nothing pushed.
|
||||
class PendingResponse
|
||||
def initialize
|
||||
@mutex = Mutex.new
|
||||
@condition = ConditionVariable.new
|
||||
@delivered = false
|
||||
@value = nil
|
||||
end
|
||||
|
||||
# Resolves the wait. Ignored when a value was already delivered.
|
||||
def push(value)
|
||||
@mutex.synchronize do
|
||||
next if @delivered
|
||||
|
||||
@delivered = true
|
||||
@value = value
|
||||
@condition.broadcast
|
||||
end
|
||||
end
|
||||
|
||||
# Blocks until a value is pushed or `timeout` seconds elapse, and yields to the caller
|
||||
# on expiry so it can decide what a timeout means. `ConditionVariable#wait` can return spuriously,
|
||||
# so the deadline is re-checked against the monotonic clock.
|
||||
#
|
||||
# The expiry block runs after the lock is released: it typically cancels the request,
|
||||
# which resolves this same object, and Ruby's `Mutex` is not reentrant.
|
||||
def pop(timeout:)
|
||||
deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
|
||||
expired = false
|
||||
|
||||
value = @mutex.synchronize do
|
||||
until @delivered
|
||||
remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
|
||||
if remaining <= 0
|
||||
expired = true
|
||||
break
|
||||
end
|
||||
|
||||
@condition.wait(@mutex, remaining)
|
||||
end
|
||||
|
||||
@value
|
||||
end
|
||||
|
||||
expired ? yield : value
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -52,6 +52,21 @@ module MCP
|
|||
UNSET_IDLE_TIMEOUT = Object.new.freeze
|
||||
private_constant :UNSET_IDLE_TIMEOUT
|
||||
|
||||
# Default deadline in seconds for a server-to-client request (sampling, elicitation, `roots/list`, `ping`).
|
||||
# The spec asks implementations to bound every sent request so a peer that never answers cannot exhaust
|
||||
# the sender's resources; without one, a client that opens a session and simply never replies parks
|
||||
# a worker thread for good.
|
||||
#
|
||||
# Ten minutes matches the TypeScript SDK, which raises its uniform 60-second request default to 600 seconds
|
||||
# for the legs of its legacy `input_required` shim because they are "human-paced, so the 60s protocol default
|
||||
# is wrong". Every request this transport can send is that kind of leg: someone answering an elicitation
|
||||
# prompt, or the client's own model producing a sample. (The Python SDK leaves the deadline unset and bounds
|
||||
# nothing by default.) Deployments that want a tighter bound pass a smaller value here; a single handler
|
||||
# that legitimately waits longer passes `timeout:`.
|
||||
#
|
||||
# https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#timeouts
|
||||
DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT = 600
|
||||
|
||||
# Default upper bound on the JSON-RPC request body. `handle_post` reads the whole
|
||||
# body into memory and parses it, so without a cap a single unauthenticated POST
|
||||
# can allocate gigabytes and OOM the worker. 4 MiB comfortably
|
||||
|
|
@ -114,6 +129,9 @@ module MCP
|
|||
# on a `subscriptions/listen` stream; the periodic write frees the stream's slot when the peer
|
||||
# has gone away. Defaults to `DEFAULT_LISTEN_KEEPALIVE_INTERVAL` (15); pass `nil` to disable
|
||||
# when an upstream proxy already keeps the stream alive.
|
||||
# @param server_to_client_request_timeout [Numeric] seconds a server-to-client request waits for its
|
||||
# response before the transport stops waiting and raises `MCP::Server::RequestTimeoutError`.
|
||||
# Defaults to `DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT` (600); individual calls override it with `timeout:`.
|
||||
def initialize(
|
||||
server,
|
||||
stateless: false,
|
||||
|
|
@ -126,7 +144,8 @@ module MCP
|
|||
session_request_validator: nil,
|
||||
max_request_bytes: DEFAULT_MAX_REQUEST_BYTES,
|
||||
max_listen_subscriptions: DEFAULT_MAX_LISTEN_SUBSCRIPTIONS,
|
||||
listen_keepalive_interval: DEFAULT_LISTEN_KEEPALIVE_INTERVAL
|
||||
listen_keepalive_interval: DEFAULT_LISTEN_KEEPALIVE_INTERVAL,
|
||||
server_to_client_request_timeout: DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT
|
||||
)
|
||||
super(server)
|
||||
# Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock, origin: origin_header }`.
|
||||
|
|
@ -193,6 +212,12 @@ module MCP
|
|||
|
||||
@listen_keepalive_interval = listen_keepalive_interval
|
||||
|
||||
unless server_to_client_request_timeout.is_a?(Numeric) && server_to_client_request_timeout.positive?
|
||||
raise ArgumentError, "server_to_client_request_timeout must be a positive number"
|
||||
end
|
||||
|
||||
@server_to_client_request_timeout = server_to_client_request_timeout
|
||||
|
||||
start_reaper_thread if @session_idle_timeout
|
||||
end
|
||||
|
||||
|
|
@ -442,14 +467,18 @@ module MCP
|
|||
end
|
||||
end
|
||||
|
||||
# Sends a server-to-client JSON-RPC request (e.g., `sampling/createMessage`) and
|
||||
# blocks until the client responds.
|
||||
# Sends a server-to-client JSON-RPC request (e.g., `sampling/createMessage`) and blocks until
|
||||
# the client responds.
|
||||
#
|
||||
# Uses a `Queue` for cross-thread synchronization. This method creates a `Queue`,
|
||||
# sends the request via SSE stream, then blocks on `queue.pop`.
|
||||
# When the client POSTs a response, `handle_response` matches it by `request_id`
|
||||
# and pushes the result onto the queue, unblocking this thread.
|
||||
def send_request(method, params = nil, session_id: nil, related_request_id: nil, parent_cancellation: nil, server_session: nil)
|
||||
# Uses a `PendingResponse` for cross-thread synchronization: this method registers one,
|
||||
# sends the request via SSE stream, then waits on it. When the client POSTs a response,
|
||||
# `handle_response` matches it by `request_id` and resolves the pending response,
|
||||
# unblocking this thread. A cancellation and session teardown resolve it the same way.
|
||||
#
|
||||
# The wait is bounded by `timeout` (defaulting to the transport's `server_to_client_request_timeout`),
|
||||
# so a client that never answers cannot park the calling thread for good. On expiry the peer is
|
||||
# sent `notifications/cancelled` and `MCP::Server::RequestTimeoutError` is raised.
|
||||
def send_request(method, params = nil, session_id: nil, related_request_id: nil, parent_cancellation: nil, server_session: nil, timeout: nil)
|
||||
if @stateless
|
||||
raise "Stateless mode does not support server-to-client requests."
|
||||
end
|
||||
|
|
@ -463,7 +492,8 @@ module MCP
|
|||
end
|
||||
|
||||
request_id = generate_request_id
|
||||
queue = Queue.new
|
||||
pending_response = PendingResponse.new
|
||||
wait_timeout = timeout || @server_to_client_request_timeout
|
||||
cancel_hook = nil
|
||||
|
||||
request = { jsonrpc: "2.0", id: request_id, method: method }
|
||||
|
|
@ -476,7 +506,7 @@ module MCP
|
|||
raise "Session not found: #{session_id}."
|
||||
end
|
||||
|
||||
@pending_responses[request_id] = { queue: queue, session_id: session_id }
|
||||
@pending_responses[request_id] = { queue: pending_response, session_id: session_id }
|
||||
|
||||
active_stream(session, related_request_id: related_request_id)
|
||||
end
|
||||
|
|
@ -512,7 +542,27 @@ module MCP
|
|||
end
|
||||
end
|
||||
|
||||
response = queue.pop
|
||||
response = pending_response.pop(timeout: wait_timeout) do
|
||||
# Expiry cancels as well as stops waiting, so a client that answers late does not act on
|
||||
# a request the server has abandoned. Only connections speaking 2025-11-25 or earlier get here:
|
||||
# the modern lifecycle forbids server-to-client requests outright, and its sessionless requests
|
||||
# never register the session this method looks up. Those revisions ask the sender to "issue
|
||||
# a cancellation notification for that request and stop waiting", letting either side send one.
|
||||
# (The 2026-07-28 rule reserving `notifications/cancelled` for `subscriptions/listen` teardown
|
||||
# governs the era that has no such requests to cancel.) Both reference SDKs send this same
|
||||
# courtesy cancel on timeout.
|
||||
server_session&.send_peer_cancellation(
|
||||
nested_request_id: request_id,
|
||||
related_request_id: related_request_id,
|
||||
reason: "Timed out after #{wait_timeout} seconds",
|
||||
)
|
||||
|
||||
raise RequestTimeoutError.new(
|
||||
"#{method} request timed out after #{wait_timeout} seconds",
|
||||
request_id: request_id,
|
||||
timeout: wait_timeout,
|
||||
)
|
||||
end
|
||||
|
||||
if response.is_a?(Hash) && response.key?(:error)
|
||||
raise StandardError, "Client returned an error for #{method} request (code: #{response[:error][:code]}): #{response[:error][:message]}"
|
||||
|
|
|
|||
|
|
@ -138,9 +138,9 @@ module MCP
|
|||
# `notifications/roots/list_changed`) is deprecated as of MCP protocol
|
||||
# version 2026-07-28 (SEP-2577). Use tool parameters, resource URIs,
|
||||
# server configuration, or environment variables instead.
|
||||
def list_roots
|
||||
def list_roots(timeout: nil)
|
||||
if @notification_target.respond_to?(:list_roots)
|
||||
@notification_target.list_roots(related_request_id: @related_request_id)
|
||||
@notification_target.list_roots(related_request_id: @related_request_id, **timeout_kwarg(timeout))
|
||||
else
|
||||
raise NoMethodError, "undefined method 'list_roots' for #{self}"
|
||||
end
|
||||
|
|
@ -160,9 +160,9 @@ module MCP
|
|||
# end
|
||||
#
|
||||
# @see https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/ping
|
||||
def ping
|
||||
def ping(timeout: nil)
|
||||
if @notification_target.respond_to?(:ping)
|
||||
@notification_target.ping(related_request_id: @related_request_id)
|
||||
@notification_target.ping(related_request_id: @related_request_id, **timeout_kwarg(timeout))
|
||||
else
|
||||
raise NoMethodError, "undefined method 'ping' for #{self}"
|
||||
end
|
||||
|
|
@ -245,5 +245,15 @@ module MCP
|
|||
def respond_to_missing?(name, include_private = false)
|
||||
@context.respond_to?(name) || super
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
# An omitted `timeout:` is not forwarded at all, so the delegated call keeps the shape it had
|
||||
# before per-request timeouts existed. A notification target that predates the keyword
|
||||
# (a custom object standing in for a session) keeps working until a caller actually asks for a timeout,
|
||||
# and the transport applies its own default in that case.
|
||||
def timeout_kwarg(timeout)
|
||||
timeout.nil? ? {} : { timeout: timeout }
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -131,7 +131,7 @@ module MCP
|
|||
# `notifications/roots/list_changed`) is deprecated as of MCP protocol
|
||||
# version 2026-07-28 (SEP-2577). Use tool parameters, resource URIs,
|
||||
# server configuration, or environment variables instead.
|
||||
def list_roots(related_request_id: nil)
|
||||
def list_roots(related_request_id: nil, timeout: nil)
|
||||
@server.send(:warn_if_deprecated_protocol_feature, :roots, session: self, uplevel: 2)
|
||||
warn_unassociated_request(__method__, related_request_id)
|
||||
|
||||
|
|
@ -139,12 +139,12 @@ module MCP
|
|||
raise "Client does not support roots."
|
||||
end
|
||||
|
||||
send_to_transport_request(Methods::ROOTS_LIST, nil, related_request_id: related_request_id)
|
||||
send_to_transport_request(Methods::ROOTS_LIST, nil, related_request_id: related_request_id, timeout: timeout)
|
||||
end
|
||||
|
||||
# Sends a `ping` request scoped to this session.
|
||||
def ping(related_request_id: nil)
|
||||
result = send_to_transport_request(Methods::PING, nil, related_request_id: related_request_id)
|
||||
def ping(related_request_id: nil, timeout: nil)
|
||||
result = send_to_transport_request(Methods::PING, nil, related_request_id: related_request_id, timeout: timeout)
|
||||
raise Server::ValidationError, "Response validation failed: invalid `result`" unless result.is_a?(Hash)
|
||||
|
||||
result
|
||||
|
|
@ -158,12 +158,17 @@ module MCP
|
|||
# @deprecated MCP Sampling (`sampling/createMessage`) is deprecated as of
|
||||
# MCP protocol version 2026-07-28 (SEP-2577). Use direct LLM provider
|
||||
# APIs instead.
|
||||
def create_sampling_message(related_request_id: nil, **kwargs)
|
||||
def create_sampling_message(related_request_id: nil, timeout: nil, **kwargs)
|
||||
@server.send(:warn_if_deprecated_protocol_feature, :sampling, session: self, uplevel: 2)
|
||||
warn_unassociated_request(__method__, related_request_id)
|
||||
|
||||
params = @server.build_sampling_params(client_capabilities, **kwargs)
|
||||
send_to_transport_request(Methods::SAMPLING_CREATE_MESSAGE, params, related_request_id: related_request_id)
|
||||
send_to_transport_request(
|
||||
Methods::SAMPLING_CREATE_MESSAGE,
|
||||
params,
|
||||
related_request_id: related_request_id,
|
||||
timeout: timeout,
|
||||
)
|
||||
end
|
||||
|
||||
# Sends an `elicitation/create` request (form mode) scoped to this session.
|
||||
|
|
@ -171,7 +176,7 @@ module MCP
|
|||
# Per SEP-2260, the request must be associated with an originating client
|
||||
# request; prefer `server_context.create_form_elicitation` inside a handler,
|
||||
# which stamps the association automatically.
|
||||
def create_form_elicitation(message:, requested_schema:, related_request_id: nil)
|
||||
def create_form_elicitation(message:, requested_schema:, related_request_id: nil, timeout: nil)
|
||||
warn_unassociated_request(__method__, related_request_id)
|
||||
|
||||
unless client_capabilities&.dig(:elicitation)
|
||||
|
|
@ -180,7 +185,12 @@ module MCP
|
|||
end
|
||||
|
||||
params = { mode: "form", message: message, requestedSchema: requested_schema }
|
||||
send_to_transport_request(Methods::ELICITATION_CREATE, params, related_request_id: related_request_id)
|
||||
send_to_transport_request(
|
||||
Methods::ELICITATION_CREATE,
|
||||
params,
|
||||
related_request_id: related_request_id,
|
||||
timeout: timeout,
|
||||
)
|
||||
end
|
||||
|
||||
# Sends an `elicitation/create` request (URL mode) scoped to this session.
|
||||
|
|
@ -188,7 +198,7 @@ module MCP
|
|||
# Per SEP-2260, the request must be associated with an originating client
|
||||
# request; prefer `server_context.create_url_elicitation` inside a handler,
|
||||
# which stamps the association automatically.
|
||||
def create_url_elicitation(message:, url:, elicitation_id:, related_request_id: nil)
|
||||
def create_url_elicitation(message:, url:, elicitation_id:, related_request_id: nil, timeout: nil)
|
||||
warn_unassociated_request(__method__, related_request_id)
|
||||
|
||||
unless client_capabilities&.dig(:elicitation, :url)
|
||||
|
|
@ -197,7 +207,12 @@ module MCP
|
|||
end
|
||||
|
||||
params = { mode: "url", message: message, url: url, elicitationId: elicitation_id }
|
||||
send_to_transport_request(Methods::ELICITATION_CREATE, params, related_request_id: related_request_id)
|
||||
send_to_transport_request(
|
||||
Methods::ELICITATION_CREATE,
|
||||
params,
|
||||
related_request_id: related_request_id,
|
||||
timeout: timeout,
|
||||
)
|
||||
end
|
||||
|
||||
# Sends `notifications/cancelled` to the peer for a nested server-to-client request
|
||||
|
|
@ -296,7 +311,7 @@ module MCP
|
|||
# `parent_cancellation:` / `server_session:` receive the nested-cancellation plumbing.
|
||||
# When `related_request_id` names an in-flight request, its `Cancellation` token is looked up
|
||||
# so that cancelling the parent also cancels this nested server-to-client request.
|
||||
def send_to_transport_request(method, params, related_request_id: nil)
|
||||
def send_to_transport_request(method, params, related_request_id: nil, timeout: nil)
|
||||
parent_cancellation = related_request_id ? lookup_in_flight(related_request_id) : nil
|
||||
|
||||
kwargs = {
|
||||
|
|
@ -304,6 +319,7 @@ module MCP
|
|||
related_request_id: related_request_id,
|
||||
parent_cancellation: parent_cancellation,
|
||||
server_session: self,
|
||||
timeout: timeout,
|
||||
}.compact
|
||||
|
||||
forward_to_transport(@transport.method(:send_request), method, params, kwargs)
|
||||
|
|
|
|||
|
|
@ -3041,6 +3041,89 @@ module MCP
|
|||
assert_equal "Hi there", result[:content][:text]
|
||||
end
|
||||
|
||||
test "send_request times out when the client never answers" do
|
||||
# The spec asks implementations to bound every sent request so an unanswered one cannot
|
||||
# exhaust the sender's resources.
|
||||
transport = StreamableHTTPTransport.new(@server, server_to_client_request_timeout: 0.05)
|
||||
session_id = open_sse_session(transport)
|
||||
|
||||
error = assert_raises(MCP::Server::RequestTimeoutError) do
|
||||
transport.send_request("sampling/createMessage", { messages: [] }, session_id: session_id)
|
||||
end
|
||||
|
||||
assert_equal 0.05, error.timeout
|
||||
assert_includes error.message, "sampling/createMessage"
|
||||
|
||||
# Left uncaught in a handler, the originating request is answered with a timeout code rather than
|
||||
# a generic internal error, so the peer that failed to answer can tell the two apart.
|
||||
assert_equal(-32001, error.error_code)
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "send_request timeout is overridable per request" do
|
||||
transport = StreamableHTTPTransport.new(@server, server_to_client_request_timeout: 60)
|
||||
session_id = open_sse_session(transport)
|
||||
|
||||
error = assert_raises(MCP::Server::RequestTimeoutError) do
|
||||
transport.send_request("ping", nil, session_id: session_id, timeout: 0.05)
|
||||
end
|
||||
|
||||
assert_equal 0.05, error.timeout
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "send_request tells the peer it gave up when the timeout fires" do
|
||||
# These requests only happen on 2025-11-25 and earlier connections, whose spec asks the sender
|
||||
# to issue a cancellation notification for the request it stops waiting on.
|
||||
transport = StreamableHTTPTransport.new(@server, server_to_client_request_timeout: 0.05)
|
||||
io = StringIO.new
|
||||
session_id = open_sse_session(transport, io: io)
|
||||
session = transport.instance_variable_get(:@sessions)[session_id][:server_session]
|
||||
|
||||
assert_raises(MCP::Server::RequestTimeoutError) do
|
||||
transport.send_request("ping", nil, session_id: session_id, server_session: session)
|
||||
end
|
||||
|
||||
events = io.string.lines.grep(/\Adata: /).map { |line| JSON.parse(line.sub("data: ", "")) }
|
||||
cancellation = events.find { |event| event["method"] == "notifications/cancelled" }
|
||||
refute_nil cancellation, "expected the peer to be told the request was abandoned"
|
||||
assert_match(/Timed out after 0.05 seconds/, cancellation.dig("params", "reason"))
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "send_request returns the response when it arrives before the timeout" do
|
||||
transport = StreamableHTTPTransport.new(@server, server_to_client_request_timeout: 30)
|
||||
io = StringIO.new
|
||||
session_id = open_sse_session(transport, io: io)
|
||||
|
||||
result_queue = Queue.new
|
||||
Thread.new do
|
||||
result_queue.push(transport.send_request("ping", nil, session_id: session_id))
|
||||
end
|
||||
wait_until { io.string.include?("ping") }
|
||||
|
||||
request_id = JSON.parse(io.string.lines.grep(/\Adata: /).first.sub("data: ", ""))["id"]
|
||||
transport.handle_request(create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
|
||||
{ jsonrpc: "2.0", id: request_id, result: { ok: true } }.to_json,
|
||||
))
|
||||
|
||||
assert_equal({ ok: true }, result_queue.pop)
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "server_to_client_request_timeout rejects a non-positive value" do
|
||||
assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, server_to_client_request_timeout: 0) }
|
||||
assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, server_to_client_request_timeout: -1) }
|
||||
assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, server_to_client_request_timeout: nil) }
|
||||
end
|
||||
|
||||
test "send_request with related_request_id rides the originating POST stream, not the GET stream" do
|
||||
# Per SEP-2260, server-to-client requests associated with a client request are delivered on
|
||||
# that request's POST SSE response stream rather than the standalone GET stream.
|
||||
|
|
@ -6028,6 +6111,36 @@ module MCP
|
|||
@transport.handle_request(init_request)[1]["mcp-session-id"]
|
||||
end
|
||||
|
||||
# Polls until the block is truthy; the work being awaited happens on another thread.
|
||||
def wait_until(timeout: 5)
|
||||
deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
|
||||
until yield
|
||||
flunk("Timed out waiting for condition") if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
|
||||
sleep(0.01)
|
||||
end
|
||||
end
|
||||
|
||||
# Initializes a session on `transport` and attaches a GET SSE stream to it, so
|
||||
# server-to-client requests have somewhere to go. Returns the session id.
|
||||
def open_sse_session(transport, io: StringIO.new)
|
||||
session_id = transport.handle_request(create_rack_request(
|
||||
"POST",
|
||||
"/",
|
||||
{ "CONTENT_TYPE" => "application/json" },
|
||||
{ jsonrpc: "2.0", method: "initialize", id: "init", params: initialize_params }.to_json,
|
||||
))[1]["mcp-session-id"]
|
||||
|
||||
get_response = transport.handle_request(create_rack_request(
|
||||
"GET",
|
||||
"/",
|
||||
{ "HTTP_MCP_SESSION_ID" => session_id },
|
||||
))
|
||||
get_response[2].call(io) if get_response[2].is_a?(Proc)
|
||||
wait_until { transport.instance_variable_get(:@sessions).dig(session_id, :get_sse_stream) }
|
||||
|
||||
session_id
|
||||
end
|
||||
|
||||
# Installs a stream that records, on every write, whether `@mutex` was held at that moment.
|
||||
# `mutex.try_lock` fails while the transport still holds the lock, so a nonzero `writes[:under_mutex]`
|
||||
# means a blocking write is being performed under the lock.
|
||||
|
|
|
|||
|
|
@ -162,6 +162,41 @@ module MCP
|
|||
assert_equal({}, result)
|
||||
end
|
||||
|
||||
test "ServerContext forwards a per-request timeout to the notification target" do
|
||||
notification_target = mock
|
||||
notification_target.expects(:ping).with(related_request_id: nil, timeout: 12).returns({})
|
||||
notification_target.expects(:list_roots).with(related_request_id: nil, timeout: 34).returns({ roots: [] })
|
||||
notification_target.expects(:create_form_elicitation).with(
|
||||
message: "Approve?",
|
||||
requested_schema: { type: "object", properties: {} },
|
||||
related_request_id: nil,
|
||||
timeout: 56,
|
||||
).returns({ action: "accept" })
|
||||
|
||||
progress = Progress.new(notification_target: notification_target, progress_token: nil)
|
||||
server_context = ServerContext.new(mock, progress: progress, notification_target: notification_target)
|
||||
|
||||
server_context.ping(timeout: 12)
|
||||
server_context.list_roots(timeout: 34)
|
||||
server_context.create_form_elicitation(
|
||||
message: "Approve?",
|
||||
requested_schema: { type: "object", properties: {} },
|
||||
timeout: 56,
|
||||
)
|
||||
end
|
||||
|
||||
test "ServerContext omits timeout entirely when the caller does not ask for one" do
|
||||
# A notification target predating the keyword (a custom object standing in for a session) keeps working,
|
||||
# and the transport applies its own default.
|
||||
notification_target = mock
|
||||
notification_target.expects(:ping).with(related_request_id: nil).returns({})
|
||||
|
||||
progress = Progress.new(notification_target: notification_target, progress_token: nil)
|
||||
server_context = ServerContext.new(mock, progress: progress, notification_target: notification_target)
|
||||
|
||||
server_context.ping
|
||||
end
|
||||
|
||||
test "ServerContext#ping raises NoMethodError when notification_target does not respond" do
|
||||
notification_target = mock
|
||||
context = mock
|
||||
|
|
|
|||
99
test/mcp/server_session_request_timeout_test.rb
Normal file
99
test/mcp/server_session_request_timeout_test.rb
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
# frozen_string_literal: true
|
||||
|
||||
require "test_helper"
|
||||
|
||||
module MCP
|
||||
# Every server-to-client request accepts a per-request deadline, which the spec asks SDKs to offer
|
||||
# ("SDKs and other middleware SHOULD allow these timeouts to be configured on a per-request basis").
|
||||
# `ServerSession` only carries the value; the transport is what waits on it.
|
||||
class ServerSessionRequestTimeoutTest < ActiveSupport::TestCase
|
||||
# Declares `timeout:`, as the bundled Streamable HTTP transport does.
|
||||
class TimeoutAwareTransport < Transport
|
||||
attr_reader :requests
|
||||
|
||||
def initialize(server)
|
||||
super
|
||||
@requests = []
|
||||
end
|
||||
|
||||
def send_request(method, params = nil, session_id: nil, related_request_id: nil, parent_cancellation: nil, server_session: nil, timeout: nil)
|
||||
@requests << { method: method, timeout: timeout }
|
||||
{}
|
||||
end
|
||||
|
||||
def send_response(response); end
|
||||
def send_notification(method, params = nil, **_kwargs); end
|
||||
def open; end
|
||||
def close; end
|
||||
def handle_request(request); end
|
||||
end
|
||||
|
||||
# Implements only the abstract `send_request(method, params = nil)` contract, as `StdioTransport` does.
|
||||
class TimeoutUnawareTransport < TimeoutAwareTransport
|
||||
def send_request(method, params = nil)
|
||||
@requests << { method: method, timeout: :not_passed }
|
||||
{}
|
||||
end
|
||||
end
|
||||
|
||||
setup do
|
||||
@server = Server.new(name: "test_server", version: "1.0.0")
|
||||
@transport = TimeoutAwareTransport.new(@server)
|
||||
@session = ServerSession.new(server: @server, transport: @transport, session_id: "session-1")
|
||||
@session.instance_variable_set(
|
||||
:@client_capabilities,
|
||||
{ roots: {}, sampling: {}, elicitation: { url: {} } },
|
||||
)
|
||||
end
|
||||
|
||||
test "every server-to-client request carries its timeout down to the transport" do
|
||||
@session.ping(related_request_id: "req-1", timeout: 11)
|
||||
@session.list_roots(related_request_id: "req-1", timeout: 22)
|
||||
@session.create_sampling_message(
|
||||
related_request_id: "req-1",
|
||||
timeout: 33,
|
||||
messages: [{ role: "user", content: { type: "text", text: "Hello" } }],
|
||||
max_tokens: 100,
|
||||
)
|
||||
@session.create_form_elicitation(
|
||||
message: "Approve?",
|
||||
requested_schema: { type: "object", properties: {} },
|
||||
related_request_id: "req-1",
|
||||
timeout: 44,
|
||||
)
|
||||
@session.create_url_elicitation(
|
||||
message: "Approve?",
|
||||
url: "https://example.com/approve",
|
||||
elicitation_id: "elicit-1",
|
||||
related_request_id: "req-1",
|
||||
timeout: 55,
|
||||
)
|
||||
|
||||
assert_equal(
|
||||
[
|
||||
{ method: "ping", timeout: 11 },
|
||||
{ method: "roots/list", timeout: 22 },
|
||||
{ method: "sampling/createMessage", timeout: 33 },
|
||||
{ method: "elicitation/create", timeout: 44 },
|
||||
{ method: "elicitation/create", timeout: 55 },
|
||||
],
|
||||
@transport.requests,
|
||||
)
|
||||
end
|
||||
|
||||
test "an omitted timeout leaves the transport to apply its own default" do
|
||||
@session.ping(related_request_id: "req-1")
|
||||
|
||||
assert_equal [{ method: "ping", timeout: nil }], @transport.requests
|
||||
end
|
||||
|
||||
test "a transport that does not declare timeout keeps its existing signature" do
|
||||
transport = TimeoutUnawareTransport.new(@server)
|
||||
session = ServerSession.new(server: @server, transport: transport, session_id: "session-1")
|
||||
|
||||
session.ping(related_request_id: "req-1", timeout: 11)
|
||||
|
||||
assert_equal [{ method: "ping", timeout: :not_passed }], transport.requests
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Reference in a new issue