Class: ModelContextProtocol::Server::StreamableHttpTransport

Inherits:
Object
  • Object
show all
Defined in:
lib/model_context_protocol/server/streamable_http_transport.rb,
lib/model_context_protocol/server/streamable_http_transport/event_counter.rb,
lib/model_context_protocol/server/streamable_http_transport/request_store.rb,
lib/model_context_protocol/server/streamable_http_transport/session_store.rb,
lib/model_context_protocol/server/streamable_http_transport/message_poller.rb,
lib/model_context_protocol/server/streamable_http_transport/stream_registry.rb,
lib/model_context_protocol/server/streamable_http_transport/notification_queue.rb,
lib/model_context_protocol/server/streamable_http_transport/server_request_store.rb,
lib/model_context_protocol/server/streamable_http_transport/session_message_queue.rb

Defined Under Namespace

Classes: ErrorResponse, EventCounter, MessagePoller, NotificationQueue, RequestStore, Response, ServerRequestStore, SessionMessageQueue, SessionStore, StreamRegistry

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(router:, configuration:) ⇒ StreamableHttpTransport

Initialize the HTTP transport with Redis-backed cross-server communication support Sets up background threads for message polling and stream monitoring in multi-server deployments



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/model_context_protocol/server/streamable_http_transport.rb', line 23

def initialize(router:, configuration:)
  @router = router
  @configuration = configuration
  @client_logger = configuration.client_logger
  @server_logger = configuration.server_logger

  @redis_pool = ModelContextProtocol::Server::RedisConfig.pool
  @redis = ModelContextProtocol::Server::RedisClientProxy.new(@redis_pool)

  @require_sessions = @configuration.require_sessions
  # Use Concurrent::Map for thread-safe access from multiple request threads
  @session_protocol_versions = Concurrent::Map.new
  @validate_origin = @configuration.validate_origin
  @allowed_origins = @configuration.allowed_origins

  @session_store = SessionStore.new(@redis, ttl: @configuration.session_ttl)
  @server_instance = "#{Socket.gethostname}-#{Process.pid}-#{SecureRandom.hex(4)}"
  @stream_registry = StreamRegistry.new(@redis, @server_instance)
  @notification_queue = NotificationQueue.new(@redis, @server_instance)
  @event_counter = EventCounter.new(@redis, @server_instance)
  @request_store = RequestStore.new(@redis, @server_instance)
  @server_request_store = ServerRequestStore.new(@redis, @server_instance)
  @ping_timeout = @configuration.ping_timeout

  @message_poller = MessagePoller.new(@redis, @stream_registry, @client_logger) do |stream, message|
    send_to_stream(stream, message)
  end
  @message_poller.start

  @stream_monitor_running = false
  @stream_monitor_thread = nil
  start_stream_monitor
end

Instance Attribute Details

#server_logger ⇒ Object (readonly)

Returns the value of attribute server_logger.



19
20
21
 
# File 'lib/model_context_protocol/server/streamable_http_transport.rb', line 19

def server_logger
  @server_logger
end

Instance Method Details

#handle(env:, session_context: {}) ⇒ Object

Main entry point for handling HTTP requests (POST, GET, DELETE) Routes requests to appropriate handlers and manages the request/response lifecycle

Parameters:

  • Rack environment hash (required)

  • (defaults to: {})

    Per-request context that will be merged with server context



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/model_context_protocol/server/streamable_http_transport.rb', line 89

def handle(env:, session_context: {})
  @server_logger.debug("Handling streamable HTTP transport request")

  case env["REQUEST_METHOD"]
  when "POST"
    @server_logger.debug("Handling POST request")
    handle_post_request(env, session_context: session_context)
  when "GET"
    @server_logger.debug("Handling GET request")
    handle_get_request(env)
  when "DELETE"
    @server_logger.debug("Handling DELETE request")
    handle_delete_request(env)
  else
    error_response = ErrorResponse[id: nil, error: {code: -32601, message: "Method not allowed"}]
    {json: error_response.serialized, status: 405}
  end
end

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

Send real-time notifications to active SSE streams or queue for delivery Used for progress updates, resource changes, and other server-initiated messages

Parameters:

  • the notification method name

  • the notification parameters

  • (defaults to: nil)

    optional session ID for targeted delivery



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
142
143
144
145
146
147
148
149
 
# File 'lib/model_context_protocol/server/streamable_http_transport.rb', line 113

def send_notification(method, params, session_id: nil)
  notification = {
    jsonrpc: "2.0",
    method: method,
    params: params
  }

  log_to_server_with_context do |logger|
    logger.info("← #{method} [outgoing]")
    logger.info("  Notification: #{notification.to_json}")
  end

  if session_id
    # Deliver to specific session/stream
    @server_logger.debug("Attempting targeted delivery to session: #{session_id}")
    if deliver_to_session_stream(session_id, notification)
      @server_logger.debug("Successfully delivered notification to specific stream: #{session_id}")
    else
      @server_logger.debug("Failed to deliver to specific stream #{session_id}, queuing notification: #{method}")
      @notification_queue.push(notification)
    end
  elsif @stream_registry.get_local_stream(nil) # Check for persistent notification stream (no-session)
    @server_logger.debug("No session_id provided, delivering notification to persistent notification stream")
    if deliver_to_session_stream(nil, notification)
      @server_logger.debug("Successfully delivered notification to persistent notification stream")
    else
      @server_logger.debug("Failed to deliver to persistent notification stream, queuing notification: #{method}")
      @notification_queue.push(notification)
    end
  elsif @stream_registry.has_any_local_streams?
    @server_logger.debug("No persistent notification stream, delivering notification to active streams")
    deliver_to_active_streams(notification)
  else
    @server_logger.debug("No active streams, queuing notification: #{method}")
    @notification_queue.push(notification)
  end
end

#shutdown ⇒ Object

Gracefully shut down the transport by stopping background threads and cleaning up resources Closes all active streams. Redis entries are left to expire naturally (they have TTLs). This method is signal-safe and avoids mutex operations.



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/model_context_protocol/server/streamable_http_transport.rb', line 60

def shutdown
  @server_logger.info("Shutting down StreamableHttpTransport")

  @message_poller&.stop

  @stream_monitor_running = false
  if @stream_monitor_thread&.alive?
    @stream_monitor_thread.kill
    @stream_monitor_thread.join(5)
  end

  # Close streams directly without Redis cleanup (signal-safe).
  # Redis entries will expire naturally via TTL.
  @stream_registry.get_all_local_streams.each do |session_id, stream|
    begin
      stream.close
    rescue IOError, Errno::EPIPE, Errno::ECONNRESET, Errno::ENOTCONN, Errno::EBADF
      # Stream already closed, ignore
    end
    @server_logger.info("← SSE stream [closed] (#{session_id}) [shutdown]")
  end

  @server_logger.info("StreamableHttpTransport shutdown complete")
end