Class: SwarmSDK::V3::MCP::StdioTransport

Inherits:
Object
  • Object
show all
Defined in:
lib/swarm_sdk/v3/mcp/stdio_transport.rb

Overview

Subprocess-based JSON-RPC transport for MCP stdio servers

Spawns a child process and communicates via JSON-RPC over stdin/stdout. Performs the MCP protocol initialization handshake automatically. Implements the same duck type as ‘MCP::Client::HTTP` (`send_request(request:)`).

Examples:

transport = StdioTransport.new(
  command: "npx",
  args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
)
response = transport.send_request(request: { jsonrpc: "2.0", id: "1", method: "tools/list" })
transport.close

Constant Summary collapse

PROTOCOL_VERSION = 
"2024-11-05"

Instance Method Summary collapse

Constructor Details

#initialize(command:, args: [], env: {}) ⇒ StdioTransport

Create a new stdio transport, spawn the subprocess, and initialize the MCP session

Examples:

StdioTransport.new(command: "node", args: ["server.js"])

Parameters:

  • command (String) —

    Command to execute

  • args (Array<String>) (defaults to: []) —

    Command arguments

  • env (Hash<String, String>) (defaults to: {}) —

    Environment variables

Raises:

  • (McpError) —

    If the subprocess fails to start or initialization fails



32
33
34
35
36
37
 
# File 'lib/swarm_sdk/v3/mcp/stdio_transport.rb', line 32

def initialize(command:, args: [], env: {})
  @stdin, @stdout, @wait_thread = Open3.popen2(
    env.transform_keys(&:to_s), command, *args
  )
  initialize_session
end

Instance Method Details

#alive? ⇒ Boolean

Whether the subprocess is still running

Returns:

  • (Boolean) 


82
83
84
 
# File 'lib/swarm_sdk/v3/mcp/stdio_transport.rb', line 82

def alive?
  @wait_thread&.alive? || false
end

#close ⇒ void

This method returns an undefined value.

Gracefully shut down the subprocess

Closes stdin to signal the subprocess to exit, then waits briefly before sending SIGTERM if still alive.



65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/swarm_sdk/v3/mcp/stdio_transport.rb', line 65

def close
  @stdin&.close unless @stdin&.closed?
  return unless @wait_thread

  # Give the process a moment to exit gracefully
  return if @wait_thread.join(2)

  # Force terminate if still running
  Process.kill("TERM", @wait_thread.pid) if @wait_thread.alive?
  @wait_thread.join(2)
rescue Errno::ESRCH, Errno::ECHILD
  # Process already exited
end

#send_request(request:) ⇒ Hash

Send a JSON-RPC request and return the response

Writes the request as a single JSON line to the subprocess stdin, then reads lines from stdout until a response with an “id” field is received (skipping server-initiated notifications).

Examples:

response = transport.send_request(
  request: { jsonrpc: "2.0", id: "1", method: "tools/list" }
)

Parameters:

  • request (Hash) —

    Complete JSON-RPC 2.0 request object

Returns:

  • (Hash) —

    Parsed JSON-RPC response

Raises:

  • (McpError) —

    If the subprocess exits unexpectedly



53
54
55
56
57
 
# File 'lib/swarm_sdk/v3/mcp/stdio_transport.rb', line 53

def send_request(request:)
  @stdin.puts(JSON.generate(request))
  @stdin.flush
  read_response
end