diff --git a/README.md b/README.md index bbccbe1..eb7e3ed 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,7 @@ It implements the Model Context Protocol specification, handling model context r - Supports prompt registration and execution - Supports resource registration and retrieval - Supports stdio & Streamable HTTP (including SSE) transports +- Supports notifications for list changes (tools, prompts, resources) ### Supported Methods - `initialize` - Initializes the protocol and returns server capabilities @@ -46,13 +47,46 @@ It implements the Model Context Protocol specification, handling model context r - `resources/read` - Retrieves a specific resource by name - `resources/templates/list` - Lists all registered resource templates and their schemas +### Notifications + +The server supports sending notifications to clients when lists of tools, prompts, or resources change. This enables real-time updates without polling. + +#### Notification Methods + +The server provides three notification methods: +- `notify_tools_list_changed()` - Send a notification when the tools list changes +- `notify_prompts_list_changed()` - Send a notification when the prompts list changes +- `notify_resources_list_changed()` - Send a notification when the resources list changes + +#### Notification Format + +Notifications follow the JSON-RPC 2.0 specification and use these method names: +- `notifications/tools/list_changed` +- `notifications/prompts/list_changed` +- `notifications/resources/list_changed` + +#### Transport Support + +- **HTTP Transport**: Notifications are sent as Server-Sent Events (SSE) to all connected sessions +- **Stdio Transport**: Notifications are sent as JSON-RPC 2.0 messages to stdout + +#### Usage Example + +```ruby +server = MCP::Server.new(name: "my_server") +transport = MCP::Transports::HTTP.new(server) +server.transport = transport + +# When tools change, notify clients +server.define_tool(name: "new_tool") { |**args| { result: "ok" } } +server.notify_tools_list_changed() +``` + ### Unsupported Features ( to be implemented in future versions ) -- Notifications - Log Level - Resource subscriptions - Completions -- Complete Streamable HTTP implementation with streaming responses ### Usage diff --git a/lib/mcp/methods.rb b/lib/mcp/methods.rb index 4f0b0b2..9a2db26 100644 --- a/lib/mcp/methods.rb +++ b/lib/mcp/methods.rb @@ -21,6 +21,11 @@ module MCP SAMPLING_CREATE_MESSAGE = "sampling/createMessage" + # Notification methods + NOTIFICATIONS_TOOLS_LIST_CHANGED = "notifications/tools/list_changed" + NOTIFICATIONS_PROMPTS_LIST_CHANGED = "notifications/prompts/list_changed" + NOTIFICATIONS_RESOURCES_LIST_CHANGED = "notifications/resources/list_changed" + class MissingRequiredCapabilityError < StandardError attr_reader :method attr_reader :capability diff --git a/lib/mcp/server.rb b/lib/mcp/server.rb index a366e83..d246121 100644 --- a/lib/mcp/server.rb +++ b/lib/mcp/server.rb @@ -98,6 +98,30 @@ module MCP @prompts[prompt.name_value] = prompt end + def notify_tools_list_changed + return unless @transport + + @transport.send_notification(Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED) + rescue => e + report_exception(e, { notification: "tools_list_changed" }) + end + + def notify_prompts_list_changed + return unless @transport + + @transport.send_notification(Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED) + rescue => e + report_exception(e, { notification: "prompts_list_changed" }) + end + + def notify_resources_list_changed + return unless @transport + + @transport.send_notification(Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED) + rescue => e + report_exception(e, { notification: "resources_list_changed" }) + end + def resources_list_handler(&block) @capabilities.support_resources @handlers[Methods::RESOURCES_LIST] = block diff --git a/lib/mcp/server/transports/stdio_transport.rb b/lib/mcp/server/transports/stdio_transport.rb index fdf2610..53b2724 100644 --- a/lib/mcp/server/transports/stdio_transport.rb +++ b/lib/mcp/server/transports/stdio_transport.rb @@ -43,10 +43,18 @@ module MCP $stdout.flush end - def send_response(message) - json_message = message.is_a?(String) ? message : JSON.generate(message) - $stdout.puts(json_message) - $stdout.flush + def send_notification(method, params = nil) + notification = { + jsonrpc: "2.0", + method: method, + } + notification[:params] = params if params + + send_response(notification) + true + rescue => e + MCP.configuration.exception_reporter.call(e, { error: "Failed to send notification" }) + false end end end diff --git a/test/mcp/server/transports/stdio_notification_integration_test.rb b/test/mcp/server/transports/stdio_notification_integration_test.rb new file mode 100644 index 0000000..34516a0 --- /dev/null +++ b/test/mcp/server/transports/stdio_notification_integration_test.rb @@ -0,0 +1,249 @@ +# frozen_string_literal: true + +require "test_helper" +require "json" +require "mcp/server" +require "mcp/server/transports/stdio_transport" + +module MCP + class Server + module Transports + class StdioNotificationIntegrationTest < ActiveSupport::TestCase + class MockIO + attr_reader :output + + def initialize + @output = [] + @closed = false + end + + def puts(message) + @output << message + end + + def write(message) + @output << message + message.length + end + + def gets + nil # Simulate end of input + end + + def set_encoding(encoding) + # Mock implementation + end + + def flush + # Mock implementation + end + + def close + @closed = true + end + + def closed? + @closed + end + end + + setup do + @original_stdout = $stdout + @original_stdin = $stdin + + @mock_stdout = MockIO.new + @mock_stdin = MockIO.new + + $stdout = @mock_stdout + $stdin = @mock_stdin + + @server = Server.new( + name: "test_server", + tools: [], + prompts: [], + resources: [], + ) + @transport = StdioTransport.new(@server) + @server.transport = @transport + end + + teardown do + $stdout = @original_stdout + $stdin = @original_stdin + end + + test "server notification methods send JSON-RPC notifications through StdioTransport" do + # Test tools notification + @server.notify_tools_list_changed + + # Test prompts notification + @server.notify_prompts_list_changed + + # Test resources notification + @server.notify_resources_list_changed + + # Check the notifications were sent + assert_equal 3, @mock_stdout.output.size + + # Parse and verify each notification + notifications = @mock_stdout.output.map { |msg| JSON.parse(msg) } + + assert_equal "2.0", notifications[0]["jsonrpc"] + assert_equal Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED, notifications[0]["method"] + assert_nil notifications[0]["params"] + + assert_equal "2.0", notifications[1]["jsonrpc"] + assert_equal Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED, notifications[1]["method"] + assert_nil notifications[1]["params"] + + assert_equal "2.0", notifications[2]["jsonrpc"] + assert_equal Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED, notifications[2]["method"] + assert_nil notifications[2]["params"] + end + + test "notifications include params when provided" do + # Test the transport's send_notification directly with params + result = @transport.send_notification("test/notification", { data: "test_value" }) + + assert result + assert_equal 1, @mock_stdout.output.size + + notification = JSON.parse(@mock_stdout.output.first) + assert_equal "2.0", notification["jsonrpc"] + assert_equal "test/notification", notification["method"] + assert_equal({ "data" => "test_value" }, notification["params"]) + end + + test "server continues to work when stdout is closed" do + # Close stdout + @mock_stdout.close + + # Server notifications should not raise errors + assert_nothing_raised do + @server.notify_tools_list_changed + @server.notify_prompts_list_changed + @server.notify_resources_list_changed + end + end + + test "notifications work with dynamic tool additions" do + # Define a new tool + @server.define_tool( + name: "dynamic_tool", + description: "A dynamically added tool", + ) do |**args| + { result: "success" } + end + + # Clear previous output + @mock_stdout.output.clear + + # Manually trigger notification + @server.notify_tools_list_changed + + # Check the notification was sent + assert_equal 1, @mock_stdout.output.size + + notification = JSON.parse(@mock_stdout.output.first) + assert_equal Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED, notification["method"] + + # Verify the tool was added to the server + assert @server.tools.key?("dynamic_tool") + end + + test "notifications are properly formatted JSON-RPC 2.0 messages" do + # Send a notification + @server.notify_prompts_list_changed + + # Verify format + assert_equal 1, @mock_stdout.output.size + output = @mock_stdout.output.first + + # Should be valid JSON + notification = JSON.parse(output) + + # Should have required JSON-RPC 2.0 fields + assert_equal "2.0", notification["jsonrpc"] + assert notification.key?("method") + refute notification.key?("id") # Notifications should not have an id + + # Method should be the expected notification type + assert_equal Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED, notification["method"] + end + + test "multiple notifications are sent as separate JSON messages" do + # Send multiple notifications rapidly + 5.times do + @server.notify_tools_list_changed + end + + # Each should be a separate JSON message + assert_equal 5, @mock_stdout.output.size + + # All should be parseable as JSON + @mock_stdout.output.each do |msg| + notification = JSON.parse(msg) + assert_equal "2.0", notification["jsonrpc"] + assert_equal Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED, notification["method"] + end + end + + test "transport handles errors gracefully" do + # Create a stdout that raises errors + error_stdout = Class.new(MockIO) do + def puts(message) + raise IOError, "Simulated IO error" + end + end.new + + $stdout = error_stdout + + # Notification should return false but not raise + result = @transport.send_notification("test/notification") + refute result + end + + test "server notification flow works end-to-end with StdioTransport" do + # This test verifies the complete integration from server to transport + + # Start with no output + assert_empty @mock_stdout.output + + # Add a prompt and notify + @server.define_prompt( + name: "test_prompt", + description: "Test prompt", + ) do |args, server_context:| + MCP::PromptResponse.new(messages: [{ role: "user", content: "Test" }]) + end + + # Manually trigger notification + @server.notify_prompts_list_changed + + # Verify notification was sent + assert_equal 1, @mock_stdout.output.size + notification = JSON.parse(@mock_stdout.output.first) + assert_equal Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED, notification["method"] + + # Add a resource and notify + @server.resources = [ + MCP::Resource.new( + uri: "test://resource", + name: "Test Resource", + description: "A test resource", + mime_type: "text/plain", + ), + ] + + # Manually trigger notification + @server.notify_resources_list_changed + + # Verify both notifications were sent + assert_equal 2, @mock_stdout.output.size + second_notification = JSON.parse(@mock_stdout.output.last) + assert_equal Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED, second_notification["method"] + end + end + end + end +end diff --git a/test/mcp/server/transports/streamable_http_notification_integration_test.rb b/test/mcp/server/transports/streamable_http_notification_integration_test.rb new file mode 100644 index 0000000..89eca43 --- /dev/null +++ b/test/mcp/server/transports/streamable_http_notification_integration_test.rb @@ -0,0 +1,249 @@ +# frozen_string_literal: true + +require "test_helper" +require "json" +require "rack" +require "mcp/server" +require "mcp/server/transports/streamable_http_transport" + +module MCP + class Server + module Transports + class StreamableHTTPNotificationIntegrationTest < ActiveSupport::TestCase + setup do + @server = Server.new( + name: "test_server", + tools: [], + prompts: [], + resources: [], + ) + @transport = StreamableHTTPTransport.new(@server) + @server.transport = @transport + end + + test "server notification methods send SSE notifications through HTTP transport" do + # Initialize a session + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + init_response = @transport.handle_request(init_request) + session_id = init_response[1]["Mcp-Session-Id"] + + # Connect with SSE + io = StringIO.new + get_request = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id }, + ) + response = @transport.handle_request(get_request) + response[2].call(io) if response[2].is_a?(Proc) + + # Give the stream time to set up + sleep(0.1) + + # Test tools notification + @server.notify_tools_list_changed + + # Test prompts notification + @server.notify_prompts_list_changed + + # Test resources notification + @server.notify_resources_list_changed + + # Check the notifications were received + io.rewind + output = io.read + + assert_includes output, "data: #{Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED}" + assert_includes output, "data: #{Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED}" + assert_includes output, "data: #{Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED}" + end + + test "notifications are broadcast to all connected sessions" do + # Create two sessions + init_request1 = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "123" }.to_json, + ) + init_response1 = @transport.handle_request(init_request1) + session_id1 = init_response1[1]["Mcp-Session-Id"] + + init_request2 = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "456" }.to_json, + ) + init_response2 = @transport.handle_request(init_request2) + session_id2 = init_response2[1]["Mcp-Session-Id"] + + # Connect both sessions with SSE + io1 = StringIO.new + get_request1 = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id1 }, + ) + response1 = @transport.handle_request(get_request1) + response1[2].call(io1) if response1[2].is_a?(Proc) + + io2 = StringIO.new + get_request2 = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id2 }, + ) + response2 = @transport.handle_request(get_request2) + response2[2].call(io2) if response2[2].is_a?(Proc) + + # Give the streams time to set up + sleep(0.1) + + # Send notification through server + @server.notify_tools_list_changed + + # Check both sessions received the notification + io1.rewind + output1 = io1.read + assert_includes output1, "data: #{Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED}" + + io2.rewind + output2 = io2.read + assert_includes output2, "data: #{Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED}" + end + + test "server continues to work when SSE connection is closed" do + # Initialize a session + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + init_response = @transport.handle_request(init_request) + session_id = init_response[1]["Mcp-Session-Id"] + + # Connect with SSE + io = StringIO.new + get_request = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id }, + ) + response = @transport.handle_request(get_request) + response[2].call(io) if response[2].is_a?(Proc) + + # Give the stream time to set up + sleep(0.1) + + # Close the stream + io.close + + # Server notifications should not raise errors + assert_nothing_raised do + @server.notify_tools_list_changed + @server.notify_prompts_list_changed + @server.notify_resources_list_changed + end + end + + test "notifications work with dynamic tool additions" do + # Initialize a session + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + init_response = @transport.handle_request(init_request) + session_id = init_response[1]["Mcp-Session-Id"] + + # Connect with SSE + io = StringIO.new + get_request = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id }, + ) + response = @transport.handle_request(get_request) + response[2].call(io) if response[2].is_a?(Proc) + + # Give the stream time to set up + sleep(0.1) + + # Define a new tool (simulating dynamic tool addition) + @server.define_tool( + name: "dynamic_tool", + description: "A dynamically added tool", + ) do |**args| + { result: "success" } + end + + # Manually trigger notification (since we removed the automatic triggers) + @server.notify_tools_list_changed + + # Check the notification was received + io.rewind + output = io.read + assert_includes output, "data: #{Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED}" + + # Verify the tool was added to the server + assert @server.tools.key?("dynamic_tool") + end + + test "SSE format is correct for notifications" do + # Initialize a session + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + init_response = @transport.handle_request(init_request) + session_id = init_response[1]["Mcp-Session-Id"] + + # Connect with SSE + io = StringIO.new + get_request = create_rack_request( + "GET", + "/", + { "HTTP_MCP_SESSION_ID" => session_id }, + ) + response = @transport.handle_request(get_request) + response[2].call(io) if response[2].is_a?(Proc) + + # Give the stream time to set up + sleep(0.1) + + # Send a notification + @server.notify_tools_list_changed + + # Check SSE format + io.rewind + output = io.read + + # SSE format should be "data: \n\n" + assert_match(/data: #{Regexp.escape(Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED)}\n/, output) + end + + private + + def create_rack_request(method, path, headers, body = nil) + env = { + "REQUEST_METHOD" => method, + "PATH_INFO" => path, + "rack.input" => StringIO.new(body.to_s), + }.merge(headers) + + Rack::Request.new(env) + end + end + end + end +end diff --git a/test/mcp/server_notification_test.rb b/test/mcp/server_notification_test.rb new file mode 100644 index 0000000..217738d --- /dev/null +++ b/test/mcp/server_notification_test.rb @@ -0,0 +1,136 @@ +# typed: true +# frozen_string_literal: true + +require "test_helper" + +module MCP + class ServerNotificationTest < ActiveSupport::TestCase + include InstrumentationTestHelper + + class MockTransport < Transport + attr_reader :notifications + + def initialize(server) + super + @notifications = [] + end + + def send_notification(method, params = nil) + @notifications << { method: method, params: params } + true + end + + def send_response(response); end + def open; end + def close; end + def handle_request(request); end + end + + setup do + configuration = MCP::Configuration.new + configuration.instrumentation_callback = instrumentation_helper.callback + + @server = Server.new( + name: "test_server", + version: "1.0.0", + configuration: configuration + ) + + @mock_transport = MockTransport.new(@server) + @server.transport = @mock_transport + end + + test "#notify_tools_list_changed sends notification through transport" do + @server.notify_tools_list_changed + + assert_equal 1, @mock_transport.notifications.size + notification = @mock_transport.notifications.first + assert_equal Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED, notification[:method] + assert_nil notification[:params] + end + + test "#notify_prompts_list_changed sends notification through transport" do + @server.notify_prompts_list_changed + + assert_equal 1, @mock_transport.notifications.size + notification = @mock_transport.notifications.first + assert_equal Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED, notification[:method] + assert_nil notification[:params] + end + + test "#notify_resources_list_changed sends notification through transport" do + @server.notify_resources_list_changed + + assert_equal 1, @mock_transport.notifications.size + notification = @mock_transport.notifications.first + assert_equal Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED, notification[:method] + assert_nil notification[:params] + end + + test "notification methods work without transport" do + server_without_transport = Server.new(name: "test_server") + + # Should not raise any errors + assert_nothing_raised do + server_without_transport.notify_tools_list_changed + server_without_transport.notify_prompts_list_changed + server_without_transport.notify_resources_list_changed + end + end + + test "notification methods handle transport errors gracefully" do + # Create a transport that raises errors + error_transport = Class.new(MockTransport) do + def send_notification(method, params = nil) + raise StandardError, "Transport error" + end + end.new(@server) + + @server.transport = error_transport + + # Mock the exception reporter + expected_contexts = [ + { notification: "tools_list_changed" }, + { notification: "prompts_list_changed" }, + { notification: "resources_list_changed" } + ] + + call_count = 0 + @server.configuration.exception_reporter.expects(:call).times(3).with do |exception, context| + assert_kind_of StandardError, exception + assert_equal "Transport error", exception.message + assert_includes expected_contexts, context + call_count += 1 + true + end + + # Should not raise errors to the caller + assert_nothing_raised do + @server.notify_tools_list_changed + @server.notify_prompts_list_changed + @server.notify_resources_list_changed + end + + assert_equal 3, call_count + end + + test "multiple notification methods can be called in sequence" do + @server.notify_tools_list_changed + @server.notify_prompts_list_changed + @server.notify_resources_list_changed + + assert_equal 3, @mock_transport.notifications.size + + notifications = @mock_transport.notifications + assert_equal Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED, notifications[0][:method] + 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 \ No newline at end of file