Class: RubyLLM::MCP::Native::Transports::SSE

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

Instance Attribute Summary collapse

Instance Method Summary collapse

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

#with_timeout

Constructor Details

#initialize(url:, coordinator:, request_timeout:, options: {}) ⇒ SSE

Returns a new instance of SSE.



12
13
14
15
16
17
18
19
20
21
22
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
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 12

def initialize(url:, coordinator:, request_timeout:, options: {})
  @event_url = url
  @messages_url = nil
  @coordinator = coordinator
  @request_timeout = request_timeout

  # Extract options
  extracted_options = options.dup
  @version = extracted_options.delete(:version) || :http2
  headers = extracted_options.delete(:headers) || {}
  oauth_provider = extracted_options.delete(:oauth_provider)

  uri = URI.parse(url)
  @root_url = "#{uri.scheme}://#{uri.host}"
  @root_url += ":#{uri.port}" if uri.port != uri.default_port

  @client_id = SecureRandom.uuid
  @headers = headers.merge({
                             "Accept" => "text/event-stream",
                             "Content-Type" => "application/json",
                             "Cache-Control" => "no-cache",
                             "X-CLIENT-ID" => @client_id
                           })

  @oauth_provider = oauth_provider
  @resource_metadata_url = nil
  @auth_retry_attempted = false

  @id_counter = 0
  @id_mutex = Mutex.new
  @pending_requests = {}
  @pending_mutex = Mutex.new
  @connection_mutex = Mutex.new
  @state_mutex = Mutex.new
  @running = false
  @sse_thread = nil
  @sse_response = nil

  RubyLLM::MCP.logger.info "Initializing SSE transport to #{@event_url} with client ID #{@client_id}"
end

Instance Attribute Details

#coordinator ⇒ Object (readonly)

Returns the value of attribute coordinator.



10
11
12
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 10

def coordinator
  @coordinator
end

#headers ⇒ Object (readonly)

Returns the value of attribute headers.



10
11
12
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 10

def headers
  @headers
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



10
11
12
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 10

def id
  @id
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


99
100
101
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 99

def alive?
  running?
end

#close ⇒ Object 



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
150
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 117

def close
  should_close = @state_mutex.synchronize do
    return unless @running

    @running = false
    true
  end

  return unless should_close

  RubyLLM::MCP.logger.info "Closing SSE transport connection"

  # Close the SSE response stream if it exists
  begin
    @sse_response&.body&.close
  rescue StandardError => e
    RubyLLM::MCP.logger.debug "Error closing SSE response: #{e.message}"
  end

  # Wait for the thread to finish (but don't join from within itself)
  if @sse_thread && Thread.current != @sse_thread
    @sse_thread.join(1)
  end
  @sse_thread = nil

  fail_pending_requests!(
    Errors::TransportError.new(
      message: "SSE transport closed",
      code: nil
    )
  )

  @messages_url = nil
end

#request(body, wait_for_response: true) ⇒ Object

rubocop:disable Metrics/MethodLength



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
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 53

def request(body, wait_for_response: true) # rubocop:disable Metrics/MethodLength
  request_id = body.is_a?(Hash) ? (body["id"] || body[:id]) : nil

  if wait_for_response && request_id.nil?
    raise ArgumentError, "Request ID must be provided in message body when wait_for_response is true"
  end

  response_queue = nil
  if wait_for_response
    response_queue = Queue.new
    @pending_mutex.synchronize do
      @pending_requests[request_id.to_s] = response_queue
    end
  end

  begin
    send_request(body, request_id)
  rescue Errors::TransportError, Errors::TimeoutError => e
    if wait_for_response && request_id
      @pending_mutex.synchronize { @pending_requests.delete(request_id.to_s) }
    end
    RubyLLM::MCP.logger.error "Request error (ID: #{request_id}): #{e.message}"
    raise e
  end

  return unless wait_for_response

  result = nil
  begin
    result = with_timeout(@request_timeout / 1000, request_id: request_id) do
      response_queue.pop
    end
  rescue Errors::TimeoutError => e
    if request_id
      @pending_mutex.synchronize { @pending_requests.delete(request_id.to_s) }
    end
    RubyLLM::MCP.logger.error "SSE request timeout (ID: #{request_id}) \
      after #{@request_timeout / 1000} seconds."
    raise e
  end

  raise result if result.is_a?(Errors::TransportError)

  result
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


103
104
105
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 103

def running?
  @state_mutex.synchronize { @running }
end

#set_protocol_version(version) ⇒ Object 



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

def set_protocol_version(version)
  @protocol_version = version
end

#start ⇒ Object 



107
108
109
110
111
112
113
114
115
 
# File 'lib/ruby_llm/mcp/native/transports/sse.rb', line 107

def start
  @state_mutex.synchronize do
    return if @running

    @running = true
  end

  start_sse_listener
end