Skip to content

Commit cb0eb38

Browse files
koickoic
committed
Support SSE Reconnection Per SEP-1699 in the HTTP Client Transport
## Motivation and Context Resolves the `sse-retry` client conformance scenario (SEP-1699, modelcontextprotocol/modelcontextprotocol#1699). Per SEP-1699, a server may close a request's SSE stream right after a priming event (an event carrying an `id:`) without delivering the response, expecting the client to treat the graceful close like a network failure: wait the server-specified `retry:` interval, then reconnect with an HTTP GET carrying `Last-Event-ID` so the server can replay the pending response on the resumed stream. The Ruby client previously read the entire SSE body to EOF and parsed it afterward, so it had no notion of per-event state (`id:`, `retry:`) or reconnection, and the scenario failed all three checks (graceful reconnect, retry timing, `Last-Event-ID`). `MCP::Client::HTTP#send_request` now consumes responses incrementally via Faraday's `on_data` streaming callback, feeding SSE chunks to `event_stream_parser` as they arrive while buffering plain JSON bodies as before. A new internal `SSEStream` tracks the last received event id, the `retry:` reconnection delay, and the awaited JSON-RPC response; once the response arrives, a held-open stream is abandoned via an internal control-flow exception since servers may never close it. When a stream closes gracefully after a priming event but before the response, the client sleeps for the `retry:` interval (default 1000ms when the server sent none), then issues a GET with `Accept: text/event-stream`, the session headers, and `Last-Event-ID`, for up to 2 attempts. The defaults match the Python SDK's `DEFAULT_RECONNECTION_DELAY_MS` and `MAX_RECONNECTION_ATTEMPTS` (`streamable_http.py`); the TypeScript SDK's `StreamableHTTPClientTransport` (`streamableHttp.ts`) likewise lets the `retry:` field override its backoff and reconnects only when a priming event was received and no response has arrived. The public behavior of `send_request` is unchanged: same return values, same error mapping, and a stream that closes without a priming event still raises the "No valid JSON-RPC response found in SSE stream" error without reconnecting. Because the previous implementation read `response.body` and therefore worked with any Faraday adapter, a fallback keeps that compatibility: when nothing was parsed during streaming, the body is read from `response.body` (adapters without `on_data` support, e.g. the Faraday test adapter) or from the buffered chunks (Faraday < 2.1, which invokes `on_data` without `env`). The conformance client gains an `sse-retry` branch that calls the harness's `test_reconnection` tool, and the scenario is removed from the expected failures baseline. ## How Has This Been Tested? New unit tests covering: reconnection with `Last-Event-ID` after a primed graceful close including the `retry:` wait, the default 1000ms delay when `retry:` is absent, raising after the reconnection attempt cap with `Last-Event-ID` advancing across attempts, no reconnection for unprimed streams, and the non-streaming fallbacks (JSON and SSE responses through the Faraday test adapter, which ignores `on_data`, plus buffered SSE chunks when `on_data` receives no `env`) ## Breaking Changes None. The `send_request` contract is unchanged; SSE bodies are now parsed incrementally instead of after EOF, and reconnection only activates when the server opts in by sending a priming event.
1 parent b7bd9bb commit cb0eb38

4 files changed

Lines changed: 363 additions & 34 deletions

File tree

‎conformance/client.rb‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,15 @@ def build_provider_for(scenario, context)
135135
add_numbers = tools.find { |t| t.name == "add_numbers" }
136136
abort("Tool add_numbers not found") unless add_numbers
137137
client.call_tool(tool: add_numbers, arguments: { a: 1, b: 2 })
138+
when "sse-retry"
139+
# SEP-1699: the server closes the tools/call SSE stream right after a priming event.
140+
# The transport waits the server's `retry:` interval, reconnects with a GET carrying `Last-Event-ID`,
141+
# and receives the tool result on the resumed stream; the harness verifies the reconnect,
142+
# its timing, and the header.
143+
tools = client.tools
144+
test_reconnection = tools.find { |t| t.name == "test_reconnection" }
145+
abort("Tool test_reconnection not found") unless test_reconnection
146+
client.call_tool(tool: test_reconnection, arguments: {})
138147
when %r|\Aauth/|
139148
# Auth-only scenarios: the protocol-level checks (PRM/AS metadata, DCR, PKCE, token usage)
140149
# are observed by the conformance server during `connect` and the subsequent request below.

‎conformance/expected_failures.yml‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
server: []
22
client:
3-
# TODO: SSE reconnection not implemented in Ruby client.
4-
- sse-retry
53
# TODO: Elicitation not implemented in Ruby client.
64
- elicitation-sep1034-client-defaults
75
# TODO: Remaining OAuth/auth scenarios not yet implemented in Ruby client.

‎lib/mcp/client/http.rb‎

Lines changed: 190 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -8,16 +8,24 @@
88

99
module MCP
1010
class Client
11-
# TODO: HTTP GET for SSE streaming is not yet implemented.
11+
# TODO: A standalone HTTP GET listening stream for server-initiated messages is not yet implemented;
12+
# GET is currently used only to resume a request's SSE stream after a disconnect.
1213
# https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#listening-for-messages-from-the-server
13-
# TODO: Resumability and redelivery with Last-Event-ID is not yet implemented.
14-
# https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#resumability-and-redelivery
1514
class HTTP
1615
ACCEPT_HEADER = "application/json, text/event-stream"
16+
SSE_ACCEPT_HEADER = "text/event-stream"
1717
SESSION_ID_HEADER = "Mcp-Session-Id"
1818
PROTOCOL_VERSION_HEADER = "MCP-Protocol-Version"
1919
METHOD_HEADER = "Mcp-Method"
2020
NAME_HEADER = "Mcp-Name"
21+
LAST_EVENT_ID_HEADER = "Last-Event-ID"
22+
23+
# SEP-1699 reconnection tuning: the SSE `retry:` field from the server takes precedence;
24+
# this default applies when the server sent none. Both values match the Python SDK
25+
# (`DEFAULT_RECONNECTION_DELAY_MS`, `MAX_RECONNECTION_ATTEMPTS`); the TypeScript SDK uses
26+
# an exponential backoff that the `retry:` field likewise overrides.
27+
DEFAULT_RECONNECTION_DELAY_MS = 1000
28+
MAX_RECONNECTION_ATTEMPTS = 2
2129

2230
# Raised when an `oauth:` provider is paired with an MCP URL that is neither HTTPS nor
2331
# a loopback `http://` URL, since a bearer token sent over plain HTTP to a remote host
@@ -54,6 +62,101 @@ def call(env)
5462
end
5563
end
5664

65+
# Internal control-flow signal raised inside the streaming callback to stop reading
66+
# an SSE stream once the awaited JSON-RPC response has arrived; servers may hold
67+
# a stream open indefinitely (notably the standalone GET stream), so EOF cannot be relied on.
68+
class StreamAbort < StandardError; end
69+
private_constant :StreamAbort
70+
71+
# Per-exchange SSE state shared between the initial POST stream and any SEP-1699 reconnection GET streams:
72+
# the incrementally parsed JSON-RPC response, the last received SSE event id (for `Last-Event-ID`),
73+
# and the server's `retry:` reconnection delay. Non-SSE bodies accumulate in `buffer` for the JSON path.
74+
class SSEStream
75+
attr_reader :buffer, :response, :last_event_id, :retry_ms
76+
attr_accessor :abortable
77+
78+
def initialize(abortable:)
79+
@abortable = abortable
80+
@buffer = +""
81+
@parser = nil
82+
@response = nil
83+
@last_event_id = nil
84+
@retry_ms = nil
85+
end
86+
87+
# Whether the server sent a SEP-1699 priming event (any event carrying an id),
88+
# which marks the stream as resumable after a graceful close.
89+
def primed?
90+
!@last_event_id.nil?
91+
end
92+
93+
# Faraday `on_data` streaming callback. SSE chunks are parsed incrementally;
94+
# anything else (JSON bodies) accumulates in `buffer`.
95+
def on_data
96+
proc do |chunk, _received_bytes, env|
97+
if event_stream?(env)
98+
feed(chunk)
99+
raise StreamAbort if @abortable && @response
100+
else
101+
@buffer << chunk
102+
end
103+
end
104+
end
105+
106+
# A fresh parser for a new HTTP connection (reconnection GET),
107+
# so a partial line from the previous stream cannot corrupt the next one.
108+
def reset_parser!
109+
@parser = nil
110+
end
111+
112+
# Parses an SSE body that was delivered outside the streaming callback:
113+
# Faraday versions that do not pass `env` to `on_data` leave SSE chunks in `buffer`
114+
# (consumed here so they are not parsed twice), and adapters without `on_data` support
115+
# yield the whole body only via `fallback_body` (the Faraday `response.body`).
116+
def ingest_pending!(fallback_body)
117+
text = @buffer.empty? ? fallback_body : @buffer.slice!(0..-1)
118+
feed(text) if text.is_a?(String) && !text.empty?
119+
end
120+
121+
private
122+
123+
def event_stream?(env)
124+
headers = env&.response_headers
125+
content_type = headers && (headers["content-type"] || headers["Content-Type"])
126+
!!content_type&.include?("text/event-stream")
127+
end
128+
129+
def feed(chunk)
130+
parser.feed(chunk) do |_type, data, id, reconnection_time|
131+
@last_event_id = id if id && !id.empty?
132+
@retry_ms = reconnection_time if reconnection_time
133+
next if data.nil? || data.empty?
134+
135+
begin
136+
parsed = JSON.parse(data)
137+
rescue JSON::ParserError
138+
next
139+
end
140+
141+
if parsed.is_a?(Hash) && (parsed.key?("result") || parsed.key?("error"))
142+
@response ||= parsed
143+
end
144+
end
145+
end
146+
147+
def parser
148+
@parser ||= begin
149+
require "event_stream_parser"
150+
EventStreamParser::Parser.new
151+
rescue LoadError
152+
raise LoadError, "The 'event_stream_parser' gem is required to parse SSE responses. " \
153+
"Add it to your Gemfile: gem 'event_stream_parser', '>= 1.0'. " \
154+
"See https://rubygems.org/gems/event_stream_parser for more details."
155+
end
156+
end
157+
end
158+
private_constant :SSEStream
159+
57160
attr_reader :url, :session_id, :protocol_version, :server_info, :oauth
58161

59162
def initialize(url:, headers: {}, oauth: nil, &block)
@@ -199,11 +302,22 @@ def send_request(request:)
199302
step_up_retried = false
200303

201304
begin
305+
# The response is consumed incrementally so that an SSE stream the server holds open
306+
# (or closes early per SEP-1699) can be handled; `initialize` streams are read to EOF
307+
# so the response object (and its `Mcp-Session-Id` header) is always available for capture.
308+
stream = SSEStream.new(abortable: method.to_s != MCP::Methods::INITIALIZE)
202309
yield if block_given?
203-
response = client.post("", request, session_headers.merge(request_metadata_headers(method, params)))
204-
body = parse_response_body(response, method, params)
310+
response = begin
311+
client.post("", request, session_headers.merge(request_metadata_headers(method, params))) do |req|
312+
req.options.on_data = stream.on_data
313+
end
314+
rescue StreamAbort
315+
nil
316+
end
317+
318+
body = resolve_response_body(stream, response, method, params)
205319

206-
capture_session_info(method, response, body)
320+
capture_session_info(method, response, body) if response
207321

208322
body
209323
rescue Faraday::BadRequestError => e
@@ -536,25 +650,41 @@ def require_faraday!
536650
"See https://rubygems.org/gems/faraday for more details."
537651
end
538652

539-
def require_event_stream_parser!
540-
require "event_stream_parser"
541-
rescue LoadError
542-
raise LoadError, "The 'event_stream_parser' gem is required to parse SSE responses. " \
543-
"Add it to your Gemfile: gem 'event_stream_parser', '>= 1.0'. " \
544-
"See https://rubygems.org/gems/event_stream_parser for more details."
545-
end
653+
# Determines the logical JSON-RPC body of the exchange after the POST stream finished
654+
# (or was aborted because the response already arrived).
655+
def resolve_response_body(stream, response, method, params)
656+
return stream.response if stream.response
546657

547-
def parse_response_body(response, method, params)
548658
# 202 Accepted is the server's ACK for a JSON-RPC notification or response; no body is expected.
549659
# https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#sending-messages-to-the-server
550660
return if response.status == 202
551661

552662
content_type = response.headers["Content-Type"]
553663

554664
if content_type&.include?("text/event-stream")
555-
parse_sse_response(response.body, method, params)
665+
# Nothing was parsed during streaming: either the adapter does not support `on_data`
666+
# (the whole body sits in `response.body`) or Faraday did not pass `env` to `on_data`
667+
# (Faraday < 2.1; the SSE chunks sit in `buffer`). Parse whichever holds the body.
668+
unless stream.primed?
669+
stream.ingest_pending!(response.body)
670+
return stream.response if stream.response
671+
end
672+
673+
# SEP-1699: a graceful close after a priming event (an event with an id) but before
674+
# the response means "reconnect via GET to resume".
675+
return await_response_after_disconnect(stream, method, params) if stream.primed?
676+
677+
raise RequestHandlerError.new(
678+
"No valid JSON-RPC response found in SSE stream",
679+
{ method: method, params: params },
680+
error_type: :parse_error,
681+
)
556682
elsif content_type&.include?("application/json")
557-
response.body
683+
return parse_json_buffer(stream.buffer, method, params) unless stream.buffer.empty?
684+
685+
# Adapters without `on_data` support deliver the body via `response.body`,
686+
# already parsed by the json response middleware.
687+
response.body.is_a?(String) ? parse_json_buffer(response.body, method, params) : response.body
558688
else
559689
raise RequestHandlerError.new(
560690
"Unsupported Content-Type: #{content_type.inspect}. Expected application/json or text/event-stream.",
@@ -564,28 +694,57 @@ def parse_response_body(response, method, params)
564694
end
565695
end
566696

567-
def parse_sse_response(body, method, params)
568-
require_event_stream_parser!
697+
def parse_json_buffer(buffer, method, params)
698+
return if buffer.empty?
569699

570-
json_rpc_response = nil
571-
parser = EventStreamParser::Parser.new
572-
parser.feed(body.to_s) do |_type, data, _id|
573-
next if data.empty?
700+
JSON.parse(buffer)
701+
rescue JSON::ParserError => e
702+
raise RequestHandlerError.new(
703+
"Failed to parse JSON response: #{e.message}",
704+
{ method: method, params: params },
705+
error_type: :parse_error,
706+
)
707+
end
574708

575-
begin
576-
parsed = JSON.parse(data)
577-
json_rpc_response = parsed if parsed.is_a?(Hash) && (parsed.key?("result") || parsed.key?("error"))
578-
rescue JSON::ParserError
579-
next
709+
# SEP-1699 resumability: the server closed the SSE stream after a priming event
710+
# without delivering the response. Treat the graceful close like a network failure:
711+
# wait the server-specified `retry:` interval (default 1000ms), then reconnect with
712+
# a GET carrying `Last-Event-ID` so the server can replay the pending response on
713+
# the standalone stream. Mirrors the TypeScript SDK's `StreamableHTTPClientTransport`
714+
# reconnection and the Python SDK's `_handle_reconnection` (including its 2-attempt cap).
715+
# https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699
716+
def await_response_after_disconnect(stream, method, params)
717+
stream.abortable = true
718+
719+
MAX_RECONNECTION_ATTEMPTS.times do
720+
sleep((stream.retry_ms || DEFAULT_RECONNECTION_DELAY_MS) / 1000.0)
721+
stream.reset_parser!
722+
723+
reconnect_response = begin
724+
client.get("") do |req|
725+
req.headers.update(session_headers)
726+
req.headers["Accept"] = SSE_ACCEPT_HEADER
727+
req.headers[LAST_EVENT_ID_HEADER] = stream.last_event_id if stream.last_event_id
728+
req.options.on_data = stream.on_data
729+
end
730+
rescue StreamAbort
731+
# The awaited response arrived on the reconnected stream.
732+
nil
580733
end
581-
end
582734

583-
return json_rpc_response if json_rpc_response
735+
# Same fallback as `resolve_response_body` for adapters that do not stream via `on_data`.
736+
if reconnect_response && stream.response.nil?
737+
stream.ingest_pending!(reconnect_response.body)
738+
end
739+
740+
return stream.response if stream.response
741+
end
584742

585743
raise RequestHandlerError.new(
586-
"No valid JSON-RPC response found in SSE stream",
744+
"Server closed the SSE stream without a response for #{method} " \
745+
"after #{MAX_RECONNECTION_ATTEMPTS} reconnection attempts",
587746
{ method: method, params: params },
588-
error_type: :parse_error,
747+
error_type: :internal_error,
589748
)
590749
end
591750
end

0 commit comments

Comments
 (0)