Merge pull request modelcontextprotocol#313 from koic/support_resource_subscription

Support resource subscriptions per MCP specification

Author: koic

README.md

@@ -53,6 +53,8 @@ It implements the Model Context Protocol specification, handling model context r
 - `resources/list` - Lists all registered resources and their schemas
 - `resources/read` - Retrieves a specific resource by name
 - `resources/templates/list` - Lists all registered resource templates and their schemas
+- `resources/subscribe` - Subscribes to updates for a specific resource
+- `resources/unsubscribe` - Unsubscribes from updates for a specific resource
 - `completion/complete` - Returns autocompletion suggestions for prompt arguments and resource URIs
 - `roots/list` - Requests filesystem roots from the client (server-to-client)
 - `sampling/createMessage` - Requests LLM completion from the client (server-to-client)
@@ -958,6 +960,44 @@ end
 - Raises `RuntimeError` if client does not support `roots` capability
 - Raises `StandardError` if client returns an error response
 
+### Resource Subscriptions
+
+Resource subscriptions allow clients to monitor specific resources for changes.
+When a subscribed resource is updated, the server sends a notification to the client.
+
+The SDK does not track subscription state internally.
+Server developers register handlers and manage their own subscription state.
+Three methods are provided:
+
+- `Server#resources_subscribe_handler` - registers a handler for `resources/subscribe` requests
+- `Server#resources_unsubscribe_handler` - registers a handler for `resources/unsubscribe` requests
+- `ServerContext#notify_resources_updated` - sends a `notifications/resources/updated` notification to the subscribing client
+
+```ruby
+subscribed_uris = Set.new
+
+server = MCP::Server.new(
+  name: "my_server",
+  resources: [my_resource],
+  capabilities: { resources: { subscribe: true } },
+)
+
+server.resources_subscribe_handler do |params|
+  subscribed_uris.add(params[:uri].to_s)
+end
+
+server.resources_unsubscribe_handler do |params|
+  subscribed_uris.delete(params[:uri].to_s)
+end
+
+server.define_tool(name: "update_resource") do |server_context:, **args|
+  if subscribed_uris.include?("test://my-resource")
+    server_context.notify_resources_updated(uri: "test://my-resource")
+  end
+  MCP::Tool::Response.new([MCP::Content::Text.new("Resource updated").to_h])
+end
+```
+
 ### Sampling
 
 The Model Context Protocol allows servers to request LLM completions from clients through the `sampling/createMessage` method.
@@ -1503,10 +1543,6 @@ end
 - Raises `MCP::Server::MethodAlreadyDefinedError` if trying to override an existing method
 - Supports the same exception reporting and instrumentation as standard methods
 
-### Unsupported Features (to be implemented in future versions)
-
-- Resource subscriptions
-
 ## Building an MCP Client
 
 The `MCP::Client` class provides an interface for interacting with MCP servers.

conformance/server.rb

@@ -2,6 +2,7 @@
 
 require "rackup"
 require "json"
+require "set"
 require "uri"
 require_relative "../lib/mcp"
 
@@ -539,6 +540,7 @@ def configure_handlers(server)
       server.server_context = server
 
       configure_resources_read_handler(server)
+      configure_subscription_handlers(server)
       configure_completion_handler(server)
     end
 
@@ -609,6 +611,18 @@ def configure_completion_handler(server)
       end
     end
 
+    def configure_subscription_handlers(server)
+      subscribed_uris = Set.new
+
+      server.resources_subscribe_handler do |params|
+        subscribed_uris.add(params[:uri].to_s)
+      end
+
+      server.resources_unsubscribe_handler do |params|
+        subscribed_uris.delete(params[:uri].to_s)
+      end
+    end
+
     def build_rack_app(transport)
       mcp_app = proc do |env|
         request = Rack::Request.new(env)

lib/mcp/server.rb

@@ -117,6 +117,8 @@ def initialize(
         Methods::RESOURCES_LIST => method(:list_resources),
         Methods::RESOURCES_READ => method(:read_resource_no_content),
         Methods::RESOURCES_TEMPLATES_LIST => method(:list_resource_templates),
+        Methods::RESOURCES_SUBSCRIBE => ->(_) { {} },
+        Methods::RESOURCES_UNSUBSCRIBE => ->(_) { {} },
         Methods::TOOLS_LIST => method(:list_tools),
         Methods::TOOLS_CALL => method(:call_tool),
         Methods::PROMPTS_LIST => method(:list_prompts),
@@ -128,10 +130,6 @@ def initialize(
         Methods::NOTIFICATIONS_ROOTS_LIST_CHANGED => ->(_) {},
         Methods::COMPLETION_COMPLETE => ->(_) { DEFAULT_COMPLETION_RESULT },
         Methods::LOGGING_SET_LEVEL => method(:configure_logging_level),
-
-        # No op handlers for currently unsupported methods
-        Methods::RESOURCES_SUBSCRIBE => ->(_) { {} },
-        Methods::RESOURCES_UNSUBSCRIBE => ->(_) { {} },
       }
       @transport = transport
     end
@@ -258,6 +256,24 @@ def completion_handler(&block)
       @handlers[Methods::COMPLETION_COMPLETE] = block
     end
 
+    # Sets a custom handler for `resources/subscribe` requests.
+    # The block receives the parsed request params. The return value is
+    # ignored; the response is always an empty result `{}` per the MCP specification.
+    #
+    # @yield [params] The request params containing `:uri`.
+    def resources_subscribe_handler(&block)
+      @handlers[Methods::RESOURCES_SUBSCRIBE] = block
+    end
+
+    # Sets a custom handler for `resources/unsubscribe` requests.
+    # The block receives the parsed request params. The return value is
+    # ignored; the response is always an empty result `{}` per the MCP specification.
+    #
+    # @yield [params] The request params containing `:uri`.
+    def resources_unsubscribe_handler(&block)
+      @handlers[Methods::RESOURCES_UNSUBSCRIBE] = block
+    end
+
     def build_sampling_params(
       capabilities,
       messages:,
@@ -391,6 +407,9 @@ def handle_request(request, method, session: nil, related_request_id: nil)
             init(params, session: session)
           when Methods::RESOURCES_READ
             { contents: @handlers[Methods::RESOURCES_READ].call(params) }
+          when Methods::RESOURCES_SUBSCRIBE, Methods::RESOURCES_UNSUBSCRIBE
+            @handlers[method].call(params)
+            {}
           when Methods::TOOLS_CALL
             call_tool(params, session: session, related_request_id: related_request_id)
           when Methods::COMPLETION_COMPLETE

lib/mcp/server_context.rb

@@ -30,6 +30,15 @@ def notify_log_message(data:, level:, logger: nil)
       @notification_target.notify_log_message(data: data, level: level, logger: logger, related_request_id: @related_request_id)
     end
 
+    # Sends a resource updated notification scoped to the originating session.
+    #
+    # @param uri [String] The URI of the updated resource.
+    def notify_resources_updated(uri:)
+      return unless @notification_target
+
+      @notification_target.notify_resources_updated(uri: uri)
+    end
+
     # Delegates to the session so the request is scoped to the originating client.
     def list_roots
       if @notification_target.respond_to?(:list_roots)

lib/mcp/server_session.rb

@@ -85,6 +85,13 @@ def notify_elicitation_complete(elicitation_id:)
       @server.report_exception(e, notification: "elicitation_complete")
     end
 
+    # Sends a resource updated notification to this session only.
+    def notify_resources_updated(uri:)
+      send_to_transport(Methods::NOTIFICATIONS_RESOURCES_UPDATED, { "uri" => uri })
+    rescue => e
+      @server.report_exception(e, notification: "resources_updated")
+    end
+
     # Sends a progress notification to this session only.
     def notify_progress(progress_token:, progress:, total: nil, message: nil, related_request_id: nil)
       params = {

test/mcp/server_context_test.rb

@@ -192,6 +192,23 @@ def context.custom_method
       assert_nothing_raised { server_context.notify_log_message(data: "test", level: "info") }
     end
 
+    test "ServerContext#notify_resources_updated delegates to notification_target" do
+      notification_target = mock
+      notification_target.expects(:notify_resources_updated).with(uri: "test://resource-1").once
+
+      progress = Progress.new(notification_target: notification_target, progress_token: nil)
+      server_context = ServerContext.new(nil, progress: progress, notification_target: notification_target)
+
+      server_context.notify_resources_updated(uri: "test://resource-1")
+    end
+
+    test "ServerContext#notify_resources_updated is a no-op when notification_target is nil" do
+      progress = Progress.new(notification_target: nil, progress_token: nil)
+      server_context = ServerContext.new(nil, progress: progress, notification_target: nil)
+
+      assert_nothing_raised { server_context.notify_resources_updated(uri: "test://resource-1") }
+    end
+
     # Tool without server_context parameter
     class SimpleToolWithoutContext < Tool
       tool_name "simple_without_context"

test/mcp/server_test.rb

@@ -2233,6 +2233,72 @@ class Example < Tool
       )
     end
 
+    test "#handle resources/subscribe with custom handler calls the handler" do
+      server = Server.new(
+        name: "test_server",
+        capabilities: { resources: { subscribe: true } },
+      )
+
+      received_params = nil
+      server.resources_subscribe_handler do |params|
+        received_params = params
+        {}
+      end
+
+      server.handle({ jsonrpc: "2.0", method: "initialize", id: 1 })
+      server.handle({ jsonrpc: "2.0", method: "notifications/initialized" })
+
+      response = server.handle({
+        jsonrpc: "2.0",
+        id: 2,
+        method: "resources/subscribe",
+        params: { uri: "https://example.com/resource" },
+      })
+
+      assert_equal(
+        {
+          jsonrpc: "2.0",
+          id: 2,
+          result: {},
+        },
+        response,
+      )
+      assert_equal "https://example.com/resource", received_params[:uri]
+    end
+
+    test "#handle resources/unsubscribe with custom handler calls the handler" do
+      server = Server.new(
+        name: "test_server",
+        capabilities: { resources: { subscribe: true } },
+      )
+
+      received_params = nil
+      server.resources_unsubscribe_handler do |params|
+        received_params = params
+        {}
+      end
+
+      server.handle({ jsonrpc: "2.0", method: "initialize", id: 1 })
+      server.handle({ jsonrpc: "2.0", method: "notifications/initialized" })
+
+      response = server.handle({
+        jsonrpc: "2.0",
+        id: 2,
+        method: "resources/unsubscribe",
+        params: { uri: "https://example.com/resource" },
+      })
+
+      assert_equal(
+        {
+          jsonrpc: "2.0",
+          id: 2,
+          result: {},
+        },
+        response,
+      )
+      assert_equal "https://example.com/resource", received_params[:uri]
+    end
+
     test "tools/call with no args" do
       server = Server.new(tools: [@tool_with_no_args])