Merge pull request #262 from koic/stdio_client_transport

Add `MCP::Client::Stdio` transport

.rubocop.yml

@@ -11,5 +11,8 @@ AllCops:
 Gemspec/DevelopmentDependencies:
   Enabled: true
 
+Lint/IncompatibleIoSelectWithFiberScheduler:
+  Enabled: true
+
 Minitest/LiteralAsActualArgument:
   Enabled: true

README.md

@@ -1082,6 +1082,52 @@ class CustomTransport
 end
 ```
 
+### Stdio Transport Layer
+
+Use the `MCP::Client::Stdio` transport to interact with MCP servers running as subprocesses over standard input/output.
+
+`MCP::Client::Stdio.new` accepts the following keyword arguments:
+
+| Parameter | Required | Description |
+|---|---|---|
+| `command:` | Yes | The command to spawn the server process (e.g., `"ruby"`, `"bundle"`, `"npx"`). |
+| `args:` | No | An array of arguments passed to the command. Defaults to `[]`. |
+| `env:` | No | A hash of environment variables to set for the server process. Defaults to `nil`. |
+| `read_timeout:` | No | Timeout in seconds for waiting for a server response. Defaults to `nil` (no timeout). |
+
+Example usage:
+
+```ruby
+stdio_transport = MCP::Client::Stdio.new(
+  command: "bundle",
+  args: ["exec", "ruby", "path/to/server.rb"],
+  env: { "API_KEY" => "my_secret_key" },
+  read_timeout: 30
+)
+client = MCP::Client.new(transport: stdio_transport)
+
+# List available tools.
+tools = client.tools
+tools.each do |tool|
+  puts "Tool: #{tool.name} - #{tool.description}"
+end
+
+# Call a specific tool.
+response = client.call_tool(
+  tool: tools.first,
+  arguments: { message: "Hello, world!" }
+)
+
+# Close the transport when done.
+stdio_transport.close
+```
+
+The stdio transport automatically handles:
+
+- Spawning the server process with `Open3.popen3`
+- MCP protocol initialization handshake (`initialize` request + `notifications/initialized`)
+- JSON-RPC 2.0 message framing over newline-delimited JSON
+
 ### HTTP Transport Layer
 
 Use the `MCP::Client::HTTP` transport to interact with MCP servers using simple HTTP requests.

examples/README.md

@@ -15,7 +15,26 @@ $ ruby examples/stdio_server.rb
 {"jsonrpc":"2.0","id":0,"method":"tools/list"}
 ```
 
-### 2. HTTP Server (`http_server.rb`)
+### 2. STDIO Client (`stdio_client.rb`)
+
+A client that connects to the STDIO server using the `MCP::Client::Stdio` transport.
+This demonstrates how to use the SDK's built-in client classes to interact with a server subprocess.
+
+**Usage:**
+
+```console
+$ ruby examples/stdio_client.rb
+```
+
+The client will automatically launch `stdio_server.rb` as a subprocess and demonstrate:
+
+- Listing and calling tools
+- Listing prompts
+- Listing and reading resources
+- Automatic MCP protocol initialization
+- Transport cleanup on exit
+
+### 3. HTTP Server (`http_server.rb`)
 
 A standalone HTTP server built with Rack that implements the MCP Streamable HTTP transport protocol. This demonstrates how to create a web-based MCP server with session management and Server-Sent Events (SSE) support.
 
@@ -41,7 +60,7 @@ The server will start on `http://localhost:9292` and provide:
 - **Prompts**: `ExamplePrompt` - echoes back arguments as a prompt
 - **Resources**: `test_resource` - returns example content
 
-### 3. HTTP Client Example (`http_client.rb`)
+### 4. HTTP Client Example (`http_client.rb`)
 
 A client that demonstrates how to interact with the HTTP server using all MCP protocol methods.
 
@@ -67,7 +86,7 @@ The client will demonstrate:
 - Listing and reading resources
 - Session cleanup
 
-### 4. Streamable HTTP Server (`streamable_http_server.rb`)
+### 5. Streamable HTTP Server (`streamable_http_server.rb`)
 
 A specialized HTTP server designed to test and demonstrate Server-Sent Events (SSE) functionality in the MCP protocol.
 
@@ -90,7 +109,7 @@ $ ruby examples/streamable_http_server.rb
 
 The server will start on `http://localhost:9393` and provide detailed instructions for testing SSE functionality.
 
-### 5. Streamable HTTP Client (`streamable_http_client.rb`)
+### 6. Streamable HTTP Client (`streamable_http_client.rb`)
 
 An interactive client that connects to the SSE stream and provides a menu-driven interface for testing SSE functionality.
 

examples/stdio_client.rb

@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift(File.expand_path("../lib", __dir__))
+require "mcp"
+require "json"
+
+# Simple stdio client example that connects to the stdio_server.rb example
+# Usage: ruby examples/stdio_client.rb
+
+server_script = File.expand_path("stdio_server.rb", __dir__)
+
+transport = MCP::Client::Stdio.new(command: "ruby", args: [server_script])
+client = MCP::Client.new(transport: transport)
+
+begin
+  # List available tools
+  puts "=== Listing tools ==="
+  tools = client.tools
+  tools.each do |tool|
+    puts "  Tool: #{tool.name} - #{tool.description}"
+  end
+
+  # Call the example_tool (adds two numbers)
+  puts "\n=== Calling tool: example_tool ==="
+  tool = tools.find { |t| t.name == "example_tool" }
+  response = client.call_tool(tool: tool, arguments: { a: 5, b: 3 })
+  puts "  Response: #{JSON.pretty_generate(response.dig("result", "content"))}"
+
+  # Call the echo tool
+  puts "\n=== Calling tool: echo ==="
+  tool = tools.find { |t| t.name == "echo" }
+  response = client.call_tool(tool: tool, arguments: { message: "Hello from stdio client!" })
+  puts "  Response: #{JSON.pretty_generate(response.dig("result", "content"))}"
+
+  # List prompts
+  puts "\n=== Listing prompts ==="
+  prompts = client.prompts
+  prompts.each do |prompt|
+    puts "  Prompt: #{prompt["name"]} - #{prompt["description"]}"
+  end
+
+  # List resources
+  puts "\n=== Listing resources ==="
+  resources = client.resources
+  resources.each do |resource|
+    puts "  Resource: #{resource["name"]} (#{resource["uri"]})"
+  end
+
+  # Read a resource
+  puts "\n=== Reading resource: https://test_resource.invalid ==="
+  contents = client.read_resource(uri: "https://test_resource.invalid")
+  puts "  Response: #{JSON.pretty_generate(contents)}"
+rescue => e
+  puts "Error: #{e.message}"
+  puts e.backtrace.first(5).join("\n")
+ensure
+  transport.close
+end

lib/mcp/client.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require_relative "client/stdio"
 require_relative "client/http"
 require_relative "client/tool"
 

lib/mcp/client/stdio.rb

@@ -0,0 +1,222 @@
+# frozen_string_literal: true
+
+require "json"
+require "open3"
+require "securerandom"
+require "timeout"
+require_relative "../../json_rpc_handler"
+require_relative "../configuration"
+require_relative "../methods"
+require_relative "../version"
+
+module MCP
+  class Client
+    class Stdio
+      # Seconds to wait for the server process to exit before sending SIGTERM.
+      # Matches the Python and TypeScript SDKs' shutdown timeout:
+      # https://github.com/modelcontextprotocol/python-sdk/blob/v1.26.0/src/mcp/client/stdio/__init__.py#L48
+      # https://github.com/modelcontextprotocol/typescript-sdk/blob/v1.27.1/src/client/stdio.ts#L221
+      CLOSE_TIMEOUT = 2
+      STDERR_READ_SIZE = 4096
+
+      attr_reader :command, :args, :env
+
+      def initialize(command:, args: [], env: nil, read_timeout: nil)
+        @command = command
+        @args = args
+        @env = env
+        @read_timeout = read_timeout
+        @stdin = nil
+        @stdout = nil
+        @stderr = nil
+        @wait_thread = nil
+        @stderr_thread = nil
+        @started = false
+        @initialized = false
+      end
+
+      def send_request(request:)
+        start unless @started
+        initialize_session unless @initialized
+
+        write_message(request)
+        read_response(request)
+      end
+
+      def start
+        raise "MCP::Client::Stdio already started" if @started
+
+        spawn_env = @env || {}
+        @stdin, @stdout, @stderr, @wait_thread = Open3.popen3(spawn_env, @command, *@args)
+        @stdout.set_encoding("UTF-8")
+        @stdin.set_encoding("UTF-8")
+
+        # Drain stderr in the background to prevent the pipe buffer from filling up,
+        # which would cause the server process to block and deadlock.
+        @stderr_thread = Thread.new do
+          loop do
+            @stderr.readpartial(STDERR_READ_SIZE)
+          end
+        rescue IOError
+          nil
+        end
+
+        @started = true
+      rescue Errno::ENOENT, Errno::EACCES, Errno::ENOEXEC => e
+        raise RequestHandlerError.new(
+          "Failed to spawn server process: #{e.message}",
+          {},
+          error_type: :internal_error,
+          original_error: e,
+        )
+      end
+
+      def close
+        return unless @started
+
+        @stdin.close
+        @stdout.close
+        @stderr.close
+
+        begin
+          Timeout.timeout(CLOSE_TIMEOUT) { @wait_thread.value }
+        rescue Timeout::Error
+          begin
+            Process.kill("TERM", @wait_thread.pid)
+            Timeout.timeout(CLOSE_TIMEOUT) { @wait_thread.value }
+          rescue Timeout::Error
+            begin
+              Process.kill("KILL", @wait_thread.pid)
+            rescue Errno::ESRCH
+              nil
+            end
+          rescue Errno::ESRCH
+            nil
+          end
+        end
+
+        @stderr_thread.join(CLOSE_TIMEOUT)
+        @started = false
+        @initialized = false
+      end
+
+      private
+
+      # The client MUST send a protocol version it supports. This SHOULD be the latest version.
+      # https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#version-negotiation
+      #
+      # Always sends `LATEST_STABLE_PROTOCOL_VERSION`, matching the Python and TypeScript SDKs:
+      # https://github.com/modelcontextprotocol/python-sdk/blob/v1.26.0/src/mcp/client/session.py#L175
+      # https://github.com/modelcontextprotocol/typescript-sdk/blob/v1.27.1/src/client/index.ts#L495
+      def initialize_session
+        init_request = {
+          jsonrpc: JsonRpcHandler::Version::V2_0,
+          id: SecureRandom.uuid,
+          method: MCP::Methods::INITIALIZE,
+          params: {
+            protocolVersion: MCP::Configuration::LATEST_STABLE_PROTOCOL_VERSION,
+            capabilities: {},
+            clientInfo: { name: "mcp-ruby-client", version: MCP::VERSION },
+          },
+        }
+
+        write_message(init_request)
+        response = read_response(init_request)
+
+        if response.key?("error")
+          error = response["error"]
+          raise RequestHandlerError.new(
+            "Server initialization failed: #{error["message"]}",
+            { method: MCP::Methods::INITIALIZE },
+            error_type: :internal_error,
+          )
+        end
+
+        unless response.key?("result")
+          raise RequestHandlerError.new(
+            "Server initialization failed: missing result in response",
+            { method: MCP::Methods::INITIALIZE },
+            error_type: :internal_error,
+          )
+        end
+
+        notification = {
+          jsonrpc: JsonRpcHandler::Version::V2_0,
+          method: MCP::Methods::NOTIFICATIONS_INITIALIZED,
+        }
+        write_message(notification)
+
+        @initialized = true
+      end
+
+      def write_message(message)
+        ensure_running!
+        json = JSON.generate(message)
+        @stdin.puts(json)
+        @stdin.flush
+      rescue IOError, Errno::EPIPE => e
+        raise RequestHandlerError.new(
+          "Failed to write to server process",
+          {},
+          error_type: :internal_error,
+          original_error: e,
+        )
+      end
+
+      def read_response(request)
+        request_id = request[:id] || request["id"]
+        method = request[:method] || request["method"]
+        params = request[:params] || request["params"]
+
+        loop do
+          ensure_running!
+          wait_for_readable!(method, params) if @read_timeout
+          line = @stdout.gets
+          raise_connection_error!(method, params) if line.nil?
+
+          parsed = JSON.parse(line.strip)
+
+          next unless parsed.key?("id")
+
+          return parsed if parsed["id"] == request_id
+        end
+      rescue JSON::ParserError => e
+        raise RequestHandlerError.new(
+          "Failed to parse server response",
+          { method: method, params: params },
+          error_type: :internal_error,
+          original_error: e,
+        )
+      end
+
+      def ensure_running!
+        return if @wait_thread.alive?
+
+        raise RequestHandlerError.new(
+          "Server process has exited",
+          {},
+          error_type: :internal_error,
+        )
+      end
+
+      def wait_for_readable!(method, params)
+        ready = @stdout.wait_readable(@read_timeout)
+        return if ready
+
+        raise RequestHandlerError.new(
+          "Timed out waiting for server response",
+          { method: method, params: params },
+          error_type: :internal_error,
+        )
+      end
+
+      def raise_connection_error!(method, params)
+        raise RequestHandlerError.new(
+          "Server process closed stdout unexpectedly",
+          { method: method, params: params },
+          error_type: :internal_error,
+        )
+      end
+    end
+  end
+end

test/mcp/client/stdio_test.rb

@@ -0,0 +1,726 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require "json"
+require "mcp/client"
+require "mcp/client/stdio"
+require "mcp/client/tool"
+
+module MCP
+  class Client
+    class StdioTest < Minitest::Test
+      def test_send_request_starts_process_and_returns_response
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        # Simulate server responses: initialize response, then tools/list response
+        Thread.new do
+          # Read and respond to initialize request
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          init_response = {
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }
+          stdout_write.puts(JSON.generate(init_response))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read and respond to tools/list request
+          tools_line = stdin_read.gets
+          tools_request = JSON.parse(tools_line)
+          tools_response = {
+            jsonrpc: "2.0",
+            id: tools_request["id"],
+            result: { tools: [{ name: "test_tool", description: "A test tool", inputSchema: {} }] },
+          }
+          stdout_write.puts(JSON.generate(tools_response))
+          stdout_write.flush
+        end
+
+        response = transport.send_request(request: request)
+
+        assert_equal("test-id", response["id"])
+        assert_equal(1, response.dig("result", "tools").size)
+        assert_equal("test_tool", response.dig("result", "tools", 0, "name"))
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_initializes_session_on_first_call
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        received_methods = []
+
+        Thread.new do
+          # Read initialize request
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          received_methods << init_request["method"]
+
+          init_response = {
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }
+          stdout_write.puts(JSON.generate(init_response))
+          stdout_write.flush
+
+          # Read initialized notification
+          notification_line = stdin_read.gets
+          notification = JSON.parse(notification_line)
+          received_methods << notification["method"]
+
+          # Read tools/list request
+          tools_line = stdin_read.gets
+          tools_request = JSON.parse(tools_line)
+          received_methods << tools_request["method"]
+
+          tools_response = {
+            jsonrpc: "2.0",
+            id: tools_request["id"],
+            result: { tools: [] },
+          }
+          stdout_write.puts(JSON.generate(tools_response))
+          stdout_write.flush
+        end
+
+        transport.send_request(request: request)
+
+        assert_equal(["initialize", "notifications/initialized", "tools/list"], received_methods)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_skips_notifications
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Handle initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read tools/list request
+          stdin_read.gets
+
+          # Send a notification before the response
+          notification = { jsonrpc: "2.0", method: "notifications/tools/list_changed" }
+          stdout_write.puts(JSON.generate(notification))
+          stdout_write.flush
+
+          # Then send the actual response
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: "test-id",
+            result: { tools: [] },
+          }))
+          stdout_write.flush
+        end
+
+        response = transport.send_request(request: request)
+
+        assert_equal("test-id", response["id"])
+        assert_equal([], response.dig("result", "tools"))
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_raises_error_when_process_exits
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+
+        dead_thread = mock("wait_thread")
+        dead_thread.stubs(:alive?).returns(false)
+        dead_thread.stubs(:value).returns(nil)
+
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, dead_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+        transport.start
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Server process has exited", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_raises_error_on_closed_stdout
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Handle initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read tools/list request, then close stdout
+          stdin_read.gets
+          stdout_write.close
+        end
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Server process closed stdout unexpectedly", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_close_resets_state
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, stderr_write = IO.pipe
+
+        wait_thread = mock("wait_thread")
+        wait_thread.stubs(:alive?).returns(true)
+        wait_thread.stubs(:value).returns(nil)
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+        transport.start
+
+        assert(transport.instance_variable_get(:@started))
+
+        transport.close
+
+        refute(transport.instance_variable_get(:@started))
+        refute(transport.instance_variable_get(:@initialized))
+      ensure
+        stdin_read.close
+        begin
+          stdin_write.close
+        rescue
+          nil
+        end
+        begin
+          stdout_read.close
+        rescue
+          nil
+        end
+        stdout_write.close
+        begin
+          stderr_read.close
+        rescue
+          nil
+        end
+        stderr_write.close
+      end
+
+      def test_send_request_skips_initialization_on_second_call
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        received_methods = []
+
+        Thread.new do
+          # First call: initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          received_methods << init_request["method"]
+
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          notification_line = stdin_read.gets
+          received_methods << JSON.parse(notification_line)["method"]
+
+          # First request: tools/list
+          first_line = stdin_read.gets
+          first_request = JSON.parse(first_line)
+          received_methods << first_request["method"]
+
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: first_request["id"],
+            result: { tools: [] },
+          }))
+          stdout_write.flush
+
+          # Second request: tools/list (no re-initialization)
+          second_line = stdin_read.gets
+          second_request = JSON.parse(second_line)
+          received_methods << second_request["method"]
+
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: second_request["id"],
+            result: { tools: [] },
+          }))
+          stdout_write.flush
+        end
+
+        transport.send_request(request: { jsonrpc: "2.0", id: "first", method: "tools/list" })
+        transport.send_request(request: { jsonrpc: "2.0", id: "second", method: "tools/list" })
+
+        assert_equal(
+          ["initialize", "notifications/initialized", "tools/list", "tools/list"],
+          received_methods,
+        )
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_env_is_passed_to_process
+        transport = Stdio.new(command: "ruby", args: ["server.rb"], env: { "FOO" => "bar" })
+
+        Open3.expects(:popen3).with({ "FOO" => "bar" }, "ruby", "server.rb").returns(
+          [StringIO.new, StringIO.new, StringIO.new, mock_wait_thread],
+        )
+
+        transport.start
+      end
+
+      def test_send_request_raises_error_on_invalid_json
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Handle initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read tools/list request, then send invalid JSON
+          stdin_read.gets
+          stdout_write.puts("not valid json")
+          stdout_write.flush
+        end
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Failed to parse server response", error.message)
+        assert_equal(:internal_error, error.error_type)
+        assert_instance_of(JSON::ParserError, error.original_error)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_raises_error_when_initialization_fails
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Read initialize request and return an error
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            error: { code: -32600, message: "Invalid Request", data: "Unsupported protocol version" },
+          }))
+          stdout_write.flush
+        end
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Server initialization failed: Invalid Request", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_close_kills_process_on_timeout
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        hanging_thread = mock("wait_thread")
+        hanging_thread.stubs(:alive?).returns(true)
+        hanging_thread.stubs(:pid).returns(99999)
+        hanging_thread.stubs(:value).raises(Timeout::Error)
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, hanging_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+        transport.start
+
+        Process.expects(:kill).with("TERM", 99999).once
+        Process.expects(:kill).with("KILL", 99999).once
+
+        transport.close
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_read_response_raises_error_on_timeout
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"], read_timeout: 0.01)
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Handle initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read tools/list request but don't respond (simulate timeout)
+          stdin_read.gets
+        end
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Timed out waiting for server response", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_send_request_raises_error_when_stdin_is_closed
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        server_thread = Thread.new do
+          # Handle initialize handshake
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+            result: {
+              protocolVersion: "2025-11-25",
+              capabilities: {},
+              serverInfo: { name: "test-server", version: "1.0.0" },
+            },
+          }))
+          stdout_write.flush
+
+          # Read initialized notification
+          stdin_read.gets
+
+          # Read and respond to first request
+          line = stdin_read.gets
+          request = JSON.parse(line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: request["id"],
+            result: {},
+          }))
+          stdout_write.flush
+        end
+
+        # Complete handshake with a successful request
+        transport.send_request(request: { jsonrpc: "2.0", id: "setup", method: "ping" })
+        server_thread.join
+
+        # Now close stdin to simulate broken pipe
+        stdin_write.close
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: { jsonrpc: "2.0", id: "test-id", method: "tools/list" })
+        end
+
+        assert_equal("Failed to write to server process", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        begin
+          stdin_write.close
+        rescue
+          nil
+        end
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_close_is_noop_when_not_started
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        # Should not raise
+        transport.close
+      end
+
+      def test_start_raises_error_when_already_started
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+        transport.start
+
+        error = assert_raises(RuntimeError) do
+          transport.start
+        end
+
+        assert_equal("MCP::Client::Stdio already started", error.message)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      def test_start_raises_error_for_invalid_command
+        Open3.stubs(:popen3).raises(Errno::ENOENT.new("No such file or directory - nonexistent_command"))
+
+        transport = Stdio.new(command: "nonexistent_command")
+
+        error = assert_raises(RequestHandlerError) do
+          transport.start
+        end
+
+        assert_match(/Failed to spawn server process/, error.message)
+        assert_equal(:internal_error, error.error_type)
+        assert_instance_of(Errno::ENOENT, error.original_error)
+      end
+
+      def test_send_request_raises_error_for_missing_result
+        stdin_read, stdin_write = IO.pipe
+        stdout_read, stdout_write = IO.pipe
+        stderr_read, _ = IO.pipe
+
+        Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
+
+        transport = Stdio.new(command: "ruby", args: ["server.rb"])
+
+        request = {
+          jsonrpc: "2.0",
+          id: "test-id",
+          method: "tools/list",
+        }
+
+        Thread.new do
+          # Read initialize request and return a response without result
+          init_line = stdin_read.gets
+          init_request = JSON.parse(init_line)
+          stdout_write.puts(JSON.generate({
+            jsonrpc: "2.0",
+            id: init_request["id"],
+          }))
+          stdout_write.flush
+        end
+
+        error = assert_raises(RequestHandlerError) do
+          transport.send_request(request: request)
+        end
+
+        assert_equal("Server initialization failed: missing result in response", error.message)
+        assert_equal(:internal_error, error.error_type)
+      ensure
+        stdin_read.close
+        stdin_write.close
+        stdout_read.close
+        stdout_write.close
+      end
+
+      private
+
+      def mock_wait_thread
+        thread = mock("wait_thread")
+        thread.stubs(:alive?).returns(true)
+        thread.stubs(:value).returns(nil)
+        thread
+      end
+    end
+  end
+end