Merge pull request #263 from koic/make_streamable_http_transport_a_standard_rack_app
Make `StreamableHTTPTransport` a Rack application
This commit is contained in:
commit
66d0722e41
6 changed files with 234 additions and 106 deletions
|
|
@ -116,6 +116,6 @@ This is the official Ruby SDK for the Model Context Protocol (MCP), implementing
|
||||||
|
|
||||||
### Integration patterns
|
### Integration patterns
|
||||||
|
|
||||||
- **Rails controllers**: Use `server.handle_json(request.body.read)` for HTTP endpoints
|
- **Rails/Rack apps**: Mount `StreamableHTTPTransport` as a Rack app (e.g., `mount transport => "/mcp"`)
|
||||||
- **Command-line tools**: Use `StdioTransport.new(server).open` for CLI applications
|
- **Command-line tools**: Use `StdioTransport.new(server).open` for CLI applications
|
||||||
- **HTTP services**: Use `StreamableHttpTransport` for web-based servers
|
- **HTTP services**: Use `StreamableHTTPTransport` for web-based servers
|
||||||
|
|
|
||||||
31
README.md
31
README.md
|
|
@ -104,14 +104,33 @@ $ ruby examples/stdio_server.rb
|
||||||
{"jsonrpc":"2.0","id":"3","method":"tools/call","params":{"name":"example_tool","arguments":{"message":"Hello"}}}
|
{"jsonrpc":"2.0","id":"3","method":"tools/call","params":{"name":"example_tool","arguments":{"message":"Hello"}}}
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Rails Controller
|
#### Rails (mount)
|
||||||
|
|
||||||
When added to a Rails controller on a route that handles POST requests, your server will be compliant with non-streaming
|
`StreamableHTTPTransport` is a Rack app that can be mounted directly in Rails routes:
|
||||||
[Streamable HTTP](https://modelcontextprotocol.io/specification/latest/basic/transports#streamable-http) transport
|
|
||||||
requests.
|
|
||||||
|
|
||||||
You can use `StreamableHTTPTransport#handle_request` to handle requests with proper HTTP
|
```ruby
|
||||||
status codes (e.g., 202 Accepted for notifications).
|
# config/routes.rb
|
||||||
|
server = MCP::Server.new(
|
||||||
|
name: "my_server",
|
||||||
|
title: "Example Server Display Name",
|
||||||
|
version: "1.0.0",
|
||||||
|
instructions: "Use the tools of this server as a last resort",
|
||||||
|
tools: [SomeTool, AnotherTool],
|
||||||
|
prompts: [MyPrompt],
|
||||||
|
)
|
||||||
|
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
||||||
|
|
||||||
|
Rails.application.routes.draw do
|
||||||
|
mount transport => "/mcp"
|
||||||
|
end
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Rails (controller)
|
||||||
|
|
||||||
|
While the mount approach creates a single server at boot time, the controller approach creates a new server per request.
|
||||||
|
This allows you to customize tools, prompts, or configuration based on the request (e.g., different tools per route).
|
||||||
|
|
||||||
|
`StreamableHTTPTransport#handle_request` returns proper HTTP status codes (e.g., 202 Accepted for notifications):
|
||||||
|
|
||||||
```ruby
|
```ruby
|
||||||
class McpController < ActionController::API
|
class McpController < ActionController::API
|
||||||
|
|
|
||||||
|
|
@ -95,52 +95,52 @@ end
|
||||||
# Create the Streamable HTTP transport
|
# Create the Streamable HTTP transport
|
||||||
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
||||||
|
|
||||||
# Create a logger for MCP-specific logging
|
# Rack middleware for MCP-specific request/response logging.
|
||||||
mcp_logger = Logger.new($stdout)
|
class McpRequestLogger
|
||||||
mcp_logger.formatter = proc do |_severity, _datetime, _progname, msg|
|
def initialize(app)
|
||||||
"[MCP] #{msg}\n"
|
@app = app
|
||||||
end
|
@logger = Logger.new($stdout)
|
||||||
|
@logger.formatter = proc { |_severity, _datetime, _progname, msg| "[MCP] #{msg}\n" }
|
||||||
# Create a Rack application with logging
|
|
||||||
app = proc do |env|
|
|
||||||
request = Rack::Request.new(env)
|
|
||||||
|
|
||||||
# Log MCP-specific details for POST requests
|
|
||||||
if request.post?
|
|
||||||
body = request.body.read
|
|
||||||
request.body.rewind
|
|
||||||
begin
|
|
||||||
parsed_body = JSON.parse(body)
|
|
||||||
mcp_logger.info("Request: #{parsed_body["method"]} (id: #{parsed_body["id"]})")
|
|
||||||
mcp_logger.debug("Request body: #{JSON.pretty_generate(parsed_body)}")
|
|
||||||
rescue JSON::ParserError
|
|
||||||
mcp_logger.warn("Request body (raw): #{body}")
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
# Handle the request
|
def call(env)
|
||||||
response = transport.handle_request(request)
|
if env["REQUEST_METHOD"] == "POST"
|
||||||
|
body = env["rack.input"].read
|
||||||
|
env["rack.input"].rewind
|
||||||
|
|
||||||
# Log the MCP response details
|
begin
|
||||||
_, _, body = response
|
parsed = JSON.parse(body)
|
||||||
if body.is_a?(Array) && !body.empty? && body.first
|
|
||||||
begin
|
@logger.info("Request: #{parsed["method"]} (id: #{parsed["id"]})")
|
||||||
parsed_response = JSON.parse(body.first)
|
@logger.debug("Request body: #{JSON.pretty_generate(parsed)}")
|
||||||
if parsed_response["error"]
|
rescue JSON::ParserError
|
||||||
mcp_logger.error("Response error: #{parsed_response["error"]["message"]}")
|
@logger.warn("Request body (raw): #{body}")
|
||||||
else
|
|
||||||
mcp_logger.info("Response: #{parsed_response["result"] ? "success" : "empty"} (id: #{parsed_response["id"]})")
|
|
||||||
end
|
end
|
||||||
mcp_logger.debug("Response body: #{JSON.pretty_generate(parsed_response)}")
|
|
||||||
rescue JSON::ParserError
|
|
||||||
mcp_logger.warn("Response body (raw): #{body}")
|
|
||||||
end
|
end
|
||||||
end
|
|
||||||
|
|
||||||
response
|
status, headers, response_body = @app.call(env)
|
||||||
|
|
||||||
|
if response_body.is_a?(Array) && !response_body.empty? && response_body.first
|
||||||
|
begin
|
||||||
|
parsed = JSON.parse(response_body.first)
|
||||||
|
|
||||||
|
if parsed["error"]
|
||||||
|
@logger.error("Response error: #{parsed["error"]["message"]}")
|
||||||
|
else
|
||||||
|
@logger.info("Response: #{parsed["result"] ? "success" : "empty"} (id: #{parsed["id"]})")
|
||||||
|
end
|
||||||
|
@logger.debug("Response body: #{JSON.pretty_generate(parsed)}")
|
||||||
|
rescue JSON::ParserError
|
||||||
|
@logger.warn("Response body (raw): #{response_body}")
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
[status, headers, response_body]
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
# Wrap the app with Rack middleware
|
# Build the Rack application with middleware.
|
||||||
|
# `StreamableHTTPTransport` responds to `call(env)`, so it can be used directly as a Rack app.
|
||||||
rack_app = Rack::Builder.new do
|
rack_app = Rack::Builder.new do
|
||||||
# Enable CORS to allow browser-based MCP clients (e.g., MCP Inspector)
|
# Enable CORS to allow browser-based MCP clients (e.g., MCP Inspector)
|
||||||
# WARNING: origins("*") allows all origins. Restrict this in production.
|
# WARNING: origins("*") allows all origins. Restrict this in production.
|
||||||
|
|
@ -158,11 +158,10 @@ rack_app = Rack::Builder.new do
|
||||||
|
|
||||||
# Use CommonLogger for standard HTTP request logging
|
# Use CommonLogger for standard HTTP request logging
|
||||||
use(Rack::CommonLogger, Logger.new($stdout))
|
use(Rack::CommonLogger, Logger.new($stdout))
|
||||||
|
|
||||||
# Add other useful middleware
|
|
||||||
use(Rack::ShowExceptions)
|
use(Rack::ShowExceptions)
|
||||||
|
use(McpRequestLogger)
|
||||||
|
|
||||||
run(app)
|
run(transport)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Start the server
|
# Start the server
|
||||||
|
|
|
||||||
|
|
@ -64,70 +64,62 @@ end
|
||||||
# Create the Streamable HTTP transport
|
# Create the Streamable HTTP transport
|
||||||
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
|
||||||
|
|
||||||
# Create a logger for MCP request/response logging
|
# Rack middleware for MCP request/response and SSE logging.
|
||||||
mcp_logger = Logger.new($stdout)
|
class McpSseLogger
|
||||||
mcp_logger.formatter = proc do |_severity, _datetime, _progname, msg|
|
def initialize(app)
|
||||||
"[MCP] #{msg}\n"
|
@app = app
|
||||||
end
|
|
||||||
|
|
||||||
# Create the Rack application
|
@mcp_logger = Logger.new($stdout)
|
||||||
app = proc do |env|
|
@mcp_logger.formatter = proc { |_severity, _datetime, _progname, msg| "[MCP] #{msg}\n" }
|
||||||
request = Rack::Request.new(env)
|
|
||||||
|
|
||||||
# Log request details
|
@sse_logger = Logger.new($stdout)
|
||||||
if request.post?
|
@sse_logger.formatter = proc { |severity, datetime, _progname, msg| "[SSE] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n" }
|
||||||
body = request.body.read
|
|
||||||
request.body.rewind
|
|
||||||
begin
|
|
||||||
parsed_body = JSON.parse(body)
|
|
||||||
mcp_logger.info("Request: #{parsed_body["method"]} (id: #{parsed_body["id"]})")
|
|
||||||
|
|
||||||
# Log SSE-specific setup
|
|
||||||
if parsed_body["method"] == "initialize"
|
|
||||||
sse_logger.info("New client initializing session")
|
|
||||||
end
|
|
||||||
rescue JSON::ParserError
|
|
||||||
mcp_logger.warn("Invalid JSON in request")
|
|
||||||
end
|
|
||||||
elsif request.get?
|
|
||||||
session_id = request.env["HTTP_MCP_SESSION_ID"] ||
|
|
||||||
Rack::Utils.parse_query(request.env["QUERY_STRING"])["sessionId"]
|
|
||||||
sse_logger.info("SSE connection request for session: #{session_id}")
|
|
||||||
end
|
end
|
||||||
|
|
||||||
# Handle the request
|
def call(env)
|
||||||
response = transport.handle_request(request)
|
if env["REQUEST_METHOD"] == "POST"
|
||||||
|
body = env["rack.input"].read
|
||||||
|
env["rack.input"].rewind
|
||||||
|
|
||||||
# Log response details
|
begin
|
||||||
status, headers, body = response
|
parsed = JSON.parse(body)
|
||||||
if body.is_a?(Array) && !body.empty? && request.post?
|
|
||||||
begin
|
|
||||||
parsed_response = JSON.parse(body.first)
|
|
||||||
if parsed_response["error"]
|
|
||||||
mcp_logger.error("Response error: #{parsed_response["error"]["message"]}")
|
|
||||||
elsif parsed_response["accepted"]
|
|
||||||
# Response was sent via SSE
|
|
||||||
server.notify_log_message(data: { details: "Response accepted and sent via SSE" }, level: "info")
|
|
||||||
sse_logger.info("Response sent via SSE stream")
|
|
||||||
else
|
|
||||||
mcp_logger.info("Response: success (id: #{parsed_response["id"]})")
|
|
||||||
|
|
||||||
# Log session ID for initialization
|
@mcp_logger.info("Request: #{parsed["method"]} (id: #{parsed["id"]})")
|
||||||
if headers["Mcp-Session-Id"]
|
@sse_logger.info("New client initializing session") if parsed["method"] == "initialize"
|
||||||
sse_logger.info("Session created: #{headers["Mcp-Session-Id"]}")
|
rescue JSON::ParserError
|
||||||
|
@mcp_logger.warn("Invalid JSON in request")
|
||||||
|
end
|
||||||
|
elsif env["REQUEST_METHOD"] == "GET"
|
||||||
|
session_id = env["HTTP_MCP_SESSION_ID"] || Rack::Utils.parse_query(env["QUERY_STRING"])["sessionId"]
|
||||||
|
|
||||||
|
@sse_logger.info("SSE connection request for session: #{session_id}")
|
||||||
|
end
|
||||||
|
|
||||||
|
status, headers, response_body = @app.call(env)
|
||||||
|
|
||||||
|
if response_body.is_a?(Array) && !response_body.empty? && env["REQUEST_METHOD"] == "POST"
|
||||||
|
begin
|
||||||
|
parsed = JSON.parse(response_body.first)
|
||||||
|
|
||||||
|
if parsed["error"]
|
||||||
|
@mcp_logger.error("Response error: #{parsed["error"]["message"]}")
|
||||||
|
else
|
||||||
|
@mcp_logger.info("Response: success (id: #{parsed["id"]})")
|
||||||
|
@sse_logger.info("Session created: #{headers["Mcp-Session-Id"]}") if headers["Mcp-Session-Id"]
|
||||||
end
|
end
|
||||||
|
rescue JSON::ParserError
|
||||||
|
@mcp_logger.warn("Invalid JSON in response")
|
||||||
end
|
end
|
||||||
rescue JSON::ParserError
|
elsif env["REQUEST_METHOD"] == "GET" && status == 200
|
||||||
mcp_logger.warn("Invalid JSON in response")
|
@sse_logger.info("SSE stream established")
|
||||||
end
|
end
|
||||||
elsif request.get? && status == 200
|
|
||||||
sse_logger.info("SSE stream established")
|
|
||||||
end
|
|
||||||
|
|
||||||
response
|
[status, headers, response_body]
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
# Build the Rack application with middleware
|
# Build the Rack application with middleware.
|
||||||
|
# `StreamableHTTPTransport` responds to `call(env)`, so it can be used directly as a Rack app.
|
||||||
rack_app = Rack::Builder.new do
|
rack_app = Rack::Builder.new do
|
||||||
# Enable CORS to allow browser-based MCP clients (e.g., MCP Inspector)
|
# Enable CORS to allow browser-based MCP clients (e.g., MCP Inspector)
|
||||||
# WARNING: origins("*") allows all origins. Restrict this in production.
|
# WARNING: origins("*") allows all origins. Restrict this in production.
|
||||||
|
|
@ -145,7 +137,9 @@ rack_app = Rack::Builder.new do
|
||||||
|
|
||||||
use(Rack::CommonLogger, Logger.new($stdout))
|
use(Rack::CommonLogger, Logger.new($stdout))
|
||||||
use(Rack::ShowExceptions)
|
use(Rack::ShowExceptions)
|
||||||
run(app)
|
use(McpSseLogger)
|
||||||
|
|
||||||
|
run(transport)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Print usage instructions
|
# Print usage instructions
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,15 @@
|
||||||
require "json"
|
require "json"
|
||||||
require_relative "../../transport"
|
require_relative "../../transport"
|
||||||
|
|
||||||
|
# This file is autoloaded only when `StreamableHTTPTransport` is referenced,
|
||||||
|
# so the `rack` dependency does not affect `StdioTransport` users.
|
||||||
|
begin
|
||||||
|
require "rack"
|
||||||
|
rescue LoadError
|
||||||
|
raise LoadError, "The 'rack' gem is required to use the StreamableHTTPTransport. " \
|
||||||
|
"Add it to your Gemfile: gem 'rack'"
|
||||||
|
end
|
||||||
|
|
||||||
module MCP
|
module MCP
|
||||||
class Server
|
class Server
|
||||||
module Transports
|
module Transports
|
||||||
|
|
@ -39,6 +48,11 @@ module MCP
|
||||||
STREAM_WRITE_ERRORS = [IOError, Errno::EPIPE, Errno::ECONNRESET].freeze
|
STREAM_WRITE_ERRORS = [IOError, Errno::EPIPE, Errno::ECONNRESET].freeze
|
||||||
SESSION_REAP_INTERVAL = 60
|
SESSION_REAP_INTERVAL = 60
|
||||||
|
|
||||||
|
# Rack app interface. This transport can be mounted as a Rack app.
|
||||||
|
def call(env)
|
||||||
|
handle_request(Rack::Request.new(env))
|
||||||
|
end
|
||||||
|
|
||||||
def handle_request(request)
|
def handle_request(request)
|
||||||
case request.env["REQUEST_METHOD"]
|
case request.env["REQUEST_METHOD"]
|
||||||
when "POST"
|
when "POST"
|
||||||
|
|
@ -546,7 +560,7 @@ module MCP
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
[200, SSE_HEADERS, body]
|
[200, SSE_HEADERS.dup, body]
|
||||||
end
|
end
|
||||||
|
|
||||||
# Returns the SSE stream available for server-to-client messages.
|
# Returns the SSE stream available for server-to-client messages.
|
||||||
|
|
@ -628,7 +642,7 @@ module MCP
|
||||||
def setup_sse_stream(session_id)
|
def setup_sse_stream(session_id)
|
||||||
body = create_sse_body(session_id)
|
body = create_sse_body(session_id)
|
||||||
|
|
||||||
[200, SSE_HEADERS, body]
|
[200, SSE_HEADERS.dup, body]
|
||||||
end
|
end
|
||||||
|
|
||||||
def create_sse_body(session_id)
|
def create_sse_body(session_id)
|
||||||
|
|
|
||||||
|
|
@ -2905,6 +2905,108 @@ module MCP
|
||||||
assert session2_logging.should_notify?("debug")
|
assert session2_logging.should_notify?("debug")
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "call(env) works as a Rack app for POST requests" do
|
||||||
|
env = {
|
||||||
|
"REQUEST_METHOD" => "POST",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new({ jsonrpc: "2.0", method: "initialize", id: "init-1" }.to_json),
|
||||||
|
"CONTENT_TYPE" => "application/json",
|
||||||
|
"HTTP_ACCEPT" => "application/json, text/event-stream",
|
||||||
|
}
|
||||||
|
|
||||||
|
response = @transport.call(env)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
assert_equal "application/json", response[1]["Content-Type"]
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert_equal "2.0", body["jsonrpc"]
|
||||||
|
assert_equal "init-1", body["id"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "call(env) returns 405 for unsupported HTTP methods" do
|
||||||
|
env = {
|
||||||
|
"REQUEST_METHOD" => "PUT",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new(""),
|
||||||
|
}
|
||||||
|
|
||||||
|
response = @transport.call(env)
|
||||||
|
assert_equal 405, response[0]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "call(env) handles GET SSE stream request" do
|
||||||
|
init_env = {
|
||||||
|
"REQUEST_METHOD" => "POST",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new({ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json),
|
||||||
|
"CONTENT_TYPE" => "application/json",
|
||||||
|
"HTTP_ACCEPT" => "application/json, text/event-stream",
|
||||||
|
}
|
||||||
|
init_response = @transport.call(init_env)
|
||||||
|
session_id = init_response[1]["Mcp-Session-Id"]
|
||||||
|
|
||||||
|
get_env = {
|
||||||
|
"REQUEST_METHOD" => "GET",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new(""),
|
||||||
|
"HTTP_ACCEPT" => "text/event-stream",
|
||||||
|
"HTTP_MCP_SESSION_ID" => session_id,
|
||||||
|
}
|
||||||
|
|
||||||
|
response = @transport.call(get_env)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
assert_equal "text/event-stream", response[1]["Content-Type"]
|
||||||
|
assert response[2].is_a?(Proc)
|
||||||
|
end
|
||||||
|
|
||||||
|
test "call(env) handles DELETE session request" do
|
||||||
|
init_env = {
|
||||||
|
"REQUEST_METHOD" => "POST",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new({ jsonrpc: "2.0", method: "initialize", id: "init" }.to_json),
|
||||||
|
"CONTENT_TYPE" => "application/json",
|
||||||
|
"HTTP_ACCEPT" => "application/json, text/event-stream",
|
||||||
|
}
|
||||||
|
init_response = @transport.call(init_env)
|
||||||
|
session_id = init_response[1]["Mcp-Session-Id"]
|
||||||
|
|
||||||
|
delete_env = {
|
||||||
|
"REQUEST_METHOD" => "DELETE",
|
||||||
|
"PATH_INFO" => "/",
|
||||||
|
"rack.input" => StringIO.new(""),
|
||||||
|
"HTTP_MCP_SESSION_ID" => session_id,
|
||||||
|
}
|
||||||
|
|
||||||
|
response = @transport.call(delete_env)
|
||||||
|
assert_equal 200, response[0]
|
||||||
|
|
||||||
|
body = JSON.parse(response[2][0])
|
||||||
|
assert body["success"]
|
||||||
|
end
|
||||||
|
|
||||||
|
test "SSE response headers are not frozen so Rack middleware can modify them" do
|
||||||
|
init_request = create_rack_request(
|
||||||
|
"POST",
|
||||||
|
"/",
|
||||||
|
{ "CONTENT_TYPE" => "application/json" },
|
||||||
|
{ jsonrpc: "2.0", method: "initialize", id: "init-frozen" }.to_json,
|
||||||
|
)
|
||||||
|
init_response = @transport.handle_request(init_request)
|
||||||
|
session_id = init_response[1]["Mcp-Session-Id"]
|
||||||
|
|
||||||
|
request = create_rack_request(
|
||||||
|
"POST",
|
||||||
|
"/",
|
||||||
|
{ "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
|
||||||
|
{ jsonrpc: "2.0", method: "tools/list", id: "sse-headers-1" }.to_json,
|
||||||
|
)
|
||||||
|
|
||||||
|
status, headers, = @transport.handle_request(request)
|
||||||
|
assert_equal 200, status
|
||||||
|
assert_equal "text/event-stream", headers["Content-Type"]
|
||||||
|
refute headers.frozen?, "SSE response headers should not be frozen"
|
||||||
|
end
|
||||||
|
|
||||||
private
|
private
|
||||||
|
|
||||||
def create_rack_request(method, path, headers, body = nil)
|
def create_rack_request(method, path, headers, body = nil)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue