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.



410
411
412
413
414
415
416
417
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 410

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

#close_streams(streams) ⇒ Object 



296
297
298
299
300
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 296

def close_streams(streams)
  streams.each do |stream|
    close_stream_safely(stream)
  end
end

#deliver_broadcast_notification(notification) ⇒ Object 



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
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 234

def deliver_broadcast_notification(notification)
  streams_to_close = []

  # Snapshot the connected streams under the lock, then write to each outside it.
  targets = @mutex.synchronize do
    expired_session_ids = []

    collected = @sessions.filter_map do |session_id, session|
      next unless (stream = session[:get_sse_stream])

      if session_expired?(session)
        expired_session_ids << session_id
        next
      end

      [session_id, stream]
    end

    expired_session_ids.each do |session_id|
      cleanup_and_collect_stream(session_id, streams_to_close)
    end

    collected
  end

  close_streams(streams_to_close)

  targets.count do |session_id, stream|
    write_notification(stream, notification, session_id, nil)
  end
end

#deliver_targeted_notification(notification, session_id, related_request_id) ⇒ Object 



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
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 206

def deliver_targeted_notification(notification, session_id, related_request_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.
  return false if @enable_json_response && related_request_id

  streams_to_close = []

  # Resolve the target stream under the lock, then write outside it: a stalled SSE reader
  # must not block every other session that needs `@mutex`.
  stream = @mutex.synchronize do
    next unless (session = @sessions[session_id])

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

    active_stream(session, related_request_id: related_request_id)
  end

  close_streams(streams_to_close)
  return false unless stream

  write_notification(stream, notification, session_id, related_request_id)
end

#drop_broken_stream(session_id, stream, related_request_id) ⇒ Object

Removes a stream that failed to accept a write. A request-scoped stream is dropped on its own; a session-scoped (GET SSE) failure tears down the whole session. The @sessions mutation runs under @mutex, and the affected streams are closed outside it.



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 280

def drop_broken_stream(session_id, stream, related_request_id)
  streams_to_close = []

  @mutex.synchronize do
    session = @sessions[session_id]
    if related_request_id && session&.dig(:post_request_streams, 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
  end

  close_streams(streams_to_close)
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
 
# 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

  if session_id
    deliver_targeted_notification(notification, session_id, related_request_id)
  else
    deliver_broadcast_notification(notification)
  end
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.



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
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 309

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

  # Register the pending response and resolve the stream under the lock, but perform
  # the write outside it so a stalled reader cannot block every other session on `@mutex`.
  stream = @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 }

    active_stream(session, related_request_id: related_request_id)
  end

  sent = false
  if stream
    begin
      send_to_stream(stream, request)
      sent = true
    rescue *STREAM_WRITE_ERRORS
      drop_broken_stream(session_id, stream, related_request_id)
    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

#write_notification(stream, notification, session_id, related_request_id) ⇒ Object

Writes a notification to an SSE stream without holding @mutex. On a write error, drops the broken stream and returns false; on success returns true.



268
269
270
271
272
273
274
275
 
# File 'lib/mcp/server/transports/streamable_http_transport.rb', line 268

def write_notification(stream, notification, session_id, related_request_id)
  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" })
  drop_broken_stream(session_id, stream, related_request_id)
  false
end