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, #initialization_params, #ping, #process_jsonrpc_response, #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



103
104
105
106
107
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 103

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



52
53
54
55
56
57
58
59
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 52

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
 
# 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 server info and capabilities
  result = res['result'] || {}
  @server_info = result['serverInfo']
  @capabilities = result['capabilities']

  # 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



131
132
133
134
135
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 131

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

#rpc_request(method, params = {}) ⇒ 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:



116
117
118
119
120
121
122
123
124
125
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 116

def rpc_request(method, params = {})
  ensure_initialized
  with_retry do
    req_id = next_id
    req = build_jsonrpc_request(method, params, req_id)
    send_request(req)
    res = wait_response(req_id)
    process_jsonrpc_response(res)
  end
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:



65
66
67
68
69
70
71
72
73
74
 
# File 'lib/mcp_client/server_stdio/json_rpc_transport.rb', line 65

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) ⇒ Hash

Wait for a response with the given request ID

Parameters:

  • id (Integer) —

    the request ID

Returns:

  • (Hash) —

    the JSON-RPC response message

Raises:



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

def wait_response(id)
  deadline = Time.now + @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::TransportError, "Timeout waiting for JSONRPC response id=#{id}" unless msg

    msg
  end
end