Class: MCP::Server::Transports::StreamableHTTPTransport

Inherits:
Transport
  • Object
show all
Defined in:
lib/mcp/server/transports/streamable_http_transport.rb

Defined Under Namespace

Classes: InvalidJsonError

Constant Summary collapse

SSE_HEADERS = 
{
  "Content-Type" => "text/event-stream",
  "Cache-Control" => "no-cache",
  "Connection" => "keep-alive",
}.freeze
DEFAULT_SESSION_IDLE_TIMEOUT =

Secure defaults for stateful mode. Without a finite idle timeout, sessions live until an explicit client DELETE, so an unauthenticated initialize flood retains unbounded ServerSession objects until memory is exhausted. These defaults expire idle sessions and cap the concurrent count, like the C# SDK (the only reference SDK that hardens this by default, with a 2h idle timeout and a 10k idle-session count). One difference: at the cap this transport rejects a new initialize with 503 (after reclaiming any already-expired slots), whereas the C# SDK evicts the oldest idle session. Rejecting keeps established sessions stable and avoids evicting a legitimate idle session on an attacker's behalf, at the cost of refusing new sessions while genuinely full. Pass session_idle_timeout: nil to opt out of expiry and max_sessions: nil to opt out of the cap.

1800
DEFAULT_MAX_SESSIONS = 
10_000
DEFAULT_MAX_REQUEST_BYTES =

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 fits a typical JSON-RPC request (a 4 MiB JSON string decodes to ~3 MiB of base64 payload); raise max_request_bytes: for unusually large payloads. Matches the TypeScript SDK's 4 MB default.

4 * 1024 * 1024
MAX_JSON_NESTING =

Conservative bound on JSON nesting depth, so a deeply nested body cannot exhaust the stack or amplify parse cost (complements the byte cap).

64
REQUIRED_POST_ACCEPT_TYPES_SSE = 
["application/json", "text/event-stream"].freeze
REQUIRED_POST_ACCEPT_TYPES_JSON = 
["application/json"].freeze
REQUIRED_GET_ACCEPT_TYPES = 
["text/event-stream"].freeze
STREAM_WRITE_ERRORS = 
[IOError, Errno::EPIPE, Errno::ECONNRESET].freeze
SESSION_REAP_INTERVAL = 
60
DEFAULT_LOOPBACK_HOSTS =

Loopback hosts always accepted by DNS rebinding protection. A locally bound MCP server (the canonical pattern) is protected out of the box; non-loopback deployments widen the list via allowed_hosts:.

["127.0.0.1", "::1", "localhost"].freeze

Instance Method Summary collapse

Methods inherited from Transport

#handle_json_request, #open, #send_response

Constructor Details

#initialize(server, stateless: false, enable_json_response: false, session_idle_timeout: UNSET_IDLE_TIMEOUT, max_sessions: DEFAULT_MAX_SESSIONS, allowed_origins: nil, allowed_hosts: nil, dns_rebinding_protection: true, session_request_validator: nil, max_request_bytes: DEFAULT_MAX_REQUEST_BYTES) ⇒ StreamableHTTPTransport

Creates a Streamable HTTP transport that can be mounted as a Rack app.

Parameters:

  • server (MCP::Server) —

    the server whose requests this transport dispatches.

  • stateless (Boolean) (defaults to: false) —

    when true, no session is issued and each POST is self-contained.

  • enable_json_response (Boolean) (defaults to: false) —

    when true, a request is answered with a single JSON object instead of an SSE stream.

  • session_idle_timeout (Numeric, nil) (defaults to: UNSET_IDLE_TIMEOUT) —

    seconds before an idle session is reaped; defaults to DEFAULT_SESSION_IDLE_TIMEOUT (1800) in stateful mode, and an explicit nil disables expiry. Not supported in stateless mode.

  • max_sessions (Integer, nil) (defaults to: DEFAULT_MAX_SESSIONS) —

    cap on the concurrent session count in stateful mode; a new initialize past the cap is rejected with HTTP 503, and nil disables the cap.

  • allowed_origins (Array<String>, nil) (defaults to: nil) —

    extra Origin values accepted in addition to same-origin requests, for DNS rebinding protection.

  • allowed_hosts (Array<String>, nil) (defaults to: nil) —

    extra Host values accepted beyond the loopback defaults (127.0.0.1, ::1, localhost); each entry matches a bare host name (any port) or a full host:port.

  • dns_rebinding_protection (Boolean) (defaults to: true) —

    when true (default), validates the Host and Origin headers to prevent DNS rebinding; pass false when an upstream proxy already validates them.

  • session_request_validator (#call, nil) (defaults to: nil) —

    An optional ->(request, session_id) { true | false } invoked on every non-initialize POST, GET, and DELETE against an existing session (regular requests, notifications, and client responses alike). Returning a falsy value rejects the request with HTTP 403. The SDK issues a random SecureRandom.uuid session ID and otherwise only checks existence/idle-timeout, so binding a session to a user is the deploying application's responsibility (the transport never receives the authenticated identity on its own); this is the seam to enforce ownership and mitigate session poisoning. Without a validator, ownership is not enforced.

  • max_request_bytes (Integer) (defaults to: DEFAULT_MAX_REQUEST_BYTES) —

    upper bound in bytes on a POST request body; larger requests are rejected with HTTP 413. Defaults to 4 MiB.



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 83

def initialize(
  server,
  stateless: false,
  enable_json_response: false,
  session_idle_timeout: UNSET_IDLE_TIMEOUT,
  max_sessions: DEFAULT_MAX_SESSIONS,
  allowed_origins: nil,
  allowed_hosts: nil,
  dns_rebinding_protection: true,
  session_request_validator: nil,
  max_request_bytes: DEFAULT_MAX_REQUEST_BYTES
)
  super(server)
  # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock, origin: origin_header }`.
  @sessions = {}
  @mutex = Mutex.new

  @stateless = stateless
  @enable_json_response = enable_json_response
  @session_request_validator = session_request_validator
  @dns_rebinding_protection = dns_rebinding_protection

  # Host names are case-insensitive, so the allow lists are compared down-cased.
  @allowed_hosts = (DEFAULT_LOOPBACK_HOSTS + Array(allowed_hosts)).map(&:downcase).freeze
  @allowed_origins = Array(allowed_origins).map(&:downcase).freeze
  @pending_responses = {}

  # 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)
    stateless ? nil : DEFAULT_SESSION_IDLE_TIMEOUT
  else
    session_idle_timeout
  end

  if @session_idle_timeout
    if @stateless
      raise ArgumentError, "session_idle_timeout is not supported in stateless mode."
    elsif @session_idle_timeout <= 0
      raise ArgumentError, "session_idle_timeout must be a positive number."
    end
  end

  unless max_sessions.nil? || (max_sessions.is_a?(Integer) && max_sessions > 0)
    raise ArgumentError, "max_sessions must be a positive Integer or nil"
  end

  # The cap guards the stateful session store; stateless mode keeps none.
  @max_sessions = stateless ? nil : max_sessions

  unless max_request_bytes.is_a?(Integer) && max_request_bytes > 0
    raise ArgumentError, "max_request_bytes must be a positive Integer"
  end

  @max_request_bytes = max_request_bytes

  start_reaper_thread if @session_idle_timeout
end

Instance Method Details

#call(env) ⇒ Object

Rack app interface. This transport can be mounted as a Rack app.



153
154
155
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 153

def call(env)
  handle_request(Rack::Request.new(env))
end

#cancel_pending_request(request_id, reason: nil) ⇒ Object

Unblocks a send_request awaiting a response when the peer is being cancelled. The waiting thread will see :cancelled on its queue and raise MCP::CancelledError.

Race note: this is first-writer-wins on the pending-response queue. If a real response has already been pushed (client responded before the cancel hook fired), that response wins and :cancelled is enqueued behind it but never read - send_request returns the real response and deletes the pending entry in its ensure block. Conversely, if :cancelled arrives first, any later client response is silently dropped in handle_response because the pending entry has been removed.



385
386
387
388
389
390
391
392
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 385

def cancel_pending_request(request_id, reason: nil)
  @mutex.synchronize do
    if (pending = @pending_responses[request_id])
      pending[:cancel_reason] = reason
      pending[:queue].push(:cancelled)
    end
  end
end

#close ⇒ Object 



173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 173

def close
  @reaper_thread&.kill
  @reaper_thread = nil

  removed_sessions = @mutex.synchronize do
    @sessions.each_key.filter_map { |session_id| cleanup_session_unsafe(session_id) }
  end

  removed_sessions.each do |session|
    close_stream_safely(session[:get_sse_stream])
    close_post_request_streams(session)
  end
end

#handle_request(request) ⇒ Object 



157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 157

def handle_request(request)
  rebinding_error = validate_dns_rebinding(request)
  return rebinding_error if rebinding_error

  case request.env["REQUEST_METHOD"]
  when "POST"
    handle_post(request)
  when "GET"
    handle_get(request)
  when "DELETE"
    handle_delete(request)
  else
    method_not_allowed_response
  end
end

#send_notification(method, params = nil, session_id: nil, related_request_id: nil) ⇒ Object 



187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 187

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

  streams_to_close = []

  result = @mutex.synchronize do
    if session_id
      # JSON response mode returns a single JSON object as the POST response,
      # so request-scoped notifications (e.g. progress, log) cannot be delivered
      # alongside it. Session-scoped standalone notifications
      # (e.g. `resources/updated`, `elicitation/complete`) still flow via GET SSE.
      next false if @enable_json_response && related_request_id

      # Send to specific session
      if (session = @sessions[session_id])
        stream = active_stream(session, related_request_id: related_request_id)
      end
      next false unless stream

      if session_expired?(session)
        cleanup_and_collect_stream(session_id, streams_to_close)
        next false
      end

      begin
        send_to_stream(stream, notification)
        true
      rescue *STREAM_WRITE_ERRORS => e
        MCP.configuration.exception_reporter.call(
          e,
          { session_id: session_id, error: "Failed to send notification" },
        )
        if related_request_id && session[:post_request_streams]&.key?(related_request_id)
          session[:post_request_streams].delete(related_request_id)
          streams_to_close << stream
        else
          cleanup_and_collect_stream(session_id, streams_to_close)
        end
        false
      end
    else
      # Broadcast to all connected SSE sessions
      sent_count = 0
      failed_sessions = []

      @sessions.each do |sid, session|
        next unless (stream = session[:get_sse_stream])

        if session_expired?(session)
          failed_sessions << sid
          next
        end

        begin
          send_to_stream(stream, notification)
          sent_count += 1
        rescue *STREAM_WRITE_ERRORS => e
          MCP.configuration.exception_reporter.call(
            e,
            { session_id: sid, error: "Failed to send notification" },
          )
          failed_sessions << sid
        end
      end

      # Clean up failed sessions
      failed_sessions.each { |sid| cleanup_and_collect_stream(sid, streams_to_close) }

      sent_count
    end
  end

  streams_to_close.each do |stream|
    close_stream_safely(stream)
  end

  result
end

#send_request(method, params = nil, session_id: nil, related_request_id: nil, parent_cancellation: nil, server_session: nil) ⇒ Object

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.



282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 282

def send_request(method, params = nil, session_id: nil, related_request_id: nil, parent_cancellation: nil, server_session: nil)
  if @stateless
    raise "Stateless mode does not support server-to-client requests."
  end

  if @enable_json_response
    raise "JSON response 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
  cancel_hook = nil

  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 = active_stream(session, related_request_id: related_request_id))
      begin
        send_to_stream(stream, request)
        sent = true
      rescue *STREAM_WRITE_ERRORS
        if related_request_id && session[:post_request_streams]&.key?(related_request_id)
          session[:post_request_streams].delete(related_request_id)
          close_stream_safely(stream)
        else
          cleanup_session_unsafe(session_id)
        end
      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 stream for #{method} request."
  end

  if parent_cancellation && server_session
    cancel_hook = parent_cancellation.on_cancel do |reason|
      server_session.send_peer_cancellation(
        nested_request_id: request_id,
        related_request_id: related_request_id,
        reason: reason,
      )
    end
  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

  if response == :cancelled
    reason = @mutex.synchronize { @pending_responses.dig(request_id, :cancel_reason) }
    raise MCP::CancelledError.new(
      "#{method} request was cancelled",
      request_id: request_id,
      reason: reason,
    )
  end

  response
ensure
  parent_cancellation.off_cancel(cancel_hook) if cancel_hook
  if request_id
    @mutex.synchronize do
      @pending_responses.delete(request_id)
    end
  end
end