Module: MCPClient::ServerSSE::SseParser

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

Overview

Wire-level SSE parsing & dispatch ===

Instance Method Summary collapse

Instance Method Details

#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



149
150
151
152
153
154
155
156
157
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 149

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)



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

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: #{e.message}")
  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



12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 12

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



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 118

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



56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 56

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



82
83
84
85
86
87
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 82

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



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 92

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
    @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



72
73
74
75
76
77
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 72

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

Parameters:

  • data (String) —

    the endpoint event payload (absolute or relative URI)

Returns:

  • (String) —

    the absolute endpoint URL



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/mcp_client/server_sse/sse_parser.rb', line 162

def resolve_endpoint_uri(data)
  URI.join(@base_url, data).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}")
  message = "Invalid endpoint URI in SSE endpoint event: #{data.inspect} (#{e.message})"
  # The SSE worker thread swallows this exception with a generic rescue,
  # so also record the failure cause (mirroring @auth_error) for the
  # connect caller blocked in wait_for_connection to surface promptly.
  @mutex.synchronize do
    @connection_error = message
    @connection_established = false
    @connection_cv.broadcast
  end
  raise MCPClient::Errors::TransportError, message
end