Add missing handler for resource subscribe and unsubscribe (#220)

src/Server/Builder.php

@@ -33,6 +33,8 @@ use Mcp\Schema\ToolAnnotations;
 use Mcp\Server;
 use Mcp\Server\Handler\Notification\NotificationHandlerInterface;
 use Mcp\Server\Handler\Request\RequestHandlerInterface;
+use Mcp\Server\Resource\SessionSubscriptionManager;
+use Mcp\Server\Resource\SubscriptionManagerInterface;
 use Mcp\Server\Session\InMemorySessionStore;
 use Mcp\Server\Session\SessionFactory;
 use Mcp\Server\Session\SessionFactoryInterface;
@@ -54,6 +56,8 @@ final class Builder
 
     private RegistryInterface $registry;
 
+    private ?SubscriptionManagerInterface $subscriptionManager = null;
+
     private ?LoggerInterface $logger = null;
 
     private ?CacheInterface $discoveryCache = null;
@@ -309,6 +313,13 @@ final class Builder
         return $this;
     }
 
+    public function setResourceSubscriptionManager(SubscriptionManagerInterface $subscriptionManager): self
+    {
+        $this->subscriptionManager = $subscriptionManager;
+
+        return $this;
+    }
+
     public function setSession(
         SessionStoreInterface $sessionStore,
         SessionFactoryInterface $sessionFactory = new SessionFactory(),
@@ -489,12 +500,16 @@ final class Builder
         $logger = $this->logger ?? new NullLogger();
         $container = $this->container ?? new Container();
         $registry = $this->registry ?? new Registry($this->eventDispatcher, $logger);
-
+        $subscriptionManager = $this->subscriptionManager ?? new SessionSubscriptionManager($logger);
         $loaders = [
             ...$this->loaders,
             new ArrayLoader($this->tools, $this->resources, $this->resourceTemplates, $this->prompts, $logger, $this->schemaGenerator),
         ];
 
+        $sessionTtl = $this->sessionTtl ?? 3600;
+        $sessionFactory = $this->sessionFactory ?? new SessionFactory();
+        $sessionStore = $this->sessionStore ?? new InMemorySessionStore($sessionTtl);
+
         if (null !== $this->discoveryBasePath) {
             $discoverer = $this->discoverer ?? $this->createDiscoverer($logger);
             $loaders[] = new DiscoveryLoader($this->discoveryBasePath, $this->discoveryScanDirs, $this->discoveryExcludeDirs, $discoverer);
@@ -504,16 +519,13 @@ final class Builder
             $loader->load($registry);
         }
 
-        $sessionTtl = $this->sessionTtl ?? 3600;
-        $sessionFactory = $this->sessionFactory ?? new SessionFactory();
-        $sessionStore = $this->sessionStore ?? new InMemorySessionStore($sessionTtl);
         $messageFactory = MessageFactory::make();
 
         $capabilities = $this->serverCapabilities ?? new ServerCapabilities(
             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,
@@ -536,6 +548,8 @@ final class Builder
             new Handler\Request\ListToolsHandler($registry, $this->paginationLimit),
             new Handler\Request\PingHandler(),
             new Handler\Request\ReadResourceHandler($registry, $referenceHandler, $logger),
+            new Handler\Request\ResourceSubscribeHandler($registry, $subscriptionManager, $logger),
+            new Handler\Request\ResourceUnsubscribeHandler($registry, $subscriptionManager, $logger),
             new Handler\Request\SetLogLevelHandler(),
         ]);
 

src/Server/Handler/Request/ResourceSubscribeHandler.php

@@ -0,0 +1,72 @@
+<?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\Resource\SubscriptionManagerInterface;
+use Mcp\Server\Session\SessionInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Psr\SimpleCache\InvalidArgumentException;
+
+/**
+ * @implements RequestHandlerInterface<EmptyResult>
+ *
+ * @author Larry Sule-balogun <suleabimbola@gmail.com>
+ */
+final class ResourceSubscribeHandler implements RequestHandlerInterface
+{
+    public function __construct(
+        private readonly RegistryInterface $registry,
+        private readonly SubscriptionManagerInterface $subscriptionManager,
+        private readonly LoggerInterface $logger = new NullLogger(),
+    ) {
+    }
+
+    public function supports(Request $request): bool
+    {
+        return $request instanceof ResourceSubscribeRequest;
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    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->subscriptionManager->subscribe($session, $uri);
+
+        return new Response(
+            $request->getId(),
+            new EmptyResult(),
+        );
+    }
+}

src/Server/Handler/Request/ResourceUnsubscribeHandler.php

@@ -0,0 +1,72 @@
+<?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\Resource\SubscriptionManagerInterface;
+use Mcp\Server\Session\SessionInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Psr\SimpleCache\InvalidArgumentException;
+
+/**
+ * @implements RequestHandlerInterface<EmptyResult>
+ *
+ * @author Larry Sule-balogun <suleabimbola@gmail.com>
+ */
+final class ResourceUnsubscribeHandler implements RequestHandlerInterface
+{
+    public function __construct(
+        private readonly RegistryInterface $registry,
+        private readonly SubscriptionManagerInterface $subscriptionManager,
+        private readonly LoggerInterface $logger = new NullLogger(),
+    ) {
+    }
+
+    public function supports(Request $request): bool
+    {
+        return $request instanceof ResourceUnsubscribeRequest;
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    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->subscriptionManager->unsubscribe($session, $uri);
+
+        return new Response(
+            $request->getId(),
+            new EmptyResult(),
+        );
+    }
+}

src/Server/Resource/SessionSubscriptionManager.php

@@ -0,0 +1,94 @@
+<?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\Resource;
+
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\SessionInterface;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Psr\SimpleCache\InvalidArgumentException;
+
+/**
+ * The default Subscription manager implementation manages subscriptions per session only.
+ * It is in-memory and does not support cross-session or cross-client subscriptions.
+ *
+ * The SDK allows injecting alternative SubscriptionManagerInterface
+ * implementations via Builder::setResourceSubscriptionManager().
+ *
+ * @author Larry Sule-balogun <suleabimbola@gmail.com>
+ */
+final class SessionSubscriptionManager implements SubscriptionManagerInterface
+{
+    public function __construct(
+        private readonly LoggerInterface $logger = new NullLogger(),
+    ) {
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    public function subscribe(SessionInterface $session, string $uri): void
+    {
+        $subscriptions = $session->get('resource_subscriptions', []);
+        $subscriptions[$uri] = true;
+        $session->set('resource_subscriptions', $subscriptions);
+        $session->save();
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    public function unsubscribe(SessionInterface $session, string $uri): void
+    {
+        $subscriptions = $session->get('resource_subscriptions', []);
+        unset($subscriptions[$uri]);
+        $session->set('resource_subscriptions', $subscriptions);
+        $session->save();
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    public function isSubscribed(SessionInterface $session, string $uri): bool
+    {
+        $subscriptions = $session->get('resource_subscriptions', []);
+
+        return isset($subscriptions[$uri]);
+    }
+
+    /**
+     * @throws InvalidArgumentException
+     */
+    public function notifyResourceChanged(Protocol $protocol, SessionInterface $session, string $uri): void
+    {
+        $activeSession = $this->isSubscribed($session, $uri);
+        if (!$activeSession) {
+            return;
+        }
+
+        try {
+            $protocol->sendNotification(
+                new ResourceUpdatedNotification($uri),
+                $session
+            );
+        } catch (InvalidArgumentException $e) {
+            $this->logger->error('Error sending resource notification to session', [
+                'session_id' => $session->getId()->toRfc4122(),
+                'uri' => $uri,
+                'exception' => $e,
+            ]);
+
+            throw $e;
+        }
+    }
+}

src/Server/Resource/SubscriptionManagerInterface.php

@@ -0,0 +1,53 @@
+<?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\Resource;
+
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\SessionInterface;
+use Psr\SimpleCache\InvalidArgumentException;
+
+/**
+ * Resource subscription interface.
+ *
+ * @author Larry Sule-balogun <suleabimbola@gmail.com>
+ */
+interface SubscriptionManagerInterface
+{
+    /**
+     * Subscribes a session to a specific resource URI.
+     *
+     * @throws InvalidArgumentException
+     */
+    public function subscribe(SessionInterface $session, string $uri): void;
+
+    /**
+     * Unsubscribes a session from a specific resource URI.
+     *
+     * @throws InvalidArgumentException
+     */
+    public function unsubscribe(SessionInterface $session, string $uri): void;
+
+    /**
+     * Check if a session is subscribed to a resource URI.
+     *
+     * @throws InvalidArgumentException
+     */
+    public function isSubscribed(SessionInterface $session, string $uri): bool;
+
+    /**
+     * Notifies all sessions subscribed to the given resource URI that the
+     * resource has changed. Sends a ResourceUpdatedNotification for each subscriber.
+     *
+     * @throws InvalidArgumentException
+     */
+    public function notifyResourceChanged(Protocol $protocol, SessionInterface $session, string $uri): void;
+}