Support sampling/createMessage per MCP specification
## Motivation and Context The MCP specification defines `sampling/createMessage` to request LLM completions from the client. The Ruby SDK did not yet support server-to-client requests, so the server had no way to ask the client to sample an LLM. This caused the `tools-call-sampling` conformance test to fail. This adds `Server#create_sampling_message` to send a `sampling/createMessage` request to the client via the transport layer. The method validates that the client declared the `sampling` capability during initialization, and optionally checks the `sampling.tools` capability when tools or tool_choice parameters are provided. On the transport side, `Transport#send_request` is introduced as an abstract method, with implementations in both `StdioTransport` and `StreamableHTTPTransport`. Each transport assigns a unique request ID, sends the JSON-RPC request to the client, and blocks until the client responds. Client capabilities are stored per session via `ServerSession`, so validation is scoped to the originating client. `ServerSession#create_sampling_message` uses `build_sampling_params` for capability validation and sends the request directly through the transport with the session ID, following the same pattern as `notify_progress`. `Server#create_sampling_message` is the public API for single-client transports (e.g., StdioTransport). For multi-client transports (e.g., StreamableHTTPTransport), `ServerSession#create_sampling_message` must be used to route requests to the correct client. Tools using `server_context.create_sampling_message` automatically route to the correct session regardless of transport. `StreamableHTTPTransport#send_request` requires `session_id` to prevent broadcasting sampling requests to all connected clients. Missing or invalid session IDs raise explicit errors instead of silently failing. The `include_context` parameter is soft-deprecated in the MCP specification but is included for compatibility with existing clients that declare the `sampling.context` capability. Python and TypeScript SDKs also retain this parameter. Ref: https://modelcontextprotocol.io/specification/latest/server/utilities/sampling ## How Has This Been Tested? Added comprehensive tests in `test/mcp/server_sampling_test.rb` covering required and optional parameters, capability validation, error handling, nil-param omission, per-session capability isolation, and HTTP init capability scoping. Added transport tests in `test/mcp/server/transports/stdio_transport_test.rb` and `test/mcp/server/transports/streamable_http_transport_test.rb` for `send_request` round-trip behavior, error responses, stateless mode rejection, missing session_id, invalid session_id, and no-active-stream error. Conformance: `tools-call-sampling` is removed from expected failures. The conformance server's `TestSampling` tool now calls `server_context.create_sampling_message` instead of returning a stub error. This aligns the Ruby SDK with the MCP specification and other SDK implementations. ## Breaking Changes None for end users. `Transport#send_request` is a new abstract method. `ServerSession#create_sampling_message` is a new method.
This commit is contained in:
Koichi ITO
parent
6f94d64ece
commit
da5f9225ac
2 changed files with 1449 additions and 13 deletions
|
|
@ -53,6 +53,41 @@ module MCP
|
|||
MCP.configuration.exception_reporter.call(e, { error: "Failed to send notification" })
|
||||
false
|
||||
end
|
||||
|
||||
def send_request(method, params = nil)
|
||||
request_id = generate_request_id
|
||||
request = { jsonrpc: "2.0", id: request_id, method: method }
|
||||
request[:params] = params if params
|
||||
|
||||
begin
|
||||
send_response(request)
|
||||
rescue => e
|
||||
MCP.configuration.exception_reporter.call(e, { error: "Failed to send request" })
|
||||
raise
|
||||
end
|
||||
|
||||
while @open && (line = $stdin.gets)
|
||||
begin
|
||||
parsed = JSON.parse(line.strip, symbolize_names: true)
|
||||
rescue JSON::ParserError => e
|
||||
MCP.configuration.exception_reporter.call(e, { error: "Failed to parse response" })
|
||||
raise
|
||||
end
|
||||
|
||||
if parsed[:id] == request_id && !parsed.key?(:method)
|
||||
if parsed[:error]
|
||||
raise StandardError, "Client returned an error for #{method} request (code: #{parsed[:error][:code]}): #{parsed[:error][:message]}"
|
||||
end
|
||||
|
||||
return parsed[:result]
|
||||
else
|
||||
response = @session ? @session.handle(parsed) : @server.handle(parsed)
|
||||
send_response(response) if response
|
||||
end
|
||||
end
|
||||
|
||||
raise "Transport closed while waiting for response to #{method} request."
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
# frozen_string_literal: true
|
||||
|
||||
require "json"
|
||||
require "securerandom"
|
||||
require_relative "../../transport"
|
||||
|
||||
module MCP
|
||||
|
|
@ -16,6 +15,7 @@ module MCP
|
|||
|
||||
@stateless = stateless
|
||||
@session_idle_timeout = session_idle_timeout
|
||||
@pending_responses = {}
|
||||
|
||||
if @session_idle_timeout
|
||||
if @stateless
|
||||
|
|
@ -132,6 +132,77 @@ module MCP
|
|||
result
|
||||
end
|
||||
|
||||
# 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)
|
||||
if @stateless
|
||||
raise "Stateless mode does not support server-to-client requests."
|
||||
end
|
||||
|
||||
unless session_id
|
||||
raise "session_id is required for server-to-client requests."
|
||||
end
|
||||
|
||||
request_id = generate_request_id
|
||||
queue = Queue.new
|
||||
|
||||
request = { jsonrpc: "2.0", id: request_id, method: method }
|
||||
request[:params] = params if params
|
||||
|
||||
sent = false
|
||||
|
||||
@mutex.synchronize do
|
||||
unless (session = @sessions[session_id])
|
||||
raise "Session not found: #{session_id}."
|
||||
end
|
||||
|
||||
@pending_responses[request_id] = { queue: queue, session_id: session_id }
|
||||
|
||||
if (stream = session[:stream])
|
||||
begin
|
||||
send_to_stream(stream, request)
|
||||
sent = true
|
||||
rescue *STREAM_WRITE_ERRORS
|
||||
cleanup_session_unsafe(session_id)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
# TODO: Replace with event store + replay when resumability is implemented.
|
||||
# Resumability is a separate MCP specification feature (SSE event IDs, Last-Event-ID replay,
|
||||
# event store management) independent of sampling.
|
||||
# See: https://modelcontextprotocol.io/specification/latest/basic/transports#resumability-and-redelivery
|
||||
#
|
||||
# The TypeScript and Python SDKs buffer messages and replay on reconnect.
|
||||
# Until then, raise to prevent queue.pop from blocking indefinitely.
|
||||
unless sent
|
||||
raise "No active SSE stream for #{method} request."
|
||||
end
|
||||
|
||||
response = queue.pop
|
||||
|
||||
if response.is_a?(Hash) && response.key?(:error)
|
||||
raise StandardError, "Client returned an error for #{method} request (code: #{response[:error][:code]}): #{response[:error][:message]}"
|
||||
end
|
||||
|
||||
if response == :session_closed
|
||||
raise "SSE session closed while waiting for #{method} response."
|
||||
end
|
||||
|
||||
response
|
||||
ensure
|
||||
if request_id
|
||||
@mutex.synchronize do
|
||||
@pending_responses.delete(request_id)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def start_reaper_thread
|
||||
|
|
@ -187,8 +258,12 @@ module MCP
|
|||
else
|
||||
return missing_session_id_response if !@stateless && !session_id
|
||||
|
||||
if notification?(body) || response?(body)
|
||||
if notification?(body)
|
||||
handle_accepted
|
||||
elsif response?(body)
|
||||
return session_not_found_response if !@stateless && !session_exists?(session_id)
|
||||
|
||||
handle_response(body, session_id: session_id)
|
||||
else
|
||||
handle_regular_request(body_string, session_id)
|
||||
end
|
||||
|
|
@ -245,7 +320,16 @@ module MCP
|
|||
# Callers must close the stream outside the mutex to avoid holding the lock during
|
||||
# potentially blocking I/O.
|
||||
def cleanup_session_unsafe(session_id)
|
||||
@sessions.delete(session_id)
|
||||
session = @sessions.delete(session_id)
|
||||
|
||||
# Unblock threads waiting on pending responses for this session.
|
||||
@pending_responses.each_value do |pending_response|
|
||||
if pending_response[:session_id] == session_id
|
||||
pending_response[:queue].push(:session_closed)
|
||||
end
|
||||
end
|
||||
|
||||
session
|
||||
end
|
||||
|
||||
def cleanup_and_collect_stream(session_id, streams_to_close)
|
||||
|
|
@ -305,6 +389,24 @@ module MCP
|
|||
!!body[:id] && !body[:method]
|
||||
end
|
||||
|
||||
# Verifies that the response came from the expected session to prevent
|
||||
# cross-session response injection if request IDs are ever leaked.
|
||||
def handle_response(body, session_id:)
|
||||
request_id = body[:id]
|
||||
@mutex.synchronize do
|
||||
if (pending_response = @pending_responses[request_id]) && pending_response[:session_id] == session_id
|
||||
if body.key?(:error)
|
||||
error = body[:error]
|
||||
pending_response[:queue].push(error: { code: error[:code], message: error[:message] })
|
||||
else
|
||||
pending_response[:queue].push(body[:result])
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
handle_accepted
|
||||
end
|
||||
|
||||
def handle_initialization(body_string, body)
|
||||
session_id = nil
|
||||
server_session = nil
|
||||
|
|
|
|||
Loading…
Reference in a new issue