Add missing handler for resource subscribe and unsubscribe

Author: bigdevlarry

.github/workflows/pipeline.yaml

@@ -100,7 +100,7 @@ jobs:
           passedTests=$(echo "$OUTPUT" | sed -nE 's/.*Total: ([0-9]+) passed.*/\1/p')
           passedTests=${passedTests:-0}
           
-          REQUIRED_TESTS_TO_PASS=22
+          REQUIRED_TESTS_TO_PASS=25
           echo "Required tests to pass: $REQUIRED_TESTS_TO_PASS"
           [ "$passedTests" -ge "$REQUIRED_TESTS_TO_PASS" ] || exit $exit_code
 

src/Capability/Registry.php

@@ -25,11 +25,14 @@
 use Mcp\Exception\PromptNotFoundException;
 use Mcp\Exception\ResourceNotFoundException;
 use Mcp\Exception\ToolNotFoundException;
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
 use Mcp\Schema\Page;
 use Mcp\Schema\Prompt;
 use Mcp\Schema\Resource;
 use Mcp\Schema\ResourceTemplate;
 use Mcp\Schema\Tool;
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\SessionInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
@@ -61,6 +64,11 @@ final class Registry implements RegistryInterface
      */
     private array $resourceTemplates = [];
 
+    /**
+     * @var array<string, array<string, SessionInterface>>
+     */
+    private array $resourceSubscriptions = [];
+
     public function __construct(
         private readonly ?EventDispatcherInterface $eventDispatcher = null,
         private readonly LoggerInterface $logger = new NullLogger(),
@@ -449,4 +457,40 @@ private function paginateResults(array $items, int $limit, ?string $cursor = nul
 
         return array_values(\array_slice($items, $offset, $limit));
     }
+
+    public function subscribe(SessionInterface $session, string $uri): void
+    {
+        if (!isset($this->resourceSubscriptions[$uri])) {
+            $this->resourceSubscriptions[$uri] = [];
+        }
+
+        $sessionId = $session->getId()->toRfc4122();
+        $this->resourceSubscriptions[$uri][$sessionId] = $session;
+    }
+
+    public function unsubscribe(SessionInterface $session, string $uri): void
+    {
+        if (!isset($this->resourceSubscriptions[$uri])) {
+            return;
+        }
+
+        $sessionId = $session->getId()->toRfc4122();
+
+        unset($this->resourceSubscriptions[$uri][$sessionId]);
+
+        if ([] === $this->resourceSubscriptions[$uri]) {
+            unset($this->resourceSubscriptions[$uri]);
+        }
+    }
+
+    public function notifyResourceChanged(Protocol $protocol, string $uri): void
+    {
+        if (!isset($this->resourceSubscriptions[$uri])) {
+            return;
+        }
+
+        foreach ($this->resourceSubscriptions[$uri] as $session) {
+            $protocol->sendNotification(new ResourceUpdatedNotification($uri), $session);
+        }
+    }
 }

src/Capability/RegistryInterface.php

@@ -25,6 +25,8 @@
 use Mcp\Schema\Resource;
 use Mcp\Schema\ResourceTemplate;
 use Mcp\Schema\Tool;
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\SessionInterface;
 
 /**
  * @phpstan-import-type Handler from ElementReference
@@ -157,4 +159,20 @@ public function getPrompts(?int $limit = null, ?string $cursor = null): Page;
      * @throws PromptNotFoundException
      */
     public function getPrompt(string $name): PromptReference;
+
+    /**
+     * Subscribes a session to a specific resource URI.
+     */
+    public function subscribe(SessionInterface $session, string $uri): void;
+
+    /**
+     * Unsubscribes a session from a specific resource URI.
+     */
+    public function unsubscribe(SessionInterface $session, string $uri): void;
+
+    /**
+     * Notifies all sessions subscribed to the given resource URI that the
+     * resource has changed. Sends a ResourceUpdatedNotification for each subscriber.
+     */
+    public function notifyResourceChanged(Protocol $protocol, string $uri): void;
 }

src/Server/Builder.php

@@ -486,7 +486,7 @@ public function build(): Server
             tools: $registry->hasTools(),
             toolsListChanged: $this->eventDispatcher instanceof EventDispatcherInterface,
             resources: $registry->hasResources() || $registry->hasResourceTemplates(),
-            resourcesSubscribe: false,
+            resourcesSubscribe: $registry->hasResources() || $registry->hasResourceTemplates(),
             resourcesListChanged: $this->eventDispatcher instanceof EventDispatcherInterface,
             prompts: $registry->hasPrompts(),
             promptsListChanged: $this->eventDispatcher instanceof EventDispatcherInterface,
@@ -509,6 +509,8 @@ public function build(): Server
             new Handler\Request\ListToolsHandler($registry, $this->paginationLimit),
             new Handler\Request\PingHandler(),
             new Handler\Request\ReadResourceHandler($registry, $referenceHandler, $logger),
+            new Handler\Request\ResourceSubscribeHandler($registry, $logger),
+            new Handler\Request\ResourceUnsubscribeHandler($registry, $logger),
             new Handler\Request\SetLogLevelHandler(),
         ]);
 

src/Server/Handler/Request/ResourceSubscribeHandler.php

@@ -0,0 +1,66 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Server\Handler\Request;
+
+use Mcp\Capability\RegistryInterface;
+use Mcp\Exception\ResourceNotFoundException;
+use Mcp\Schema\JsonRpc\Error;
+use Mcp\Schema\JsonRpc\Request;
+use Mcp\Schema\JsonRpc\Response;
+use Mcp\Schema\Request\ResourceSubscribeRequest;
+use Mcp\Schema\Result\EmptyResult;
+use Mcp\Server\Session\SessionInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+/**
+ * @implements RequestHandlerInterface<EmptyResult>
+ *
+ * @author Larry Sule-balogun <[email protected]>
+ */
+final class ResourceSubscribeHandler implements RequestHandlerInterface
+{
+    public function __construct(
+        private readonly RegistryInterface $registry,
+        private readonly LoggerInterface $logger = new NullLogger(),
+    ) {
+    }
+
+    public function supports(Request $request): bool
+    {
+        return $request instanceof ResourceSubscribeRequest;
+    }
+
+    public function handle(Request $request, SessionInterface $session): Response|Error
+    {
+        \assert($request instanceof ResourceSubscribeRequest);
+
+        $uri = $request->uri;
+
+        try {
+            $this->registry->getResource($uri);
+        } catch (ResourceNotFoundException $e) {
+            $this->logger->error('Resource not found', ['uri' => $uri]);
+
+            return Error::forResourceNotFound($e->getMessage(), $request->getId());
+        }
+
+        $this->logger->debug('Subscribing to resource', ['uri' => $uri]);
+
+        $this->registry->subscribe($session, $uri);
+
+        return new Response(
+            $request->getId(),
+            new EmptyResult(),
+        );
+    }
+}

src/Server/Handler/Request/ResourceUnsubscribeHandler.php

@@ -0,0 +1,66 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Server\Handler\Request;
+
+use Mcp\Capability\RegistryInterface;
+use Mcp\Exception\ResourceNotFoundException;
+use Mcp\Schema\JsonRpc\Error;
+use Mcp\Schema\JsonRpc\Request;
+use Mcp\Schema\JsonRpc\Response;
+use Mcp\Schema\Request\ResourceUnsubscribeRequest;
+use Mcp\Schema\Result\EmptyResult;
+use Mcp\Server\Session\SessionInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+/**
+ * @implements RequestHandlerInterface<EmptyResult>
+ *
+ * @author Larry Sule-balogun <[email protected]>
+ */
+final class ResourceUnsubscribeHandler implements RequestHandlerInterface
+{
+    public function __construct(
+        private readonly RegistryInterface $registry,
+        private readonly LoggerInterface $logger = new NullLogger(),
+    ) {
+    }
+
+    public function supports(Request $request): bool
+    {
+        return $request instanceof ResourceUnsubscribeRequest;
+    }
+
+    public function handle(Request $request, SessionInterface $session): Response|Error
+    {
+        \assert($request instanceof ResourceUnsubscribeRequest);
+
+        $uri = $request->uri;
+
+        try {
+            $this->registry->getResource($uri);
+        } catch (ResourceNotFoundException $e) {
+            $this->logger->error('Resource not found', ['uri' => $uri]);
+
+            return Error::forResourceNotFound($e->getMessage(), $request->getId());
+        }
+
+        $this->logger->debug('Unsubscribing from resource', ['uri' => $uri]);
+
+        $this->registry->unsubscribe($session, $uri);
+
+        return new Response(
+            $request->getId(),
+            new EmptyResult(),
+        );
+    }
+}

tests/Conformance/server.php

@@ -9,10 +9,13 @@
  * file that was distributed with this source code.
  */
 
+ini_set('display_errors', '0');
+
 require_once dirname(__DIR__, 2).'/vendor/autoload.php';
 
 use Http\Discovery\Psr17Factory;
 use Laminas\HttpHandlerRunner\Emitter\SapiEmitter;
+use Mcp\Capability\Registry;
 use Mcp\Schema\Content\AudioContent;
 use Mcp\Schema\Content\EmbeddedResource;
 use Mcp\Schema\Content\ImageContent;
@@ -32,6 +35,7 @@
 $request = $psr17Factory->createServerRequestFromGlobals();
 
 $transport = new StreamableHttpTransport($request, logger: $logger);
+$registry = new Registry(null, $logger);
 
 $server = Server::builder()
     ->setServerInfo('mcp-conformance-test-server', '1.0.0')
@@ -51,7 +55,6 @@
     ->addResource(fn () => 'This is the content of the static text resource.', 'test://static-text', 'static-text', 'A static text resource for testing')
     ->addResource(fn () => fopen('data://image/png;base64,'.Elements::TEST_IMAGE_BASE64, 'r'), 'test://static-binary', 'static-binary', 'A static binary resource (image) for testing')
     ->addResourceTemplate([Elements::class, 'resourceTemplate'], 'test://template/{id}/data', 'template', 'A resource template with parameter substitution', 'application/json')
-    // TODO: Handler for resources/subscribe and resources/unsubscribe
     ->addResource(fn () => 'Watched resource content', 'test://watched-resource', 'watched-resource', 'A resource that can be watched')
     // Prompts
     ->addPrompt(fn () => [['role' => 'user', 'content' => 'This is a simple prompt for testing.']], 'test_simple_prompt', 'A simple prompt without arguments')