Module: MCPClient::ServerSSE::SseParser

Includes:
OriginPolicy
Included in:
MCPClient::ServerSSE
Defined in:
lib/mcp_client/server_sse/sse_parser.rb

Overview

Wire-level SSE parsing & dispatch ===

Instance Method Summary collapse

Methods included from OriginPolicy

#origin_of, #reject_cross_origin_redirect!, #same_origin?

Instance Method Details

#fail_endpoint_handshake!(message) ⇒ Object

Record the handshake failure cause and raise. The SSE worker thread swallows this exception with a generic rescue, so also record the failure (mirroring @auth_error) for the connect caller blocked in wait_for_connection to surface promptly.

Parameters:

  • message (String) —

    the failure description

Raises:



203
204
205
206
207
208
209
210
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 203

def fail_endpoint_handshake!(message)
  @mutex.synchronize do
    @connection_error = message
    @connection_established = false
    @connection_cv.broadcast
  end
  raise MCPClient::Errors::TransportError, message
end

#handle_endpoint_event(data) ⇒ Object

Handle the special "endpoint" control frame (for SSE handshake). The event data is a URI reference (MCP 2024-11-05 HTTP with SSE: the server sends "an endpoint event containing a URI for the client to use for sending messages") which must be resolved against the SSE connection URL per RFC 3986 section 5.1.3, so relative endpoint URIs POST to the URL the server actually designated.

Parameters:

  • data (String) —

    the raw endpoint payload



161
162
163
164
165
166
167
168
169
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 161

def handle_endpoint_event(data)
  endpoint = resolve_endpoint_uri(data)
  @mutex.synchronize do
    @rpc_endpoint = endpoint
    @sse_connected = true
    @connection_established = true
    @connection_cv.broadcast
  end
end

#handle_message_event(event) ⇒ Object

Handle a "message" SSE event (payload is JSON-RPC over SSE)

Parameters:

  • event (Hash) —

    the parsed SSE event (with :data, :id, etc)



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 31

def handle_message_event(event)
  return if event[:data].empty?

  begin
    data = JSON.parse(event[:data])

    return if process_error_in_message?(data)
    return if process_server_request?(data)
    return if process_notification?(data)

    process_response?(data)
  rescue MCPClient::Errors::ConnectionError
    raise
  rescue JSON::ParserError => e
    @logger.warn("Failed to parse JSON from event data: #{describe_parse_error(e, event[:data])}")
  rescue StandardError => e
    @logger.error("Error processing SSE event: #{e.message}")
  end
end

#parse_and_handle_sse_event(event_data) ⇒ Object

Parse and handle a raw SSE event payload.

Parameters:

  • event_data (String) —

    the raw event chunk



15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 15

def parse_and_handle_sse_event(event_data)
  event = parse_sse_event(event_data)
  return if event.nil?

  case event[:event]
  when 'endpoint'
    handle_endpoint_event(event[:data])
  when 'ping'
    # no-op
  when 'message'
    handle_message_event(event)
  end
end

#parse_sse_event(event_data) ⇒ Hash?

Parse a raw SSE chunk into its :event, :data, :id fields

Parameters:

  • event_data (String) —

    the raw SSE block

Returns:

  • (Hash, nil) —

    parsed fields or nil if it was pure comment/blank



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 130

def parse_sse_event(event_data)
  event       = { event: 'message', data: '', id: nil }
  data_lines  = []
  has_content = false

  event_data.each_line do |line|
    line = line.chomp
    next if line.empty? # blank line
    next if line.start_with?(':') # SSE comment

    has_content = true
    if line.start_with?('event:')
      event[:event] = line[6..].strip
    elsif line.start_with?('data:')
      data_lines << line[5..].strip
    elsif line.start_with?('id:')
      event[:id] = line[3..].strip
    end
  end

  event[:data] = data_lines.join("\n")
  has_content ? event : nil
end

#process_error_in_message?(data) ⇒ Boolean

Process a connection-level JSON-RPC error payload in the SSE stream. Error RESPONSES (id-bearing) belong to a pending request and are delivered to the waiting caller via process_response? instead, per the MCP lifecycle "Error Handling" section (implementations SHOULD handle error cases such as protocol version mismatch), so they must not be swallowed here.

Parameters:

  • data (Hash) —

    the parsed JSON payload

Returns:

  • (Boolean) —

    true if we saw & handled an id-less error



59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 59

def process_error_in_message?(data)
  return false unless data['error']
  return false if data['id']

  error_message = data['error']['message'] || 'Unknown server error'
  error_code    = data['error']['code']

  handle_sse_auth_error_message(error_message) if authorization_error?(error_message, error_code)

  @logger.error("Server error: #{error_message}")
  true
end

#process_notification?(data) ⇒ Boolean

Process a JSON-RPC notification (no id => notification)

Parameters:

  • data (Hash) —

    the parsed JSON payload

Returns:

  • (Boolean) —

    true if we saw & handled a notification



85
86
87
88
89
90
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 85

def process_notification?(data)
  return false unless data['method'] && !data.key?('id')

  @notification_callback&.call(data['method'], data['params'])
  true
end

#process_response?(data) ⇒ Boolean

Process a JSON-RPC response (id => response)

Parameters:

  • data (Hash) —

    the parsed JSON payload

Returns:

  • (Boolean) —

    true if we saw & handled a response



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 95

def process_response?(data)
  return false unless data['id']

  # Deliver the response to the waiting caller via @sse_results only.
  # We intentionally do NOT write @tools_data here: request_tools_list is
  # the sole writer of that cache and sets it to the COMPLETE, fully
  # paginated list. Writing each page as it arrives would let a concurrent
  # list_tools observe a partial (page-1-only) cache mid-pagination.
  @mutex.synchronize do
    # The stream is peer-controlled: only ids some caller is actually
    # waiting on are stored. Without this check a server could stream
    # unsolicited responses with fresh ids and grow @sse_results without
    # bound for the lifetime of the client.
    unless @pending_request_ids.include?(data['id'])
      @logger.debug("Discarding unsolicited response id #{data['id'].inspect}")
      return true
    end

    @sse_results[data['id']] =
      if data['error']
        # JSON-RPC error response: store the error under a Symbol key
        # (JSON.parse only produces String keys, so this cannot collide
        # with a success result) for the waiter to raise ServerError.
        { error: data['error'] }
      else
        data['result']
      end
  end

  true
end

#process_server_request?(data) ⇒ Boolean

Process a JSON-RPC request from server (has both id AND method)

Parameters:

  • data (Hash) —

    the parsed JSON payload

Returns:

  • (Boolean) —

    true if we saw & handled a server request



75
76
77
78
79
80
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 75

def process_server_request?(data)
  return false unless data['method'] && data.key?('id')

  handle_server_request(data)
  true
end

#resolve_endpoint_uri(data) ⇒ String

Resolve an endpoint URI reference against the SSE connection URL. The resolved endpoint MUST stay on the SSE connection's origin: the event payload is server-controlled input, and honoring a cross-origin target would redirect every JSON-RPC POST — including the configured Authorization/API-key headers and callback response bodies — to a server the caller never chose.

Parameters:

  • data (String) —

    the endpoint event payload (absolute or relative URI)

Returns:

  • (String) —

    the absolute endpoint URL



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 179

def resolve_endpoint_uri(data)
  endpoint = URI.join(@base_url, data)
  base = URI.parse(@base_url)
  unless same_origin?(base, endpoint)
    fail_endpoint_handshake!(
      "Cross-origin endpoint in SSE endpoint event: #{data.inspect} " \
      "does not match the connection origin #{origin_of(base)}"
    )
  end
  endpoint.to_s
rescue URI::Error => e
  # The endpoint event is the handshake's core payload; an unresolvable
  # URI must fail the handshake rather than deferring a broken POST
  # target to the first request.
  @logger.error("Failed to resolve endpoint URI #{data.inspect} against #{@base_url}: #{e.message}")
  fail_endpoint_handshake!("Invalid endpoint URI in SSE endpoint event: #{data.inspect} (#{e.message})")
end