Module: MCPClient::ServerStdio::JsonRpcTransport

Includes:
JsonRpcCommon
Included in:
MCPClient::ServerStdio
Defined in:
lib/mcp_client/server_stdio/json_rpc_transport.rb

Overview

JSON-RPC request/notification plumbing for stdio transport

Instance Method Summary collapse

Methods included from JsonRpcCommon

#build_jsonrpc_notification, #build_jsonrpc_request, #build_named_request_params, #cancellable_request?, #client_capabilities, #client_info_payload, #declare_sampling_tools, #initialization_params, #ping, #process_jsonrpc_response, #registered_callback?, #sampling_tools_supported?, #split_request_meta, #validate_protocol_version!, #with_retry

Instance Method Details

#call_tool_streaming(tool_name, parameters) ⇒ Enumerator

Stream tool call fallback for stdio transport (yields single result)

Parameters:

  • tool_name (String) —

    the name of the tool to call

  • parameters (Hash) —

    the parameters to pass to the tool

Returns:

  • (Enumerator) —

    a stream containing a single result



106
107
108
109
110
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 106

def call_tool_streaming(tool_name, parameters)
  Enumerator.new do |yielder|
    yielder << call_tool(tool_name, parameters)
  end
end

#ensure_initialized ⇒ void

This method returns an undefined value.

Ensure the server process is started and initialized (handshake)

Raises:



14
15
16
17
18
19
20
21
22
23
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 14

def ensure_initialized
  return if @initialized

  connect
  start_reader
  start_stderr_reader
  perform_initialize

  @initialized = true
end

#next_id ⇒ Integer

Generate a new unique request ID and mark it as awaiting a response. Registering the id before the request is sent lets the reader thread distinguish expected responses from late/unsolicited ones.

Returns:

  • (Integer) —

    a unique request ID



55
56
57
58
59
60
61
62
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 55

def next_id
  @mutex.synchronize do
    id = @next_id
    @next_id += 1
    @awaiting[id] = true
    id
  end
end

#perform_initialize ⇒ void

This method returns an undefined value.

Handshake: send initialize request and initialized notification

Raises:



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

def perform_initialize
  # Initialize request
  init_id = next_id
  init_req = build_jsonrpc_request('initialize', initialization_params, init_id)
  send_request(init_req)
  res = wait_response(init_id)
  if (err = res['error'])
    raise MCPClient::Errors::ConnectionError, "Initialize failed: #{err['message']}"
  end

  # Store negotiated protocol version, server info and capabilities.
  # Disconnects if the server negotiated a version we cannot speak.
  result = res['result'] || {}
  @protocol_version = validate_protocol_version!(result)
  @server_info = result['serverInfo']
  @capabilities = result['capabilities']
  @instructions = result['instructions']

  # Send initialized notification
  notif = build_jsonrpc_notification('notifications/initialized', {})
  @stdin.puts(notif.to_json)
end

#rpc_notify(method, params = {}) ⇒ void

This method returns an undefined value.

Send a JSON-RPC notification (no response expected)

Parameters:

  • method (String) —

    JSON-RPC method

  • params (Hash) (defaults to: {}) —

    parameters for the notification



154
155
156
157
158
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 154

def rpc_notify(method, params = {})
  ensure_initialized
  notif = build_jsonrpc_notification(method, params)
  @stdin.puts(notif.to_json)
end

#rpc_request(method, params = {}, timeout: nil) ⇒ Object

Generic JSON-RPC request: send method with params and wait for result

Parameters:

  • method (String) —

    JSON-RPC method

  • params (Hash) (defaults to: {}) —

    parameters for the request

Returns:

  • (Object) —

    result from JSON-RPC response

Raises:



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 119

def rpc_request(method, params = {}, timeout: nil)
  ensure_initialized
  with_retry do
    req_id = next_id
    req = build_jsonrpc_request(method, params, req_id)
    send_request(req)
    begin
      res = wait_response(req_id, timeout: timeout)
    rescue MCPClient::Errors::RequestTimeoutError
      # MCP lifecycle: on timeout the sender SHOULD issue a cancellation
      # notification for the abandoned request and stop waiting.
      send_cancellation_notification(req_id) if cancellable_request?(method, params)
      raise
    end
    process_jsonrpc_response(res)
  end
end

#send_cancellation_notification(request_id) ⇒ void

This method returns an undefined value.

Best-effort notifications/cancelled for a request the client stopped waiting on. Failures are swallowed: the transport may be the reason the request timed out in the first place.

Parameters:

  • request_id (Integer) —

    id of the abandoned request



142
143
144
145
146
147
148
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 142

def send_cancellation_notification(request_id)
  notif = build_jsonrpc_notification('notifications/cancelled',
                                     { 'requestId' => request_id, 'reason' => 'Request timed out' })
  @stdin.puts(notif.to_json)
rescue StandardError => e
  @logger.debug("Failed to send cancellation notification: #{e.message}")
end

#send_request(req) ⇒ void

This method returns an undefined value.

Send a JSON-RPC request and return nothing

Parameters:

  • req (Hash) —

    the JSON-RPC request

Raises:



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 68

def send_request(req)
  @logger.debug("Sending JSONRPC request: #{req.to_json}")
  @stdin.puts(req.to_json)
rescue StandardError => e
  # A request that failed to send will never receive a response, so drop
  # its awaiting marker; otherwise a broken transport (e.g. the server
  # exited) would leak an entry per retry/attempt into @awaiting.
  @mutex.synchronize { @awaiting.delete(req['id']) } if req.is_a?(Hash) && req['id']
  raise MCPClient::Errors::TransportError, "Failed to send JSONRPC request: #{e.message}"
end

#wait_response(id, timeout: nil) ⇒ Hash

Wait for a response with the given request ID

Parameters:

  • id (Integer) —

    the request ID

Returns:

  • (Hash) —

    the JSON-RPC response message

Raises:



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 83

def wait_response(id, timeout: nil)
  deadline = Time.now + (timeout || @read_timeout)
  @mutex.synchronize do
    until @pending.key?(id)
      remaining = deadline - Time.now
      break if remaining <= 0

      @cond.wait(@mutex, remaining)
    end
    # Remove the response and the awaiting marker on both success and
    # timeout so neither @pending nor @awaiting accumulates entries.
    msg = @pending.delete(id)
    @awaiting.delete(id)
    raise MCPClient::Errors::RequestTimeoutError, "Timeout waiting for JSONRPC response id=#{id}" unless msg

    msg
  end
end