Class: MCP::Client::HTTP

Inherits:
Object
  • Object
show all
Defined in:
lib/mcp/client/http.rb

Defined Under Namespace

Classes: InsecureURLError, OAuthURLGuard

Constant Summary collapse

ACCEPT_HEADER = 
"application/json, text/event-stream"
SSE_ACCEPT_HEADER = 
"text/event-stream"
SESSION_ID_HEADER = 
"Mcp-Session-Id"
PROTOCOL_VERSION_HEADER = 
"MCP-Protocol-Version"
METHOD_HEADER = 
"Mcp-Method"
NAME_HEADER = 
"Mcp-Name"
LAST_EVENT_ID_HEADER = 
"Last-Event-ID"
DEFAULT_RECONNECTION_DELAY_MS =

SEP-1699 reconnection tuning: the SSE retry: field from the server takes precedence; this default applies when the server sent none. Both values match the Python SDK (DEFAULT_RECONNECTION_DELAY_MS, MAX_RECONNECTION_ATTEMPTS); the TypeScript SDK uses an exponential backoff that the retry: field likewise overrides.

1000
MAX_RECONNECTION_ATTEMPTS = 
2
SSE_LISTENER_READ_TIMEOUT =

How long the standalone GET listening stream may stay idle before the read times out and the connection is counted as a failure and retried. Matches the Python SDK's sse_read_timeout default of 5 minutes; without this, the adapter's default read timeout (60 seconds for Net::HTTP) would recycle quiet streams too eagerly.

300
MAX_MESSAGE_BYTES =

Upper bound in bytes on a single JSON-RPC message from the server - an SSE event or a JSON response body - buffered in memory while reading a response. Without a bound, a server that never terminates an SSE event (or never ends a JSON body) grows the buffer indefinitely. Matches the 4 MiB default of MCP::Client::Stdio::MAX_LINE_BYTES and the server transports' request cap.

4 * 1024 * 1024

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url:, headers: {}, oauth: nil, max_message_bytes: MAX_MESSAGE_BYTES, &block) ⇒ HTTP

Returns a new instance of HTTP.



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
 
# File 'lib/mcp/client/http.rb', line 210

def initialize(url:, headers: {}, oauth: nil, max_message_bytes: MAX_MESSAGE_BYTES, &block)
  # `nil` or a non-positive value would make the buffering unbounded and silently
  # disable the protection, so reject it up front.
  unless max_message_bytes.is_a?(Integer) && max_message_bytes > 0
    raise ArgumentError, "max_message_bytes must be a positive Integer"
  end

  if oauth && !MCP::Client::OAuth::Discovery.secure_url?(url)
    # Mask credentials (userinfo) and query parameters before quoting the URL in the error message
    # so they cannot leak into logs.
    safe_url = MCP::Client::OAuth::Discovery.canonicalize_origin_and_path(url)
    raise InsecureURLError,
      "MCP URL #{safe_url.inspect} must use https or be a loopback http URL when an oauth provider is set; " \
        "sending bearer tokens over plain http to a remote host would leak them on the wire."
  end

  @url = url
  @headers = headers
  @faraday_customizer = block
  @oauth = oauth
  @max_message_bytes = max_message_bytes
  # Snapshot the canonical URL at construction time. This single value
  # serves two related roles, both of which need to see the query string:
  #
  # - As the RFC 8707 `resource` claim sent on the authorization and
  #   token requests (and as the base for PRM discovery URLs) -
  #   matching the TS / Python SDKs' `resourceUrlFromServerUrl` /
  #   `resource_url_from_server_url` so multi-tenant servers that scope
  #   by `?tenant=...` round-trip correctly.
  # - As the comparison value for the URL guard middleware. Comparing
  #   query strings as well as origin + path is required so a Faraday
  #   middleware that rewrites `env.url.query` to a different tenant
  #   cannot send the bearer token to the wrong audience while
  #   the resource binding on the OAuth side stays correct.
  #
  # Saved only when `oauth:` is set so non-OAuth transports keep their
  # existing behavior.
  @oauth_server_url = oauth ? MCP::Client::OAuth::Discovery.canonicalize_url(url) : nil
  @session_id = nil
  @protocol_version = nil
  @server_info = nil
  @connected = false
  @server_request_handlers = {}
  @listener_thread = nil
end

Instance Attribute Details

#oauth ⇒ Object (readonly)

Returns the value of attribute oauth.



208
209
210
 
# File 'lib/mcp/client/http.rb', line 208

def oauth
  @oauth
end

#protocol_version ⇒ Object (readonly)

Returns the value of attribute protocol_version.



208
209
210
 
# File 'lib/mcp/client/http.rb', line 208

def protocol_version
  @protocol_version
end

#server_info ⇒ Object (readonly)

Returns the value of attribute server_info.



208
209
210
 
# File 'lib/mcp/client/http.rb', line 208

def server_info
  @server_info
end

#session_id ⇒ Object (readonly)

Returns the value of attribute session_id.



208
209
210
 
# File 'lib/mcp/client/http.rb', line 208

def session_id
  @session_id
end

#url ⇒ Object (readonly)

Returns the value of attribute url.



208
209
210
 
# File 'lib/mcp/client/http.rb', line 208

def url
  @url
end

Instance Method Details

#close ⇒ Object

Terminates the session by sending an HTTP DELETE to the MCP endpoint with the current Mcp-Session-Id header, and clears locally tracked session state afterward. No-op when no session has been established.

Per spec, the server MAY respond with HTTP 405 Method Not Allowed when it does not support client-initiated termination, and returns 404 for a session it has already terminated. Both mean the session is gone — the desired end state. Other errors surface to the caller; local session state is cleared either way. https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#session-management



516
517
518
519
520
521
522
523
524
525
526
527
528
529
 
# File 'lib/mcp/client/http.rb', line 516

def close
  unless @session_id
    clear_session
    return
  end

  begin
    client.delete("", nil, session_headers)
  rescue Faraday::ClientError => e
    raise unless [404, 405].include?(e.response&.dig(:status))
  ensure
    clear_session
  end
end

#connect(client_info: nil, protocol_version: nil, capabilities: {}) ⇒ Hash

Performs the MCP initialize handshake: sends an initialize request followed by the required notifications/initialized notification. The server's InitializeResult (protocol version, capabilities, server info, instructions) is cached on the transport and returned.

Idempotent: a second call returns the cached InitializeResult without contacting the server. After close, state is cleared and connect will handshake again.

https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#initialization

Parameters:

  • client_info (Hash, nil) (defaults to: nil) —

    { name:, version: } identifying the client. Defaults to { name: "mcp-ruby-client", version: MCP::VERSION }.

  • protocol_version (String, nil) (defaults to: nil) —

    Protocol version to offer. Defaults to MCP::Configuration::LATEST_STABLE_PROTOCOL_VERSION.

  • capabilities (Hash) (defaults to: {}) —

    Capabilities advertised by the client. Defaults to {}.

Returns:

  • (Hash) —

    The server's InitializeResult.

Raises:

  • (RequestHandlerError) —

    If the server responds with a JSON-RPC error or a malformed result.



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
 
# File 'lib/mcp/client/http.rb', line 290

def connect(client_info: nil, protocol_version: nil, capabilities: {})
  return @server_info if connected?

  client_info ||= { name: "mcp-ruby-client", version: MCP::VERSION }
  protocol_version ||= MCP::Configuration::LATEST_STABLE_PROTOCOL_VERSION

  response = send_request(request: {
    jsonrpc: JsonRpcHandler::Version::V2_0,
    id: SecureRandom.uuid,
    method: MCP::Methods::INITIALIZE,
    params: {
      protocolVersion: protocol_version,
      capabilities: capabilities,
      clientInfo: client_info,
    },
  })

  if response.is_a?(Hash) && response.key?("error")
    clear_session
    error = response["error"]
    raise RequestHandlerError.new(
      "Server initialization failed: #{error["message"]}",
      { method: MCP::Methods::INITIALIZE },
      error_type: :internal_error,
    )
  end

  unless response.is_a?(Hash) && response["result"].is_a?(Hash)
    clear_session
    raise RequestHandlerError.new(
      "Server initialization failed: missing result in response",
      { method: MCP::Methods::INITIALIZE },
      error_type: :internal_error,
    )
  end

  @server_info = response["result"]
  negotiated_protocol_version = @server_info["protocolVersion"]
  unless MCP::Configuration::SUPPORTED_STABLE_PROTOCOL_VERSIONS.include?(negotiated_protocol_version)
    clear_session
    raise RequestHandlerError.new(
      "Server initialization failed: unsupported protocol version #{negotiated_protocol_version.inspect}",
      { method: MCP::Methods::INITIALIZE },
      error_type: :internal_error,
    )
  end

  begin
    send_request(request: {
      jsonrpc: JsonRpcHandler::Version::V2_0,
      method: MCP::Methods::NOTIFICATIONS_INITIALIZED,
    })
  rescue StandardError
    clear_session
    raise
  end

  @connected = true
  start_listening if @server_request_handlers.any?
  @server_info
end

#connected? ⇒ Boolean

Returns true once connect has completed the full handshake (initialize response received and notifications/initialized sent). Returns false before the first handshake and after close.

Returns:

  • (Boolean) 


355
356
357
 
# File 'lib/mcp/client/http.rb', line 355

def connected?
  @connected
end

#on_server_request(method, &handler) ⇒ Object

Registers a handler for a server-to-client request (e.g. elicitation/create) delivered on an SSE stream. The handler receives the request's params (a Hash with string keys, possibly empty) and its return value is sent back to the server as the JSON-RPC result. The handler may raise MCP::Client::ServerRequestError to answer with a specific JSON-RPC error code. Requests for methods without a registered handler are answered with a JSON-RPC "method not found" (-32601) error. Registering a handler opens a standalone GET SSE listening stream (once connected), since servers send requests that are not tied to a client request on that stream - matching the TypeScript and Python SDK clients, which start listening after the initialize handshake. https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#listening-for-messages-from-the-server

Raises:

  • (ArgumentError) 


265
266
267
268
269
270
 
# File 'lib/mcp/client/http.rb', line 265

def on_server_request(method, &handler)
  raise ArgumentError, "A handler block is required" unless handler

  @server_request_handlers[method.to_s] = handler
  start_listening if connected?
end

#send_notification(notification:) ⇒ Object

Sends a JSON-RPC notification (no response expected). Used by Client#cancel to deliver notifications/cancelled for an in-flight request. The server acknowledges with HTTP 202 Accepted per the Streamable HTTP spec.



492
493
494
495
496
497
498
499
500
501
502
503
504
 
# File 'lib/mcp/client/http.rb', line 492

def send_notification(notification:)
  method = notification[:method] || notification["method"]

  client.post("", notification, session_headers)
  nil
rescue Faraday::Error => e
  raise RequestHandlerError.new(
    "Failed to send #{method} notification",
    { method: method },
    error_type: :internal_error,
    original_error: e,
  )
end

#send_request(request:) ⇒ Object

Sends a JSON-RPC request and returns the parsed response body. After a successful initialize handshake, the session ID and protocol version returned by the server are captured and automatically included on subsequent requests.

If a block is given, it is invoked just before Faraday's post is called. Faraday's synchronous post does not expose a post-write / pre-response hook, so this is the latest send-boundary signal the adapter exposes; the actual TCP write happens inside post. MCP::Client#dispatch_with_cancellation uses this yield to release the cancel-dispatch thread, which then issues a separate notifications/cancelled POST that may overlap with the original request on the network. The spec covers this: the sender has issued the request and still believes it in-progress, and receivers MAY ignore a cancellation referring to an unknown request id when the cancel POST happens to arrive first. https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/cancellation



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
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
 
# File 'lib/mcp/client/http.rb', line 372

def send_request(request:)
  method = request[:method] || request["method"]
  params = request[:params] || request["params"]
  oauth_retried = false
  step_up_retried = false

  begin
    # The response is consumed incrementally so that an SSE stream the server holds open
    # (or closes early per SEP-1699) can be handled; `initialize` streams are read to EOF
    # so the response object (and its `Mcp-Session-Id` header) is always available for capture.
    stream = SSEStream.new(
      abortable: method.to_s != MCP::Methods::INITIALIZE,
      max_message_bytes: @max_message_bytes,
      on_request: ->(message) { dispatch_server_request(message) },
    )

    yield if block_given?

    response = begin
      client.post("", request, session_headers.merge(request_metadata_headers(method, params))) do |req|
        req.options.on_data = stream.on_data
      end
    rescue StreamAbort
      nil
    end

    body = resolve_response_body(stream, response, method, params)

    capture_session_info(method, response, body) if response

    body
  rescue MessageTooLargeError => e
    raise RequestHandlerError.new(
      e.message,
      { method: method, params: params },
      error_type: :internal_error,
    )
  rescue Faraday::BadRequestError => e
    raise RequestHandlerError.new(
      "The #{method} request is invalid",
      { method: method, params: params },
      error_type: :bad_request,
      original_error: e,
    )
  rescue Faraday::UnauthorizedError => e
    # Run the OAuth flow at most once per `send_request` invocation.
    # The `oauth_retried` flag lives outside the `begin` so it survives `retry`,
    # ensuring a server returning 401 indefinitely raises rather than loops.
    if @oauth && !oauth_retried
      oauth_retried = true
      run_oauth_flow!(unauthorized_error: e)
      retry
    end

    raise RequestHandlerError.new(
      "You are unauthorized to make #{method} requests",
      { method: method, params: params },
      error_type: :unauthorized,
      original_error: e,
    )
  rescue Faraday::ForbiddenError => e
    # OAuth 2.0 step-up: a 403 carrying `error="insufficient_scope"` in
    # the Bearer challenge means the existing access token is valid
    # but lacks scopes the server now requires for this operation.
    # Re-run the full authorization flow with the escalated scope from
    # the challenge and retry once. A plain 403 without the challenge is
    # surfaced unchanged.
    if @oauth && !step_up_retried && insufficient_scope_challenge?(e)
      step_up_retried = true
      run_step_up_flow!(forbidden_error: e)

      retry
    end

    raise RequestHandlerError.new(
      "You are forbidden to make #{method} requests",
      { method: method, params: params },
      error_type: :forbidden,
      original_error: e,
    )
  rescue Faraday::ResourceNotFound => e
    # Per spec, 404 is the session-expired signal only when the request
    # actually carried an `Mcp-Session-Id`. A 404 without a session attached
    # (e.g. wrong URL or a stateless server) surfaces as a generic not-found.
    # https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#session-management
    if @session_id
      clear_session
      raise SessionExpiredError.new(
        "The #{method} request is not found",
        { method: method, params: params },
        original_error: e,
      )
    else
      raise RequestHandlerError.new(
        "The #{method} request is not found",
        { method: method, params: params },
        error_type: :not_found,
        original_error: e,
      )
    end
  rescue Faraday::UnprocessableEntityError => e
    raise RequestHandlerError.new(
      "The #{method} request is unprocessable",
      { method: method, params: params },
      error_type: :unprocessable_entity,
      original_error: e,
    )
  rescue Faraday::Error => e
    raise RequestHandlerError.new(
      "Internal error handling #{method} request",
      { method: method, params: params },
      error_type: :internal_error,
      original_error: e,
    )
  end
end