Class: VectorMCP::Transport::HttpStream

Inherits:
Object
  • Object
show all
Defined in:
lib/vector_mcp/transport/http_stream.rb,
lib/vector_mcp/transport/http_stream/event_store.rb,
lib/vector_mcp/transport/http_stream/stream_handler.rb,
lib/vector_mcp/transport/http_stream/session_manager.rb

Overview

Implements the Model Context Protocol transport over HTTP with streaming support according to the MCP specification for Streamable HTTP transport.

This transport supports:

  • Client-to-server communication via HTTP POST
  • Optional server-to-client streaming via Server-Sent Events (SSE)
  • Session management with Mcp-Session-Id headers
  • Resumable connections with event IDs and Last-Event-ID support
  • Bidirectional communication patterns

Endpoints:

  • POST /mcp - Client sends JSON-RPC requests
  • GET /mcp - Optional SSE streaming for server-initiated messages
  • DELETE /mcp - Session termination

rubocop:disable Metrics/ClassLength

Examples:

Basic Usage

server = VectorMCP::Server.new("http-stream-server")
transport = VectorMCP::Transport::HttpStream.new(server, port: 8080)
server.run(transport: transport)

Defined Under Namespace

Classes: EventStore, SessionManager, StreamHandler

Constant Summary collapse

DEFAULT_HOST =

Default configuration values

"localhost"
DEFAULT_PORT = 
8000
DEFAULT_PATH_PREFIX = 
"/mcp"
DEFAULT_SESSION_TIMEOUT =

5 minutes

300
DEFAULT_EVENT_RETENTION =

Keep last 100 events for resumability

100
DEFAULT_REQUEST_TIMEOUT =

Default timeout for server-initiated requests

30
DEFAULT_MIN_THREADS = 
4
DEFAULT_MAX_THREADS = 
32
DEFAULT_MAX_BODY_BYTES =

Accommodates the 10 MiB ImageUtil limit after base64 expansion and JSON framing.

16 * 1024 * 1024
DEFAULT_ALLOWED_ORIGINS =

Default allowed origins — restrict to localhost by default for security.

%w[
  http://localhost
  https://localhost
  http://127.0.0.1
  https://127.0.0.1
  http://[::1]
  https://[::1]
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(server, options = {}) ⇒ HttpStream

Initializes a new HTTP Stream transport.

Parameters:

  • server (VectorMCP::Server) —

    The server instance that will handle messages

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

    Configuration options for the transport

Options Hash (options):

  • :host (String) — default: "localhost" —

    The hostname or IP to bind to

  • :port (Integer) — default: 8000 —

    The port to listen on

  • :path_prefix (String) — default: "/mcp" —

    The base path for HTTP endpoints

  • :session_timeout (Integer) — default: 300 —

    Session timeout in seconds

  • :event_retention (Integer) — default: 100 —

    Number of events to retain for resumability

  • :min_threads (Integer) — default: 4 —

    Minimum Puma thread pool size

  • :max_threads (Integer) — default: 32 —

    Maximum Puma thread pool size

  • :max_body_bytes (Integer) — default: 16777216 —

    Maximum POST body size in bytes

  • :allowed_origins (Array<String>) —

    Allowed origins for CORS validation. Defaults to localhost origins only. Pass ["*"] to allow all origins (NOT recommended for production).



85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 85

def initialize(server, options = {})
  @server = server
  @logger = server.logger
  initialize_configuration(options)
  initialize_components
  initialize_request_tracking
  initialize_object_pools
  initialize_server_state

  logger.info { "HttpStream transport initialized: #{@host}:#{@port}#{@path_prefix}" }
end

Instance Attribute Details

#event_store ⇒ HttpStream::EventStore (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Provides access to event store for internal components.

Returns:



250
251
252
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 250

def event_store
  @event_store
end

#host ⇒ String (readonly)

The hostname or IP address the server will bind to

Returns:

  • (String) —

    the current value of host



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def host
  @host
end

#logger ⇒ Logger (readonly)

The logger instance, shared with the server

Returns:

  • (Logger) —

    the current value of logger



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def logger
  @logger
end

#max_body_bytes ⇒ Integer (readonly)

Maximum accepted POST body size

Returns:

  • (Integer) —

    the current value of max_body_bytes



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def max_body_bytes
  @max_body_bytes
end

#path_prefix ⇒ String (readonly)

The base URL path for MCP endpoints

Returns:

  • (String) —

    the current value of path_prefix



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def path_prefix
  @path_prefix
end

#port ⇒ Integer (readonly)

The port number the server will listen on

Returns:

  • (Integer) —

    the current value of port



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def port
  @port
end

#server ⇒ VectorMCP::Server (readonly)

The server instance this transport is bound to

Returns:



46
47
48
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 46

def server
  @server
end

#session_manager ⇒ HttpStream::SessionManager (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Provides access to session manager for internal components.

Returns:



244
245
246
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 244

def session_manager
  @session_manager
end

#stream_handler ⇒ HttpStream::StreamHandler (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Provides access to stream handler for internal components.

Returns:



256
257
258
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 256

def stream_handler
  @stream_handler
end

Instance Method Details

#call(env) ⇒ Array(Integer, Hash, Object)

Handles incoming HTTP requests (Rack interface). Routes requests to appropriate handlers based on path and method.

Parameters:

  • env (Hash) —

    The Rack environment hash

Returns:

  • (Array(Integer, Hash, Object)) —

    Standard Rack response triplet



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
141
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 113

def call(env)
  start_time = Time.now
  path = env["PATH_INFO"]
  method = env["REQUEST_METHOD"]
  transport_context = build_transport_context(env, method, path, start_time)

  logger.debug { "Processing HTTP request #{method} #{path}" }

  transport_context = execute_transport_hooks(:before_request, transport_context)
  raise transport_context.error if transport_context.error?
  return transport_context.result if transport_context.result

  response = route_request(path, method, env)
  transport_context.result = response
  transport_context = execute_transport_hooks(:after_response, transport_context)
  raise transport_context.error if transport_context.error?

  response = transport_context.result || response
  log_request_completion(method, path, start_time, response[0])
  response
rescue StandardError => e
  if transport_context
    transport_context.error = e
    transport_context = execute_transport_hooks(:on_transport_error, transport_context)
    return transport_context.result if transport_context.result
  end

  handle_request_error(method, path, e)
end

#run ⇒ void

This method returns an undefined value.

Starts the HTTP Stream transport. This method will block until the server is stopped.

Raises:

  • (StandardError) —

    if there's a fatal error during server startup



102
103
104
105
106
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 102

def run
  start_puma_server
rescue StandardError => e
  handle_fatal_error(e)
end

#send_notification(method, params = nil) ⇒ Boolean

Sends a notification to the first available session.

Parameters:

  • method (String) —

    The notification method name

  • params (Hash, Array, nil) (defaults to: nil) —

    The notification parameters

Returns:

  • (Boolean) —

    True if notification was sent successfully



148
149
150
151
152
153
154
155
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 148

def send_notification(method, params = nil)
  # Find the first available session
  first_session = find_first_session
  return false unless first_session

  message = build_notification(method, params)
  @stream_handler.send_message_to_session(first_session, message)
end

#send_notification_to_session(session_id, method, params = nil) ⇒ Boolean

Sends a notification to a specific session.

Parameters:

  • session_id (String) —

    The target session ID

  • method (String) —

    The notification method name

  • params (Hash, Array, nil) (defaults to: nil) —

    The notification parameters

Returns:

  • (Boolean) —

    True if notification was sent successfully



163
164
165
166
167
168
169
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 163

def send_notification_to_session(session_id, method, params = nil)
  session = @session_manager.get_session(session_id)
  return false unless session

  message = build_notification(method, params)
  @stream_handler.send_message_to_session(session, message)
end

#send_request(method, params = nil, timeout: DEFAULT_REQUEST_TIMEOUT) ⇒ Object

Sends a server-initiated JSON-RPC request compatible with Session expectations. This method will block until a response is received or the timeout is reached. For HTTP transport, this requires finding an appropriate session with streaming connection.

Parameters:

  • method (String) —

    The request method name

  • params (Hash, Array, nil) (defaults to: nil) —

    The request parameters

  • timeout (Numeric) (defaults to: DEFAULT_REQUEST_TIMEOUT) —

    How long to wait for a response, in seconds

Returns:

  • (Object) —

    The result part of the client's response

Raises:



181
182
183
184
185
186
187
188
189
190
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 181

def send_request(method, params = nil, timeout: DEFAULT_REQUEST_TIMEOUT)
  raise ArgumentError, "Method cannot be blank" if method.to_s.strip.empty?

  # Find the first session with streaming connection
  # In HTTP transport, we need an active streaming connection to send server-initiated requests
  streaming_session = find_streaming_session
  raise ArgumentError, "No streaming session available for server-initiated requests" unless streaming_session

  send_request_to_session(streaming_session.id, method, params, timeout: timeout)
end

#send_request_to_session(session_id, method, params = nil, timeout: DEFAULT_REQUEST_TIMEOUT) ⇒ Object

Sends a server-initiated JSON-RPC request to a specific session and waits for a response. This method will block until a response is received or the timeout is reached.

Parameters:

  • session_id (String) —

    The target session ID

  • method (String) —

    The request method name

  • params (Hash, Array, nil) (defaults to: nil) —

    The request parameters

  • timeout (Numeric) (defaults to: DEFAULT_REQUEST_TIMEOUT) —

    How long to wait for a response, in seconds

Returns:

  • (Object) —

    The result part of the client's response

Raises:



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
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 202

def send_request_to_session(session_id, method, params = nil, timeout: DEFAULT_REQUEST_TIMEOUT)
  raise ArgumentError, "Method cannot be blank" if method.to_s.strip.empty?
  raise ArgumentError, "Session ID cannot be blank" if session_id.to_s.strip.empty?

  session = @session_manager.get_session(session_id)
  raise ArgumentError, "Session not found: #{session_id}" unless session

  raise ArgumentError, "Session must have streaming connection for server-initiated requests" unless session.streaming?

  request_id = generate_request_id
  request_payload = { jsonrpc: "2.0", id: request_id, method: method }
  request_payload[:params] = params if params

  setup_request_tracking(request_id)
  # Sending request to session

  # Send request via existing streaming connection
  unless @stream_handler.send_message_to_session(session, request_payload)
    cleanup_request_tracking(request_id)
    raise VectorMCP::SamplingError, "Failed to send request to session #{session_id}"
  end

  response = wait_for_response(request_id, method, timeout)
  process_response(response, request_id, method)
end

#stop ⇒ void

This method returns an undefined value.

Stops the transport and cleans up resources.



231
232
233
234
235
236
237
238
 
# File 'lib/vector_mcp/transport/http_stream.rb', line 231

def stop
  logger.info { "Stopping HttpStream transport" }
  @running = false
  cleanup_all_pending_requests
  @session_manager.cleanup_all_sessions
  @puma_server&.stop
  logger.info { "HttpStream transport stopped" }
end