Merge pull request #490 from koic/stream_modern_request_notifications
Stream modern request notifications and honor the envelope `logLevel` per SEP-2575
This commit is contained in:
Koichi ITO
GitHub
commit
02aed75882
6 changed files with 286 additions and 8 deletions
|
|
@ -344,6 +344,21 @@ module Conformance
|
|||
end
|
||||
end
|
||||
|
||||
class TestStreamingElicitation < MCP::Tool
|
||||
tool_name "test_streaming_elicitation"
|
||||
description "A tool whose response stream carries only notifications, never independent requests (SEP-2575)"
|
||||
|
||||
class << self
|
||||
def call(server_context:, **_args)
|
||||
# No independent server-to-client request is ever written to the response stream;
|
||||
# the progress notification only goes out when the request carried a progressToken.
|
||||
server_context.report_progress(50, total: 100)
|
||||
|
||||
MCP::Tool::Response.new([MCP::Content::Text.new("Streaming complete").to_h])
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
class TestInputRequiredResultElicitation < MCP::Tool
|
||||
tool_name "test_input_required_result_elicitation"
|
||||
description "A tool that asks for the user's name through an input_required result (SEP-2322)"
|
||||
|
|
@ -609,6 +624,21 @@ module Conformance
|
|||
end
|
||||
end
|
||||
end
|
||||
|
||||
class TestLoggingTool < MCP::Tool
|
||||
tool_name "test_logging_tool"
|
||||
description "A tool that logs only for requests opting in via the _meta logLevel (SEP-2575)"
|
||||
|
||||
class << self
|
||||
def call(server_context:, **_args)
|
||||
# On modern requests the notification is dropped unless `io.modelcontextprotocol/logLevel` authorized it,
|
||||
# so the call itself is unconditional.
|
||||
server_context.notify_log_message(data: "Log level honored", level: "info", logger: "conformance")
|
||||
|
||||
MCP::Tool::Response.new([MCP::Content::Text.new("Logging evaluated").to_h])
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
module Prompts
|
||||
|
|
@ -846,6 +876,8 @@ module Conformance
|
|||
Tools::TestInputRequiredResultMultipleInputs,
|
||||
Tools::TestInputRequiredResultMultiRound,
|
||||
Tools::TestInputRequiredResultCapabilities,
|
||||
Tools::TestStreamingElicitation,
|
||||
Tools::TestLoggingTool,
|
||||
],
|
||||
prompts: [
|
||||
Prompts::TestSimplePrompt,
|
||||
|
|
|
|||
|
|
@ -584,6 +584,16 @@ module MCP
|
|||
exception_already_reported: ->(e) { reported_exception.equal?(e) },
|
||||
) do
|
||||
envelope = lift_request_envelope(params, method: method, session: session)
|
||||
|
||||
# The envelope's `logLevel` member replaces `logging/setLevel` in the modern lifecycle:
|
||||
# it authorizes `notifications/message` for this request only (SEP-2575). Without it,
|
||||
# `ServerSession#notify_log_message` stays silent on modern-era sessions. An unrecognized level
|
||||
# reads as absent, so delivery stays off (the safe direction), matching the Python SDK.
|
||||
if envelope&.log_level && session.respond_to?(:configure_logging)
|
||||
request_logging = LoggingMessageNotification.new(level: envelope.log_level)
|
||||
session.configure_logging(request_logging) if request_logging.valid_level?
|
||||
end
|
||||
|
||||
result = case method
|
||||
when Methods::INITIALIZE
|
||||
init(params, session: session)
|
||||
|
|
|
|||
|
|
@ -51,6 +51,13 @@ module MCP
|
|||
# the stack or amplify parse cost (complements the byte cap).
|
||||
MAX_JSON_NESTING = 64
|
||||
|
||||
# Cap on the notifications buffered for one modern request (SEP-2575). The sink holds them
|
||||
# in memory until the handler returns, so a handler that emits notifications in proportion to
|
||||
# client-supplied input would otherwise let one request grow memory without bound.
|
||||
# Notifications past the cap are not delivered and the notify helpers report `false`,
|
||||
# the same non-delivery degradation they have on every other undeliverable path.
|
||||
MAX_MODERN_REQUEST_NOTIFICATIONS = 1_000
|
||||
|
||||
# Creates a Streamable HTTP transport that can be mounted as a Rack app.
|
||||
#
|
||||
# @param server [MCP::Server] the server whose requests this transport dispatches.
|
||||
|
|
@ -107,6 +114,10 @@ module MCP
|
|||
@allowed_origins = Array(allowed_origins).map(&:downcase).freeze
|
||||
@pending_responses = {}
|
||||
|
||||
# Maps a modern request's ephemeral session id to the Array collecting the notifications its handler emits;
|
||||
# `handle_modern` registers the sink and flushes it as SSE frames ahead of the final response (SEP-2575).
|
||||
@modern_request_sinks = {}
|
||||
|
||||
# Resolve the idle timeout: an explicit value (including `nil` to opt out) wins; otherwise apply the secure default,
|
||||
# which does not apply to stateless mode since it retains no sessions.
|
||||
@session_idle_timeout = if session_idle_timeout.equal?(UNSET_IDLE_TIMEOUT)
|
||||
|
|
@ -235,17 +246,30 @@ module MCP
|
|||
end
|
||||
|
||||
def send_notification(method, params = nil, session_id: nil, related_request_id: nil)
|
||||
# Stateless mode has no streams to deliver notifications on. Report non-delivery instead of raising
|
||||
# so the ephemeral per-request session's notify_* helpers (e.g. progress or log notifications from
|
||||
# a tool handler) degrade gracefully rather than spamming the exception reporter on every call.
|
||||
return false if @stateless
|
||||
|
||||
notification = {
|
||||
jsonrpc: "2.0",
|
||||
method: method,
|
||||
}
|
||||
notification[:params] = params if params
|
||||
|
||||
# A modern request's notifications ride its own response stream (SEP-2575): `handle_modern` registers
|
||||
# a per-request sink for its ephemeral session and flushes it as SSE frames ahead of the final response.
|
||||
# Checked before the stateless guard, since modern requests are served in stateless deployments too.
|
||||
# The sink is bounded; past `MAX_MODERN_REQUEST_NOTIFICATIONS` the notification is dropped as
|
||||
# non-delivery (`false`), never handed to the legacy paths below.
|
||||
sink = @mutex.synchronize { session_id && @modern_request_sinks[session_id] }
|
||||
if sink
|
||||
return false if sink.size >= MAX_MODERN_REQUEST_NOTIFICATIONS
|
||||
|
||||
sink << notification
|
||||
return true
|
||||
end
|
||||
|
||||
# Stateless mode has no streams to deliver notifications on. Report non-delivery instead of raising
|
||||
# so the ephemeral per-request session's notify_* helpers (e.g. progress or log notifications from
|
||||
# a tool handler) degrade gracefully rather than spamming the exception reporter on every call.
|
||||
return false if @stateless
|
||||
|
||||
if session_id
|
||||
deliver_targeted_notification(notification, session_id, related_request_id)
|
||||
else
|
||||
|
|
@ -564,12 +588,29 @@ module MCP
|
|||
mismatch_error = validate_modern_headers(request, body, header_version)
|
||||
return mismatch_error if mismatch_error
|
||||
|
||||
response = @server.handle(body, session: modern_session)
|
||||
session = modern_session
|
||||
notifications = @mutex.synchronize { @modern_request_sinks[session.session_id] = [] }
|
||||
begin
|
||||
response = @server.handle(body, session: session)
|
||||
ensure
|
||||
@mutex.synchronize { @modern_request_sinks.delete(session.session_id) }
|
||||
end
|
||||
|
||||
# `nil` covers notifications and cancellation-suppressed responses; ack with 202 like the legacy notification path.
|
||||
return handle_accepted if response.nil?
|
||||
|
||||
[modern_http_status(response), { "content-type" => "application/json" }, [response.to_json]]
|
||||
# Notifications a handler emitted during the request ride the request's own response stream as SSE frames ahead of
|
||||
# the final response (SEP-2575); a request that emitted none keeps the single JSON exchange and its HTTP status ladder.
|
||||
# Delivery is buffered: frames flush after the handler returns, preserving order but not real-time interleaving
|
||||
# (the TypeScript and Python SDKs stream live), and the sink grows with the handler's notification count.
|
||||
# Live streaming can follow without changing the wire shape.
|
||||
if notifications.empty?
|
||||
[modern_http_status(response), { "content-type" => "application/json" }, [response.to_json]]
|
||||
else
|
||||
frames = notifications + [response]
|
||||
sse_body = frames.map { |frame| "event: message\ndata: #{frame.to_json}\n\n" }.join
|
||||
[200, SSE_HEADERS.dup, [sse_body]]
|
||||
end
|
||||
rescue StandardError => e
|
||||
MCP.configuration.exception_reporter.call(e, { request: body_string })
|
||||
json_rpc_error_response(
|
||||
|
|
|
|||
|
|
@ -252,7 +252,15 @@ module MCP
|
|||
def notify_log_message(data:, level:, logger: nil, related_request_id: nil)
|
||||
@server.send(:warn_if_deprecated_protocol_feature, :logging, session: self, uplevel: 2)
|
||||
|
||||
effective_logging = @logging_message_notification || @server.logging_message_notification
|
||||
# In the modern lifecycle, log delivery is authorized per request through the `_meta` envelope's `logLevel` member
|
||||
# (applied via `configure_logging`); without it no `notifications/message` is sent, and the server-wide level
|
||||
# does not apply (SEP-2575).
|
||||
effective_logging = if @era == :modern
|
||||
@logging_message_notification
|
||||
else
|
||||
@logging_message_notification || @server.logging_message_notification
|
||||
end
|
||||
|
||||
return unless effective_logging&.should_notify?(level)
|
||||
|
||||
params = { "data" => data, "level" => level }
|
||||
|
|
|
|||
|
|
@ -5418,6 +5418,168 @@ module MCP
|
|||
assert_equal({ "elicitation" => { "form" => {} } }, body.dig("error", "data", "requiredCapabilities"))
|
||||
end
|
||||
|
||||
test "modern POST streams handler notifications ahead of the final response" do
|
||||
@server.define_tool(name: "progress_tool") do |server_context:|
|
||||
server_context.report_progress(50, total: 100)
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
body = {
|
||||
jsonrpc: "2.0",
|
||||
method: "tools/call",
|
||||
id: 1,
|
||||
params: {
|
||||
name: "progress_tool",
|
||||
arguments: {},
|
||||
_meta: {
|
||||
progressToken: "tok-1",
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
},
|
||||
},
|
||||
}.to_json
|
||||
|
||||
response = @transport.handle_request(modern_rack_request(body))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "text/event-stream", response[1]["content-type"]
|
||||
frames = response[2][0].split("\n\n").reject(&:empty?).map do |chunk|
|
||||
JSON.parse(chunk.sub(/\Aevent: message\ndata: /, ""))
|
||||
end
|
||||
assert_equal "notifications/progress", frames.first["method"]
|
||||
assert_equal "tok-1", frames.first.dig("params", "progressToken")
|
||||
assert frames.last.key?("result"), "expected the final frame to carry the result"
|
||||
assert_equal 1, frames.last["id"]
|
||||
end
|
||||
|
||||
test "modern POST bounds the buffered notifications per request" do
|
||||
# The sink holds notifications in memory until the handler returns, so a handler
|
||||
# emitting in proportion to client input must not grow one request without bound.
|
||||
overflow = 5
|
||||
delivery_results = []
|
||||
@server.define_tool(name: "flood_tool") do |server_context:|
|
||||
(StreamableHTTPTransport::MAX_MODERN_REQUEST_NOTIFICATIONS + overflow).times do |i|
|
||||
delivery_results << server_context.notify_log_message(data: "message #{i}", level: "info")
|
||||
end
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
body = {
|
||||
jsonrpc: "2.0",
|
||||
method: "tools/call",
|
||||
id: 1,
|
||||
params: {
|
||||
name: "flood_tool",
|
||||
arguments: {},
|
||||
_meta: {
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
"io.modelcontextprotocol/logLevel": "info",
|
||||
},
|
||||
},
|
||||
}.to_json
|
||||
|
||||
response = @transport.handle_request(modern_rack_request(body))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "text/event-stream", response[1]["content-type"]
|
||||
frames = response[2][0].split("\n\n").reject(&:empty?)
|
||||
assert_equal StreamableHTTPTransport::MAX_MODERN_REQUEST_NOTIFICATIONS + 1, frames.size
|
||||
assert_equal overflow, delivery_results.count { |result| result == false }
|
||||
end
|
||||
|
||||
test "modern POST without handler notifications keeps the single JSON exchange" do
|
||||
response = @transport.handle_request(modern_rack_request(modern_body("tools/list", {})))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "application/json", response[1]["content-type"]
|
||||
end
|
||||
|
||||
test "modern POST suppresses log notifications without the envelope logLevel" do
|
||||
# SEP-2575: log delivery is authorized per request via the `_meta` logLevel;
|
||||
# the server-wide level does not apply to modern requests.
|
||||
@server.logging_message_notification = MCP::LoggingMessageNotification.new(level: "debug")
|
||||
@server.define_tool(name: "logging_tool") do |server_context:|
|
||||
server_context.notify_log_message(data: "hi", level: "info")
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
response = @transport.handle_request(modern_rack_request(
|
||||
modern_body("tools/call", { name: "logging_tool", arguments: {} }),
|
||||
))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "application/json", response[1]["content-type"]
|
||||
refute_includes response[2][0], "notifications/message"
|
||||
end
|
||||
|
||||
test "modern POST delivers log notifications when the envelope carries logLevel" do
|
||||
@server.define_tool(name: "logging_tool") do |server_context:|
|
||||
server_context.notify_log_message(data: "hi", level: "info")
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
body = {
|
||||
jsonrpc: "2.0",
|
||||
method: "tools/call",
|
||||
id: 1,
|
||||
params: {
|
||||
name: "logging_tool",
|
||||
arguments: {},
|
||||
_meta: {
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
"io.modelcontextprotocol/logLevel": "info",
|
||||
},
|
||||
},
|
||||
}.to_json
|
||||
|
||||
response = @transport.handle_request(modern_rack_request(body))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "text/event-stream", response[1]["content-type"]
|
||||
assert_includes response[2][0], "notifications/message"
|
||||
end
|
||||
|
||||
test "modern POST treats an unrecognized envelope logLevel as absent" do
|
||||
# Matching the Python SDK, an unrecognized level reads as absent:
|
||||
# delivery stays off instead of erroring on every log call.
|
||||
reported = []
|
||||
configuration = MCP::Configuration.new(exception_reporter: ->(e, _ctx) { reported << e })
|
||||
server = Server.new(name: "log_level_test", configuration: configuration)
|
||||
transport = StreamableHTTPTransport.new(server, dns_rebinding_protection: false)
|
||||
server.define_tool(name: "logging_tool") do |server_context:|
|
||||
server_context.notify_log_message(data: "hi", level: "info")
|
||||
Tool::Response.new([{ type: "text", text: "done" }])
|
||||
end
|
||||
|
||||
body = {
|
||||
jsonrpc: "2.0",
|
||||
method: "tools/call",
|
||||
id: 1,
|
||||
params: {
|
||||
name: "logging_tool",
|
||||
arguments: {},
|
||||
_meta: {
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
"io.modelcontextprotocol/logLevel": "bogus",
|
||||
},
|
||||
},
|
||||
}.to_json
|
||||
|
||||
response = transport.handle_request(modern_rack_request(body))
|
||||
|
||||
assert_equal 200, response[0]
|
||||
assert_equal "application/json", response[1]["content-type"]
|
||||
refute_includes response[2][0], "notifications/message"
|
||||
assert_empty reported
|
||||
end
|
||||
|
||||
test "modern POST serves server/discover without an envelope" do
|
||||
response = @transport.handle_request(modern_rack_request(
|
||||
{ jsonrpc: "2.0", method: "server/discover", id: 1 }.to_json,
|
||||
|
|
|
|||
|
|
@ -170,6 +170,31 @@ module MCP
|
|||
assert_empty mock_transport.notifications
|
||||
end
|
||||
|
||||
test "ServerSession#notify_log_message on a modern-era session ignores the server-wide level" do
|
||||
# SEP-2575: on the modern lifecycle, log delivery is authorized per request via
|
||||
# the envelope's logLevel; the server-wide level MUST NOT apply.
|
||||
server = Server.new(name: "test_server", version: "1.0.0")
|
||||
mock_transport = MockTransport.new(server)
|
||||
server.logging_message_notification = MCP::LoggingMessageNotification.new(level: "debug")
|
||||
session = ServerSession.new(server: server, transport: mock_transport, era: :modern)
|
||||
|
||||
session.notify_log_message(data: { message: "test" }, level: "error")
|
||||
|
||||
assert_empty mock_transport.notifications
|
||||
end
|
||||
|
||||
test "ServerSession#notify_log_message on a modern-era session honors a per-request level" do
|
||||
server = Server.new(name: "test_server", version: "1.0.0")
|
||||
mock_transport = MockTransport.new(server)
|
||||
session = ServerSession.new(server: server, transport: mock_transport, era: :modern)
|
||||
session.configure_logging(MCP::LoggingMessageNotification.new(level: "info"))
|
||||
|
||||
session.notify_log_message(data: { message: "test" }, level: "error")
|
||||
|
||||
assert_equal 1, mock_transport.notifications.length
|
||||
assert_equal Methods::NOTIFICATIONS_MESSAGE, mock_transport.notifications.first[:method]
|
||||
end
|
||||
|
||||
test "#notify_log_message sends notification with logger through transport" do
|
||||
@server.logging_message_notification = MCP::LoggingMessageNotification.new(level: "error")
|
||||
@server.notify_log_message(data: { error: "Connection Failed" }, level: "error", logger: "DatabaseLogger")
|
||||
|
|
|
|||
Loading…
Reference in a new issue