Class: RubyLLM::MCP::Native::Transports::StreamableHTTP

Inherits:
Object
  • Object
show all
Includes:
RubyLLM::MCP::Native::Transports::Support::Timeout
Defined in:
lib/ruby_llm/mcp/native/transports/streamable_http.rb

Overview

Main StreamableHTTP transport class

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from RubyLLM::MCP::Native::Transports::Support::Timeout

#with_timeout

Constructor Details

#initialize(url:, request_timeout:, coordinator:, headers: {}, reconnection: {}, version: :http2, oauth_provider: nil, rate_limit: nil, reconnection_options: nil, session_id: nil, sse_timeout: nil, options: {}) ⇒ StreamableHTTP

rubocop:disable Metrics/MethodLength, Metrics/ParameterLists



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 43

def initialize( # rubocop:disable Metrics/MethodLength, Metrics/ParameterLists
  url:,
  request_timeout:,
  coordinator:,
  headers: {},
  reconnection: {},
  version: :http2,
  oauth_provider: nil,
  rate_limit: nil,
  reconnection_options: nil,
  session_id: nil,
  sse_timeout: nil,
  options: {}
)
  # Extract options if provided (for backward compatibility)
  extracted_options = options.dup
  headers = extracted_options.delete(:headers) || headers
  version = extracted_options.delete(:version) || version
  oauth_provider = extracted_options.delete(:oauth_provider) || oauth_provider
  reconnection = extracted_options.delete(:reconnection) || reconnection
  reconnection_options = extracted_options.delete(:reconnection_options) || reconnection_options
  rate_limit = extracted_options.delete(:rate_limit) || rate_limit
  session_id = extracted_options.delete(:session_id) || session_id
  sse_timeout = extracted_options.delete(:sse_timeout) || sse_timeout

  @url = URI(url)
  @coordinator = coordinator
  @request_timeout = request_timeout
  @sse_timeout = sse_timeout
  @headers = headers || {}
  @session_id = session_id
  @sse_fallback_supported = true

  @version = version
  @protocol_version = nil

  @client_id = SecureRandom.uuid

  # Reconnection options precedence: explicit > hash > defaults
  @reconnection_options = if reconnection_options
                            reconnection_options
                          elsif reconnection && !reconnection.empty?
                            ReconnectionOptions.new(**reconnection)
                          else
                            ReconnectionOptions.new
                          end

  @oauth_provider = oauth_provider
  @rate_limiter = Support::RateLimiter.new(**rate_limit) if rate_limit

  @id_counter = 0
  @id_mutex = Mutex.new
  @pending_requests = {}
  @pending_mutex = Mutex.new
  @running = true
  @sse_stopped = false
  @state_mutex = Mutex.new
  @sse_thread = nil
  @sse_mutex = Mutex.new
  @last_sse_event_id = nil

  # Track if we've attempted auth flow to prevent infinite loops
  @auth_retry_attempted = false

  # Thread-safe collection of all HTTPX clients
  @clients = []
  @clients_mutex = Mutex.new

  @connection = create_connection
end

Instance Attribute Details

#coordinator ⇒ Object (readonly)

Returns the value of attribute coordinator.



41
42
43
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 41

def coordinator
  @coordinator
end

#oauth_provider ⇒ Object (readonly)

Returns the value of attribute oauth_provider.



41
42
43
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 41

def oauth_provider
  @oauth_provider
end

#protocol_version ⇒ Object (readonly)

Returns the value of attribute protocol_version.



41
42
43
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 41

def protocol_version
  @protocol_version
end

#session_id ⇒ Object (readonly)

Returns the value of attribute session_id.



41
42
43
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 41

def session_id
  @session_id
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 132

def alive?
  running?
end

#close ⇒ Object 



136
137
138
139
140
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 136

def close
  terminate_session
  cleanup_sse_resources
  cleanup_connection
end

#on_close(&block) ⇒ Object 



160
161
162
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 160

def on_close(&block)
  @on_close_callback = block
end

#on_error(&block) ⇒ Object 



156
157
158
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 156

def on_error(&block)
  @on_error_callback = block
end

#on_message(&block) ⇒ Object 



152
153
154
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 152

def on_message(&block)
  @on_message_callback = block
end

#request(body, wait_for_response: true) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 114

def request(body, wait_for_response: true)
  if @rate_limiter&.exceeded?
    sleep(1) while @rate_limiter&.exceeded?
  end
  @rate_limiter&.add

  # Extract the request ID from the body (if present)
  request_id = body.is_a?(Hash) ? (body["id"] || body[:id]) : nil

  response_queue = setup_response_queue(request_id, wait_for_response)
  result = send_http_request(body, request_id)
  return result if result.is_a?(RubyLLM::MCP::Result)

  if wait_for_response && request_id
    wait_for_response_with_timeout(request_id.to_s, response_queue)
  end
end

#set_protocol_version(version) ⇒ Object 



148
149
150
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 148

def set_protocol_version(version)
  @protocol_version = version
end

#start ⇒ Object 



142
143
144
145
146
 
# File 'lib/ruby_llm/mcp/native/transports/streamable_http.rb', line 142

def start
  @state_mutex.synchronize do
    @sse_stopped = false
  end
end