7db22f4d5b## Motivation and Context
The Ruby SDK accepted duplicate `initialize` requests after a session was already initialized.
On stdio, a second `initialize` silently overwrote the per-session `clientInfo` and `clientCapabilities`
(including with an older `protocolVersion`). On Streamable HTTP, every `initialize` minted a fresh
`Mcp-Session-Id` and `ServerSession`, abandoning the originally negotiated session.
MCP specification (`2025-06-18` / `2025-11-25` lifecycle) states that the initialization phase
MUST be the first interaction between client and server;
re-initialization on an established session is not part of the defined lifecycle.
TypeScript SDK rejects duplicate `initialize` on a live session with HTTP 400 + JSON-RPC `-32600`
("Invalid Request: Server already initialized"), and Python SDK does not mint a new session
on duplicate `initialize`. The Ruby SDK was the outlier; this change aligns it with the TypeScript SDK.
- `ServerSession` tracks an `@initialized` flag, exposed via `initialized?` and
set by `mark_initialized!` after a successful `initialize` response.
- `Server#init` raises `RequestHandlerError(error_type: :invalid_request)` when
the session is already initialized, which the existing error mapping converts
to JSON-RPC `-32600 Invalid Request`.
- `StreamableHTTPTransport#handle_post` short-circuits at the transport layer:
duplicate `initialize` against a live session returns HTTP 400 + JSON-RPC
`-32600`; a stale or expired `Mcp-Session-Id` returns 404 (evicting the expired entry
instead of misreporting it as a duplicate).
- `handle_initialization` evicts the registered session and omits the `Mcp-Session-Id`
header when the first `initialize` fails before `mark_initialized!` is reached,
so retries do not collide with an orphaned ID.
- Non-Hash JSON-RPC POST bodies (e.g. batched arrays, which are not supported in `2025-11-25`)
are explicitly rejected with HTTP 400 + JSON-RPC `-32600` rather than falling through to
an unparseable Rack response.
## How Has This Been Tested?
- Server tests: a second `initialize` on the same `ServerSession` returns
`code: -32600` and the original `clientInfo` is preserved.
- Streamable HTTP tests: duplicate `initialize` with a live `Mcp-Session-Id`
returns HTTP 400 + `-32600` and the original session remains usable for
subsequent `ping`; stale `Mcp-Session-Id` returns 404; an idle-expired
session is evicted on duplicate `initialize` and returns 404; a failed
`initialize` (invalid `jsonrpc` envelope) does not leak `Mcp-Session-Id` and
leaves `@sessions` empty; an array body is rejected with HTTP 400 + `-32600`.
- Stdio tests: a second `initialize` on the same stdio session returns
`code: -32600` and the original `clientInfo` is preserved.
## Breaking Changes
Clients that previously sent `initialize` more than once on the same session
now receive a JSON-RPC error with `code: -32600` for the second and later requests
instead of silently overwriting session state (stdio) or being re-issued
a new `Mcp-Session-Id` (Streamable HTTP). Clients that follow the MCP specification
(single `initialize` per session) are unaffected.
Additionally, non-Hash JSON-RPC POST bodies on Streamable HTTP now return
HTTP 400 + JSON-RPC `-32600` rather than falling through to a broken Rack response.
The previous behavior produced an unparseable response, so this is unlikely to
affect any working client.
Closes #349.
447 lines
14 KiB
Ruby
447 lines
14 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
require "test_helper"
|
|
|
|
module MCP
|
|
class Server
|
|
module Transports
|
|
class StdioTransportTest < ActiveSupport::TestCase
|
|
include InstrumentationTestHelper
|
|
|
|
setup do
|
|
configuration = MCP::Configuration.new
|
|
configuration.instrumentation_callback = instrumentation_helper.callback
|
|
@server = Server.new(name: "test_server", configuration: configuration)
|
|
@transport = StdioTransport.new(@server)
|
|
end
|
|
|
|
test "initializes with server and closed state" do
|
|
server = @transport.instance_variable_get(:@server)
|
|
assert_equal @server.object_id, server.object_id
|
|
refute @transport.instance_variable_get(:@open)
|
|
end
|
|
|
|
test "processes JSON-RPC requests from stdin and sends responses to stdout" do
|
|
request = {
|
|
jsonrpc: "2.0",
|
|
method: "ping",
|
|
id: "123",
|
|
}
|
|
input = StringIO.new(JSON.generate(request) + "\n")
|
|
output = StringIO.new
|
|
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = input
|
|
$stdout = output
|
|
|
|
thread = Thread.new { @transport.open }
|
|
sleep(0.1)
|
|
@transport.close
|
|
thread.join
|
|
|
|
response = JSON.parse(output.string, symbolize_names: true)
|
|
assert_equal("2.0", response[:jsonrpc])
|
|
assert_equal("123", response[:id])
|
|
assert_equal({}, response[:result])
|
|
refute(@transport.instance_variable_get(:@open))
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "sends string responses to stdout" do
|
|
output = StringIO.new
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdout = output
|
|
@transport.send_response("test response")
|
|
assert_equal("test response\n", output.string)
|
|
ensure
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "sends JSON responses to stdout" do
|
|
output = StringIO.new
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdout = output
|
|
response = { key: "value" }
|
|
@transport.send_response(response)
|
|
assert_equal(JSON.generate(response) + "\n", output.string)
|
|
ensure
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "handles valid JSON-RPC requests" do
|
|
request = {
|
|
jsonrpc: "2.0",
|
|
method: "ping",
|
|
id: "123",
|
|
}
|
|
output = StringIO.new
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdout = output
|
|
@transport.send(:handle_request, JSON.generate(request))
|
|
response = JSON.parse(output.string, symbolize_names: true)
|
|
assert_equal("2.0", response[:jsonrpc])
|
|
assert_nil(response[:id])
|
|
assert_nil(response[:result])
|
|
ensure
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "open creates a ServerSession and processes requests through it" do
|
|
request = {
|
|
jsonrpc: "2.0",
|
|
method: "initialize",
|
|
id: "1",
|
|
params: {
|
|
protocolVersion: "2025-11-25",
|
|
clientInfo: { name: "stdio-client", version: "1.0" },
|
|
},
|
|
}
|
|
input = StringIO.new(JSON.generate(request) + "\n")
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = input
|
|
$stdout = output
|
|
@transport.open
|
|
|
|
# Verify a session was created.
|
|
session = @transport.instance_variable_get(:@session)
|
|
assert_instance_of(ServerSession, session)
|
|
|
|
# Verify client info was stored on the session, not on the server.
|
|
assert_equal({ name: "stdio-client", version: "1.0" }, session.client)
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "rejects duplicate initialize on the same stdio session with -32600" do
|
|
first = {
|
|
jsonrpc: "2.0",
|
|
method: "initialize",
|
|
id: "first",
|
|
params: {
|
|
protocolVersion: "2025-11-25",
|
|
clientInfo: { name: "original", version: "1.0" },
|
|
},
|
|
}
|
|
second = {
|
|
jsonrpc: "2.0",
|
|
method: "initialize",
|
|
id: "second",
|
|
params: {
|
|
protocolVersion: "2024-11-05",
|
|
clientInfo: { name: "intruder", version: "9.9" },
|
|
},
|
|
}
|
|
input = StringIO.new("#{JSON.generate(first)}\n#{JSON.generate(second)}\n")
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = input
|
|
$stdout = output
|
|
@transport.open
|
|
|
|
lines = output.string.lines
|
|
assert_equal(2, lines.length)
|
|
first_response = JSON.parse(lines[0], symbolize_names: true)
|
|
second_response = JSON.parse(lines[1], symbolize_names: true)
|
|
|
|
assert_equal("first", first_response[:id])
|
|
refute_nil(first_response[:result])
|
|
|
|
assert_equal("second", second_response[:id])
|
|
assert_equal(-32600, second_response[:error][:code])
|
|
assert_equal("Invalid Request", second_response[:error][:message])
|
|
|
|
session = @transport.instance_variable_get(:@session)
|
|
assert_equal({ name: "original", version: "1.0" }, session.client)
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "handles invalid JSON requests" do
|
|
invalid_json = "invalid json"
|
|
output = StringIO.new
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdout = output
|
|
@transport.send(:handle_request, invalid_json)
|
|
response = JSON.parse(output.string, symbolize_names: true)
|
|
assert_equal("2.0", response[:jsonrpc])
|
|
assert_nil(response[:id])
|
|
assert_equal(-32600, response[:error][:code])
|
|
assert_equal("Invalid Request", response[:error][:message])
|
|
assert_equal("Request must be an array or a hash", response[:error][:data])
|
|
ensure
|
|
$stdout = original_stdout
|
|
end
|
|
end
|
|
|
|
test "send_request sends request to stdout and waits for response" do
|
|
reader, writer = IO.pipe
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = reader
|
|
$stdout = output
|
|
@transport.instance_variable_set(:@open, true)
|
|
|
|
# Send response from client in a thread.
|
|
Thread.new do
|
|
sleep(0.05) # Wait for request to be written to `StringIO`.
|
|
request = JSON.parse(output.string.lines.first, symbolize_names: true)
|
|
response = {
|
|
jsonrpc: "2.0",
|
|
id: request[:id],
|
|
result: { content: "test response" },
|
|
}
|
|
writer.puts(response.to_json)
|
|
writer.flush
|
|
end
|
|
|
|
result = @transport.send_request("test/method", { param: "value" })
|
|
|
|
assert_equal({ content: "test response" }, result)
|
|
|
|
# Verify request was sent.
|
|
request = JSON.parse(output.string.lines.first, symbolize_names: true)
|
|
assert_equal("2.0", request[:jsonrpc])
|
|
assert_equal("test/method", request[:method])
|
|
assert_equal({ param: "value" }, request[:params])
|
|
assert(request[:id])
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
begin
|
|
writer.close
|
|
rescue
|
|
nil
|
|
end
|
|
begin
|
|
reader.close
|
|
rescue
|
|
nil
|
|
end
|
|
end
|
|
end
|
|
|
|
test "send_request raises on error response from client" do
|
|
reader, writer = IO.pipe
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = reader
|
|
$stdout = output
|
|
@transport.instance_variable_set(:@open, true)
|
|
|
|
Thread.new do
|
|
sleep(0.05) # Wait for request to be written to `StringIO`.
|
|
request = JSON.parse(output.string.lines.first, symbolize_names: true)
|
|
error_response = {
|
|
jsonrpc: "2.0",
|
|
id: request[:id],
|
|
error: { code: -1, message: "User rejected sampling request" },
|
|
}
|
|
writer.puts(error_response.to_json)
|
|
writer.flush
|
|
end
|
|
|
|
error = assert_raises(StandardError) do
|
|
@transport.send_request("sampling/createMessage", { messages: [] })
|
|
end
|
|
|
|
assert_equal("Client returned an error for sampling/createMessage request (code: -1): User rejected sampling request", error.message)
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
begin
|
|
writer.close
|
|
rescue
|
|
nil
|
|
end
|
|
begin
|
|
reader.close
|
|
rescue
|
|
nil
|
|
end
|
|
end
|
|
end
|
|
|
|
test "send_request does not double-report intentional raises via exception_reporter" do
|
|
reader, writer = IO.pipe
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
reported_errors = []
|
|
original_reporter = MCP.configuration.exception_reporter
|
|
|
|
begin
|
|
MCP.configuration.exception_reporter = ->(e, ctx) { reported_errors << [e, ctx] }
|
|
$stdin = reader
|
|
$stdout = output
|
|
@transport.instance_variable_set(:@open, true)
|
|
|
|
Thread.new do
|
|
sleep(0.05) # Wait for request to be written to `StringIO`.
|
|
request = JSON.parse(output.string.lines.first, symbolize_names: true)
|
|
error_response = {
|
|
jsonrpc: "2.0",
|
|
id: request[:id],
|
|
error: { code: -1, message: "rejected" },
|
|
}
|
|
writer.puts(error_response.to_json)
|
|
writer.flush
|
|
end
|
|
|
|
assert_raises(StandardError) do
|
|
@transport.send_request("sampling/createMessage", { messages: [] })
|
|
end
|
|
|
|
assert_empty(reported_errors)
|
|
ensure
|
|
MCP.configuration.exception_reporter = original_reporter
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
begin
|
|
writer.close
|
|
rescue
|
|
nil
|
|
end
|
|
begin
|
|
reader.close
|
|
rescue
|
|
nil
|
|
end
|
|
end
|
|
end
|
|
|
|
test "send_request processes interleaved requests via session" do
|
|
reader, writer = IO.pipe
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = reader
|
|
$stdout = output
|
|
@transport.instance_variable_set(:@open, true)
|
|
|
|
# Initialize a session so @session is set.
|
|
session = MCP::ServerSession.new(server: @server, transport: @transport)
|
|
@transport.instance_variable_set(:@session, session)
|
|
|
|
Thread.new do
|
|
sleep(0.05) # Wait for request to be written to `StringIO`.
|
|
request = JSON.parse(output.string.lines.first, symbolize_names: true)
|
|
|
|
# Send an interleaved ping request before the response.
|
|
ping = { jsonrpc: "2.0", method: "ping", id: "ping-1" }
|
|
writer.puts(ping.to_json)
|
|
writer.flush
|
|
|
|
sleep(0.05) # Wait for the ping to be processed.
|
|
|
|
# Then send the actual response.
|
|
response = {
|
|
jsonrpc: "2.0",
|
|
id: request[:id],
|
|
result: { content: "done" },
|
|
}
|
|
writer.puts(response.to_json)
|
|
writer.flush
|
|
end
|
|
|
|
result = @transport.send_request("test/method", { param: "value" })
|
|
|
|
assert_equal({ content: "done" }, result)
|
|
|
|
# Verify the interleaved ping was handled (response sent to output).
|
|
lines = output.string.lines
|
|
ping_response = lines.find { |l| l.include?("ping-1") }
|
|
assert(ping_response, "Interleaved ping request should have been handled")
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
begin
|
|
writer.close
|
|
rescue
|
|
nil
|
|
end
|
|
begin
|
|
reader.close
|
|
rescue
|
|
nil
|
|
end
|
|
end
|
|
end
|
|
|
|
test "send_request raises when transport is closed while waiting" do
|
|
reader, writer = IO.pipe
|
|
output = StringIO.new
|
|
original_stdin = $stdin
|
|
original_stdout = $stdout
|
|
|
|
begin
|
|
$stdin = reader
|
|
$stdout = output
|
|
@transport.instance_variable_set(:@open, true)
|
|
|
|
# Close transport while waiting for response.
|
|
Thread.new do
|
|
sleep(0.05) # Wait for request to be written to `StringIO`.
|
|
@transport.instance_variable_set(:@open, false)
|
|
writer.close
|
|
end
|
|
|
|
error = assert_raises(RuntimeError) do
|
|
@transport.send_request("sampling/createMessage", { messages: [] })
|
|
end
|
|
|
|
assert_equal("Transport closed while waiting for response to sampling/createMessage request.", error.message)
|
|
ensure
|
|
$stdin = original_stdin
|
|
$stdout = original_stdout
|
|
begin
|
|
writer.close
|
|
rescue IOError
|
|
nil
|
|
end
|
|
begin
|
|
reader.close
|
|
rescue IOError
|
|
nil
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|