Merge pull request #101 from 1debit/stateless
Add `stateless` feature to Streamable HTTP protocol
This commit is contained in:
commit
400a56e83d
3 changed files with 179 additions and 14 deletions
|
|
@ -126,12 +126,19 @@ Notifications follow the JSON-RPC 2.0 specification and use these method names:
|
||||||
|
|
||||||
- **stdio**: Notifications are sent as JSON-RPC 2.0 messages to stdout
|
- **stdio**: Notifications are sent as JSON-RPC 2.0 messages to stdout
|
||||||
- **Streamable HTTP**: Notifications are sent as JSON-RPC 2.0 messages over HTTP with streaming (chunked transfer or SSE)
|
- **Streamable HTTP**: Notifications are sent as JSON-RPC 2.0 messages over HTTP with streaming (chunked transfer or SSE)
|
||||||
|
- **Stateless Streamable HTTP**: Notifications are not supported and all calls are request/response interactions; allows for easy multi-node deployment.
|
||||||
|
|
||||||
#### Usage Example
|
#### Usage Example
|
||||||
|
|
||||||
```ruby
|
```ruby
|
||||||
server = MCP::Server.new(name: "my_server")
|
server = MCP::Server.new(name: "my_server")
|
||||||
|
|
||||||
|
# Default Streamable HTTP - session oriented
|
||||||
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
||||||
|
|
||||||
|
# OR Stateless Streamable HTTP - session-less
|
||||||
|
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, stateless: true)
|
||||||
|
|
||||||
server.transport = transport
|
server.transport = transport
|
||||||
|
|
||||||
# When tools change, notify clients
|
# When tools change, notify clients
|
||||||
|
|
|
||||||
|
|
@ -8,11 +8,13 @@ module MCP
|
||||||
class Server
|
class Server
|
||||||
module Transports
|
module Transports
|
||||||
class StreamableHTTPTransport < Transport
|
class StreamableHTTPTransport < Transport
|
||||||
def initialize(server)
|
def initialize(server, stateless: false)
|
||||||
super
|
super(server)
|
||||||
# { session_id => { stream: stream_object }
|
# { session_id => { stream: stream_object }
|
||||||
@sessions = {}
|
@sessions = {}
|
||||||
@mutex = Mutex.new
|
@mutex = Mutex.new
|
||||||
|
|
||||||
|
@stateless = stateless
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_request(request)
|
def handle_request(request)
|
||||||
|
|
@ -24,7 +26,7 @@ module MCP
|
||||||
when "DELETE"
|
when "DELETE"
|
||||||
handle_delete(request)
|
handle_delete(request)
|
||||||
else
|
else
|
||||||
[405, { "Content-Type" => "application/json" }, [{ error: "Method not allowed" }.to_json]]
|
method_not_allowed_response
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
@ -35,6 +37,9 @@ module MCP
|
||||||
end
|
end
|
||||||
|
|
||||||
def send_notification(method, params = nil, session_id: nil)
|
def send_notification(method, params = nil, session_id: nil)
|
||||||
|
# Stateless mode doesn't support notifications
|
||||||
|
raise "Stateless mode does not support notifications" if @stateless
|
||||||
|
|
||||||
notification = {
|
notification = {
|
||||||
jsonrpc: "2.0",
|
jsonrpc: "2.0",
|
||||||
method:,
|
method:,
|
||||||
|
|
@ -119,6 +124,10 @@ module MCP
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_get(request)
|
def handle_get(request)
|
||||||
|
if @stateless
|
||||||
|
return method_not_allowed_response
|
||||||
|
end
|
||||||
|
|
||||||
session_id = extract_session_id(request)
|
session_id = extract_session_id(request)
|
||||||
|
|
||||||
return missing_session_id_response unless session_id
|
return missing_session_id_response unless session_id
|
||||||
|
|
@ -128,6 +137,13 @@ module MCP
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_delete(request)
|
def handle_delete(request)
|
||||||
|
success_response = [200, { "Content-Type" => "application/json" }, [{ success: true }.to_json]]
|
||||||
|
|
||||||
|
if @stateless
|
||||||
|
# Stateless mode doesn't support sessions, so we can just return a success response
|
||||||
|
return success_response
|
||||||
|
end
|
||||||
|
|
||||||
session_id = request.env["HTTP_MCP_SESSION_ID"]
|
session_id = request.env["HTTP_MCP_SESSION_ID"]
|
||||||
|
|
||||||
return [
|
return [
|
||||||
|
|
@ -137,7 +153,7 @@ module MCP
|
||||||
] unless session_id
|
] unless session_id
|
||||||
|
|
||||||
cleanup_session(session_id)
|
cleanup_session(session_id)
|
||||||
[200, { "Content-Type" => "application/json" }, [{ success: true }.to_json]]
|
success_response
|
||||||
end
|
end
|
||||||
|
|
||||||
def cleanup_session(session_id)
|
def cleanup_session(session_id)
|
||||||
|
|
@ -177,6 +193,9 @@ module MCP
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_initialization(body_string, body)
|
def handle_initialization(body_string, body)
|
||||||
|
session_id = nil
|
||||||
|
|
||||||
|
unless @stateless
|
||||||
session_id = SecureRandom.uuid
|
session_id = SecureRandom.uuid
|
||||||
|
|
||||||
@mutex.synchronize do
|
@mutex.synchronize do
|
||||||
|
|
@ -184,14 +203,16 @@ module MCP
|
||||||
stream: nil,
|
stream: nil,
|
||||||
}
|
}
|
||||||
end
|
end
|
||||||
|
end
|
||||||
|
|
||||||
response = @server.handle_json(body_string)
|
response = @server.handle_json(body_string)
|
||||||
|
|
||||||
headers = {
|
headers = {
|
||||||
"Content-Type" => "application/json",
|
"Content-Type" => "application/json",
|
||||||
"Mcp-Session-Id" => session_id,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
headers["Mcp-Session-Id"] = session_id if session_id
|
||||||
|
|
||||||
[200, headers, [response]]
|
[200, headers, [response]]
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
@ -200,21 +221,32 @@ module MCP
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_regular_request(body_string, session_id)
|
def handle_regular_request(body_string, session_id)
|
||||||
|
unless @stateless
|
||||||
# If session ID is provided, but not in the sessions hash, return an error
|
# If session ID is provided, but not in the sessions hash, return an error
|
||||||
if session_id && !@sessions.key?(session_id)
|
if session_id && !@sessions.key?(session_id)
|
||||||
return [400, { "Content-Type" => "application/json" }, [{ error: "Invalid session ID" }.to_json]]
|
return [400, { "Content-Type" => "application/json" }, [{ error: "Invalid session ID" }.to_json]]
|
||||||
end
|
end
|
||||||
|
end
|
||||||
|
|
||||||
response = @server.handle_json(body_string)
|
response = @server.handle_json(body_string) || ""
|
||||||
|
|
||||||
|
# Stream can be nil since stateless mode doesn't retain streams
|
||||||
stream = get_session_stream(session_id) if session_id
|
stream = get_session_stream(session_id) if session_id
|
||||||
|
|
||||||
if stream
|
if stream
|
||||||
send_response_to_stream(stream, response, session_id)
|
send_response_to_stream(stream, response, session_id)
|
||||||
|
elsif response.nil? && notification_request?(body_string)
|
||||||
|
[202, { "Content-Type" => "application/json" }, [response]]
|
||||||
else
|
else
|
||||||
[200, { "Content-Type" => "application/json" }, [response]]
|
[200, { "Content-Type" => "application/json" }, [response]]
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def notification_request?(body_string)
|
||||||
|
body = parse_request_body(body_string)
|
||||||
|
body.is_a?(Hash) && body["method"].start_with?("notifications/")
|
||||||
|
end
|
||||||
|
|
||||||
def get_session_stream(session_id)
|
def get_session_stream(session_id)
|
||||||
@mutex.synchronize { @sessions[session_id]&.fetch(:stream, nil) }
|
@mutex.synchronize { @sessions[session_id]&.fetch(:stream, nil) }
|
||||||
end
|
end
|
||||||
|
|
@ -236,6 +268,10 @@ module MCP
|
||||||
@mutex.synchronize { @sessions.key?(session_id) }
|
@mutex.synchronize { @sessions.key?(session_id) }
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def method_not_allowed_response
|
||||||
|
[405, { "Content-Type" => "application/json" }, [{ error: "Method not allowed" }.to_json]]
|
||||||
|
end
|
||||||
|
|
||||||
def missing_session_id_response
|
def missing_session_id_response
|
||||||
[400, { "Content-Type" => "application/json" }, [{ error: "Missing session ID" }.to_json]]
|
[400, { "Content-Type" => "application/json" }, [{ error: "Missing session ID" }.to_json]]
|
||||||
end
|
end
|
||||||
|
|
|
||||||
|
|
@ -573,6 +573,128 @@ module MCP
|
||||||
assert_equal "Method not allowed", body["error"]
|
assert_equal "Method not allowed", body["error"]
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "stateless mode allows requests without session IDs, responding with no session ID" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
init_request = create_rack_request(
|
||||||
|
"POST",
|
||||||
|
"/",
|
||||||
|
{ "CONTENT_TYPE" => "application/json" },
|
||||||
|
{ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
|
||||||
|
)
|
||||||
|
init_response = stateless_transport.handle_request(init_request)
|
||||||
|
assert_nil init_response[1]["Mcp-Session-Id"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode responds without any session ID when session ID is present" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
request = create_rack_request(
|
||||||
|
"POST",
|
||||||
|
"/",
|
||||||
|
{
|
||||||
|
"CONTENT_TYPE" => "application/json",
|
||||||
|
"HTTP_MCP_SESSION_ID" => "unseen_session_id",
|
||||||
|
},
|
||||||
|
{ jsonrpc: "2.0", method: "ping", id: "123" }.to_json,
|
||||||
|
)
|
||||||
|
|
||||||
|
response = stateless_transport.handle_request(request)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
assert_equal(
|
||||||
|
{
|
||||||
|
"Content-Type" => "application/json",
|
||||||
|
},
|
||||||
|
response[1],
|
||||||
|
)
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert_equal "2.0", body["jsonrpc"]
|
||||||
|
assert_equal "123", body["id"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode responds with 405 when SSE is requested" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
get_request = create_rack_request(
|
||||||
|
"GET",
|
||||||
|
"/",
|
||||||
|
{
|
||||||
|
"CONTENT_TYPE" => "application/json,text/event-stream",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
response = stateless_transport.handle_request(get_request)
|
||||||
|
assert_equal 405, response[0]
|
||||||
|
assert_equal({ "Content-Type" => "application/json" }, response[1])
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert_equal "Method not allowed", body["error"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode silently responds with success to session DELETE when session ID is not present" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
delete_request = create_rack_request(
|
||||||
|
"DELETE",
|
||||||
|
"/",
|
||||||
|
{},
|
||||||
|
)
|
||||||
|
response = stateless_transport.handle_request(delete_request)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
assert_equal({ "Content-Type" => "application/json" }, response[1])
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert body["success"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode silently responds with success to session DELETE when session ID is provided" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
delete_request = create_rack_request(
|
||||||
|
"DELETE",
|
||||||
|
"/",
|
||||||
|
{ "HTTP_MCP_SESSION_ID" => "session_id" },
|
||||||
|
)
|
||||||
|
response = stateless_transport.handle_request(delete_request)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
assert_equal({ "Content-Type" => "application/json" }, response[1])
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert body["success"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode does not support server-sent events" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
e = assert_raises(RuntimeError) do
|
||||||
|
stateless_transport.send_notification(
|
||||||
|
"test_notification",
|
||||||
|
{ message: "Hello" },
|
||||||
|
session_id: "some_session_id",
|
||||||
|
)
|
||||||
|
end
|
||||||
|
|
||||||
|
assert_equal("Stateless mode does not support notifications", e.message)
|
||||||
|
end
|
||||||
|
|
||||||
|
test "stateless mode responds with 202 when client sends a notification/initialized request" do
|
||||||
|
stateless_transport = StreamableHTTPTransport.new(@server, stateless: true)
|
||||||
|
|
||||||
|
request = create_rack_request(
|
||||||
|
"POST",
|
||||||
|
"/",
|
||||||
|
{ "CONTENT_TYPE" => "application/json" },
|
||||||
|
{ jsonrpc: "2.0", method: "notifications/initialized" }.to_json,
|
||||||
|
)
|
||||||
|
|
||||||
|
response = stateless_transport.handle_request(request)
|
||||||
|
assert_equal 202, response[0]
|
||||||
|
assert_empty(response[1])
|
||||||
|
|
||||||
|
body = response[2][0]
|
||||||
|
assert_nil(body)
|
||||||
|
end
|
||||||
|
|
||||||
test "handle post request with a standard error" do
|
test "handle post request with a standard error" do
|
||||||
request = create_rack_request(
|
request = create_rack_request(
|
||||||
"POST",
|
"POST",
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue