Addressing feedback

examples/streaming_http_test_client.rb

@@ -0,0 +1,207 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require "net/http"
+require "uri"
+require "json"
+require "logger"
+
+# Logger for client operations
+logger = Logger.new($stdout)
+logger.formatter = proc do |severity, datetime, _progname, msg|
+  "[CLIENT] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n"
+end
+
+# Server configuration
+SERVER_URL = "http://localhost:9393/mcp"
+PROTOCOL_VERSION = "2024-11-05"
+
+# Helper method to make JSON-RPC requests
+def make_request(session_id, method, params = {}, id = nil)
+  uri = URI(SERVER_URL)
+  http = Net::HTTP.new(uri.host, uri.port)
+
+  request = Net::HTTP::Post.new(uri)
+  request["Content-Type"] = "application/json"
+  request["Mcp-Session-Id"] = session_id if session_id
+
+  body = {
+    jsonrpc: "2.0",
+    method: method,
+    params: params,
+    id: id || SecureRandom.uuid,
+  }
+
+  request.body = body.to_json
+  response = http.request(request)
+
+  {
+    status: response.code,
+    headers: response.to_hash,
+    body: JSON.parse(response.body),
+  }
+rescue => e
+  { error: e.message }
+end
+
+# Connect to SSE stream
+def connect_sse(session_id, logger)
+  uri = URI(SERVER_URL)
+
+  logger.info("Connecting to SSE stream...")
+
+  Net::HTTP.start(uri.host, uri.port) do |http|
+    request = Net::HTTP::Get.new(uri)
+    request["Mcp-Session-Id"] = session_id
+    request["Accept"] = "text/event-stream"
+    request["Cache-Control"] = "no-cache"
+
+    http.request(request) do |response|
+      if response.code == "200"
+        logger.info("SSE stream connected successfully")
+
+        response.read_body do |chunk|
+          chunk.split("\n").each do |line|
+            if line.start_with?("data: ")
+              data = line[6..-1]
+              begin
+                logger.info("SSE data: #{data}")
+              rescue JSON::ParserError
+                logger.debug("Non-JSON SSE data: #{data}")
+              end
+            elsif line.start_with?(": ")
+              logger.debug("SSE keepalive received: #{line}")
+            end
+          end
+        end
+      else
+        logger.error("Failed to connect to SSE: #{response.code} #{response.message}")
+      end
+    end
+  end
+rescue Interrupt
+  logger.info("SSE connection interrupted by user")
+rescue => e
+  logger.error("SSE connection error: #{e.message}")
+end
+
+# Main client flow
+def main
+  logger = Logger.new($stdout)
+  logger.formatter = proc do |severity, datetime, _progname, msg|
+    "[CLIENT] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n"
+  end
+
+  puts "=== MCP SSE Test Client ==="
+  puts ""
+
+  # Step 1: Initialize session
+  logger.info("Initializing session...")
+
+  init_response = make_request(
+    nil,
+    "initialize",
+    {
+      protocolVersion: PROTOCOL_VERSION,
+      capabilities: {},
+      clientInfo: {
+        name: "sse-test-client",
+        version: "1.0",
+      },
+    },
+    "init-1",
+  )
+
+  if init_response[:error]
+    logger.error("Failed to initialize: #{init_response[:error]}")
+    exit(1)
+  end
+
+  session_id = init_response[:headers]["mcp-session-id"]&.first
+
+  if session_id.nil?
+    logger.error("No session ID received")
+    exit(1)
+  end
+
+  logger.info("Session initialized: #{session_id}")
+  logger.info("Server info: #{init_response[:body]["result"]["serverInfo"]}")
+
+  # Step 2: Start SSE connection in a separate thread
+  sse_thread = Thread.new { connect_sse(session_id, logger) }
+
+  # Give SSE time to connect
+  sleep(1)
+
+  # Step 3: Interactive menu
+  loop do
+    puts "\n=== Available Actions ==="
+    puts "1. Send custom notification"
+    puts "2. Test echo"
+    puts "3. List tools"
+    puts "0. Exit"
+    puts ""
+    print("Choose an action: ")
+
+    choice = gets.chomp
+
+    case choice
+    when "1"
+      print("Enter notification message: ")
+      message = gets.chomp
+      print("Enter delay in seconds (0 for immediate): ")
+      delay = gets.chomp.to_f
+
+      response = make_request(
+        session_id,
+        "tools/call",
+        {
+          name: "notification_tool",
+          arguments: {
+            message: message,
+            delay: delay,
+          },
+        },
+      )
+
+      if response[:body]["result"]
+        logger.info("Notification tool response: #{response[:body]["result"]["content"]}")
+      else
+        logger.error("Error: #{response[:body]["error"]}")
+      end
+
+    when "2"
+      print("Enter message to echo: ")
+      message = gets.chomp
+      make_request(session_id, "tools/call", { name: "echo", arguments: { message: message } })
+
+    when "3"
+      make_request(session_id, "tools/list")
+
+    when "0"
+      logger.info("Exiting...")
+      break
+
+    else
+      puts "Invalid choice"
+    end
+  end
+
+  # Clean up
+  sse_thread.kill if sse_thread.alive?
+
+  # Close session
+  logger.info("Closing session...")
+  make_request(session_id, "close")
+  logger.info("Session closed")
+rescue Interrupt
+  logger.info("Client interrupted by user")
+rescue => e
+  logger.error("Client error: #{e.message}")
+  logger.error(e.backtrace.join("\n"))
+end
+
+# Run the client
+if __FILE__ == $PROGRAM_NAME
+  main
+end

examples/streaming_http_test_server.rb

@@ -0,0 +1,173 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift(File.expand_path("../lib", __dir__))
+require "mcp"
+require "mcp/server/transports/streamable_http_transport"
+require "rack"
+require "rackup"
+require "json"
+require "logger"
+
+# Create a logger for SSE-specific logging
+sse_logger = Logger.new($stdout)
+sse_logger.formatter = proc do |severity, datetime, _progname, msg|
+  "[SSE] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n"
+end
+
+# Tool that returns a response that will be sent via SSE if a stream is active
+class NotificationTool < MCP::Tool
+  tool_name "notification_tool"
+  description "Returns a notification message that will be sent via SSE if stream is active"
+  input_schema(
+    properties: {
+      message: { type: "string", description: "Message to send via SSE" },
+      delay: { type: "number", description: "Delay in seconds before returning (optional)" },
+    },
+    required: ["message"],
+  )
+
+  class << self
+    attr_accessor :logger
+
+    def call(message:, delay: 0)
+      sleep(delay) if delay > 0
+
+      logger&.info("Returning notification message: #{message}")
+
+      MCP::Tool::Response.new([{
+        type: "text",
+        text: "Notification: #{message} (timestamp: #{Time.now.iso8601})",
+      }])
+    end
+  end
+end
+
+# Create the server
+server = MCP::Server.new(
+  name: "sse_test_server",
+  tools: [NotificationTool],
+  prompts: [],
+  resources: [],
+)
+
+# Set logger for tools
+NotificationTool.logger = sse_logger
+
+# Add a simple echo tool for basic testing
+server.define_tool(
+  name: "echo",
+  description: "Simple echo tool",
+  input_schema: { properties: { message: { type: "string" } }, required: ["message"] },
+) do |message:|
+  MCP::Tool::Response.new([{ type: "text", text: "Echo: #{message}" }])
+end
+
+# Create the Streamable HTTP transport
+transport = MCP::Server::Transports::StreamableHTTPTransport.new(server)
+server.transport = transport
+
+# Create a logger for MCP request/response logging
+mcp_logger = Logger.new($stdout)
+mcp_logger.formatter = proc do |_severity, _datetime, _progname, msg|
+  "[MCP] #{msg}\n"
+end
+
+# Create the Rack application
+app = proc do |env|
+  request = Rack::Request.new(env)
+
+  # Log request details
+  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"]})")
+
+      # 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
+
+  # Handle the request
+  response = transport.handle_request(request)
+
+  # Log response details
+  status, headers, body = response
+  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
+        sse_logger.info("Response sent via SSE stream")
+      else
+        mcp_logger.info("Response: success (id: #{parsed_response["id"]})")
+
+        # Log session ID for initialization
+        if headers["Mcp-Session-Id"]
+          sse_logger.info("Session created: #{headers["Mcp-Session-Id"]}")
+        end
+      end
+    rescue JSON::ParserError
+      mcp_logger.warn("Invalid JSON in response")
+    end
+  elsif request.get? && status == 200
+    sse_logger.info("SSE stream established")
+  end
+
+  response
+end
+
+# Build the Rack application with middleware
+rack_app = Rack::Builder.new do
+  use(Rack::CommonLogger, Logger.new($stdout))
+  use(Rack::ShowExceptions)
+  run(app)
+end
+
+# Print usage instructions
+puts "=== MCP Streaming HTTP Test Server ==="
+puts ""
+puts "Starting server on http://localhost:9393"
+puts ""
+puts "Available Tools:"
+puts "1. NotificationTool - Returns messages that are sent via SSE when stream is active"
+puts "2. echo - Simple echo tool"
+puts ""
+puts "Testing SSE:"
+puts ""
+puts "1. Initialize session:"
+puts '   curl -X POST http://localhost:9393 -H "Content-Type: application/json" \\'
+puts '     -d \'{"jsonrpc":"2.0","method":"initialize","id":1,"params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"sse-test","version":"1.0"}}}\''
+puts ""
+puts "2. Connect SSE stream (use the session ID from step 1):"
+puts '   curl -N -H "Mcp-Session-Id: YOUR_SESSION_ID" http://localhost:9393'
+puts ""
+puts "3. In another terminal, test tools (responses will be sent via SSE if stream is active):"
+puts ""
+puts "   Echo tool:"
+puts '   curl -X POST http://localhost:9393 -H "Content-Type: application/json" -H "Mcp-Session-Id: YOUR_SESSION_ID" \\'
+puts '     -d \'{"jsonrpc":"2.0","method":"tools/call","id":2,"params":{"name":"echo","arguments":{"message":"Hello SSE!"}}}\''
+puts ""
+puts "   Notification tool (with 2 second delay):"
+puts '   curl -X POST http://localhost:9393 -H "Content-Type: application/json" -H "Mcp-Session-Id: YOUR_SESSION_ID" \\'
+puts '     -d \'{"jsonrpc":"2.0","method":"tools/call","id":3,"params":{"name":"notification_tool","arguments":{"message":"Hello SSE!", "delay": 2}}}\''
+puts ""
+puts "Note: When an SSE stream is active, tool responses will appear in the SSE stream and the POST request will return {\"accepted\": true}"
+puts ""
+puts "Press Ctrl+C to stop the server"
+puts ""
+
+# Start the server
+Rackup::Handler.get("puma").run(rack_app, Port: 9393, Host: "localhost")

lib/mcp/server/transports/stdio_transport.rb

@@ -8,11 +8,6 @@ module MCP
     module Transports
       class StdioTransport < Transport
         STATUS_INTERRUPTED = Signal.list["INT"] + 128
-        class << self
-          def register
-            super("stdio", self)
-          end
-        end
 
         def initialize(server)
           @server = server

lib/mcp/server/transports/streamable_http_transport.rb

@@ -8,12 +8,6 @@ module MCP
   class Server
     module Transports
       class StreamableHTTPTransport < Transport
-        class << self
-          def register
-            super("http", self)
-          end
-        end
-
         def initialize(server)
           super
           # { session_id => { stream: stream_object }

lib/mcp/transport.rb

@@ -2,19 +2,6 @@
 
 module MCP
   class Transport
-    class << self
-      def register(name, transport_class)
-        @transports ||= {}
-        @transports[name] = transport_class
-      end
-
-      def get(name)
-        raise "Transport #{name} not found" unless @transports
-
-        @transports.fetch(name) { raise "Transport #{name} not found" }
-      end
-    end
-
     # Initialize the transport with the server instance
     def initialize(server)
       @server = server

test/mcp/server_notification_test.rb

@@ -126,11 +126,5 @@ module MCP
       assert_equal Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED, notifications[1][:method]
       assert_equal Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED, notifications[2][:method]
     end
-
-    test "notification method constants are defined correctly" do
-      assert_equal "notifications/tools/list_changed", Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED
-      assert_equal "notifications/prompts/list_changed", Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED
-      assert_equal "notifications/resources/list_changed", Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED
-    end
   end
 end

test/mcp/transport_test.rb

@@ -28,22 +28,6 @@ module MCP
         resources: [],
       )
       @transport = TestTransport.new(@server)
-      # Clear transports registry before each test
-      Transport.instance_variable_set(:@transports, nil)
-    end
-
-    teardown do
-      # Clear transports registry after each test
-      Transport.instance_variable_set(:@transports, nil)
-    end
-
-    test "registers transport class" do
-      Transport.register("test", TestTransport)
-      assert_equal TestTransport, Transport.get("test")
-    end
-
-    test "raises error for unknown transport" do
-      assert_raises(RuntimeError, "Transport unknown not found") { Transport.get("unknown") }
     end
 
     test "initializes with server instance" do