[Server] Make subscriptions/listen actually deliver (#451)

The stream acknowledged, slept thirty seconds and closed, so tools and resource changes never reached a client — and with resources/subscribe gone, that left the revision with no server-push at all. A notification bus carries them: in-memory where publisher and stream share a process, PSR-16 where php-fpm puts them in different ones. The conformance server proves the latter — a tools/call in one worker reaches a stream held in another.
2026-08-19 00:08:02 +02:00 · 2026-08-19 00:08:02 +02:00 · daac88d626
commit daac88d626
parent 7dd72d82de
12 changed files with 907 additions and 13 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@ -5,6 +5,7 @@ All notable changes to `mcp/sdk` will be documented in this file.
 0.8.0
 -----

+* Deliver notifications on a `subscriptions/listen` stream (SEP-2575), which previously acknowledged and then carried nothing for the rest of its life. New `Mcp\Server\Subscription\NotificationBusInterface` with two implementations — `InMemoryNotificationBus` for stdio and persistent runtimes, `Psr16NotificationBus` for PHP-FPM, where the worker holding the stream open and the worker publishing are different processes — set with `Builder::setNotificationBus()`. Registry changes are published automatically through a `PublishingEventDispatcher` that wraps whatever PSR-14 dispatcher was configured. `Builder::setSubscriptionLifetime()` replaces the hard-coded 30-second ceiling, where `0` means "until the client or the runtime ends it".
 * Add `Mcp\Server\Wire\CachePolicy`, set with `Builder::setCachePolicy()`, to configure the SEP-2549 caching hints the 2026-07-28 lifecycle stamps on a cacheable result. The conservative `ttlMs: 0, cacheScope: private` stays the default, since `public` lets a shared proxy serve one caller's answer to another and only the operator can make that call. A `ReadResourceResult` may also carry its own `ttlMs`/`cacheScope`, which win over the policy.
 * Add typed readers to `Mcp\Server\Stateless\InputContext`: `elicitResult()`, `samplingResult()` and `rootsResult()` deserialize a multi round-trip answer instead of handing back the raw array `response()` returns. A malformed answer reads as absent rather than throwing, so a handler asks again — which is what the spec says a server SHOULD do when the information it needs is still missing.
 * Answer a request over a response stream under the 2026-07-28 lifecycle: `StatelessProtocol` runs handlers in a fiber, so `$gateway->progress()` and `$gateway->log()` work there as they do in the handshake era. The stream opens only if the handler actually emits something *and* the client's `Accept` admits `text/event-stream`, and the choice is made after the handler's first suspension, so a request that turns out to need `-32021` or `-32602` is still answered with the status the spec fixes for it.
--- a/src/Server/Builder.php
+++ b/src/Server/Builder.php
@ -59,6 +59,10 @@ use Mcp\Server\Session\SessionStoreInterface;
 use Mcp\Server\Stateless\RequestStateCodec;
 use Mcp\Server\Stateless\StandardHeaderValidator;
 use Mcp\Server\Stateless\StatelessProtocol;
+use Mcp\Server\Subscription\InMemoryNotificationBus;
+use Mcp\Server\Subscription\NotificationBusInterface;
+use Mcp\Server\Subscription\Psr16NotificationBus;
+use Mcp\Server\Subscription\PublishingEventDispatcher;
 use Mcp\Server\Wire\CachePolicy;
 use Psr\Container\ContainerInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
@ -110,6 +114,10 @@ final class Builder

    private ?CachePolicy $cachePolicy = null;

+    private ?NotificationBusInterface $notificationBus = null;
+
+    private float $subscriptionLifetime = 30.0;
+
    private ?string $requestStateKey = null;

    private int $requestStateTtl = 600;
@ -258,6 +266,45 @@ final class Builder
        return $this;
    }

+    /**
+     * Sets the bus carrying server-initiated notifications to open
+     * `subscriptions/listen` streams (SEP-2575).
+     *
+     * Without one, a listen stream acknowledges and then carries nothing: there
+     * is no safe default, because the right implementation depends on whether
+     * the publisher and the stream share a process.
+     * {@see InMemoryNotificationBus} is correct for stdio and persistent
+     * runtimes; under PHP-FPM, where they are different workers, use
+     * {@see Psr16NotificationBus} or an implementation over your own broker.
+     *
+     * Registry changes are published automatically when an event dispatcher is
+     * configured; anything else — `notifications/resources/updated` above all —
+     * is published by the application calling
+     * {@see NotificationBusInterface::publish()}.
+     */
+    public function setNotificationBus(NotificationBusInterface $bus): self
+    {
+        $this->notificationBus = $bus;
+
+        return $this;
+    }
+
+    /**
+     * Sets how long a `subscriptions/listen` stream is held open before the
+     * server closes it gracefully.
+     *
+     * The real ceiling is the runtime's: under PHP-FPM a stream cannot outlive
+     * `max_execution_time`, and a value above it buys a killed worker instead
+     * of a longer subscription. Pass `0` for "until the client or the runtime
+     * ends it", which is what a persistent runtime wants.
+     */
+    public function setSubscriptionLifetime(float $seconds): self
+    {
+        $this->subscriptionLifetime = max(0.0, $seconds);
+
+        return $this;
+    }
+
    /**
     * Sets how long, and to whom, this server's answers may be cached (SEP-2549).
     *
@ -757,7 +804,7 @@ final class Builder
            messageFactory: $parts['messageFactory'],
            sessionManager: $parts['sessionManager'],
            logger: $parts['logger'],
-            eventDispatcher: $this->eventDispatcher,
+            eventDispatcher: $parts['eventDispatcher'],
        );

        return new Server($protocol, $parts['logger']);
@ -782,11 +829,13 @@ final class Builder
            configuration: $parts['configuration'],
            supportedVersions: $supportedVersions,
            logger: $parts['logger'],
+            subscriptionLifetime: $this->subscriptionLifetime,
            headerValidator: $this->headerValidation ? new StandardHeaderValidator($parts['registry']) : null,
            requestStateCodec: null !== $this->requestStateKey
                ? new RequestStateCodec($this->requestStateKey, $this->requestStateTtl)
                : null,
            cachePolicy: $this->cachePolicy,
+            notificationBus: $this->notificationBus,
        );
    }

@ -795,6 +844,7 @@ final class Builder
     *
     * @return array{
     *     logger: LoggerInterface,
+     *     eventDispatcher: ?EventDispatcherInterface,
     *     configuration: Configuration,
     *     messageFactory: MessageFactory,
     *     sessionManager: SessionManagerInterface,
@ -807,6 +857,12 @@ final class Builder
    {
        $logger = $this->logger ?? new NullLogger();
        $container = $this->container ?? new Container();
+
+        // A configured bus needs the registry's change events, and PSR-14 hands
+        // the SDK a dispatcher it cannot register listeners on — so it wraps.
+        $eventDispatcher = null !== $this->notificationBus
+            ? new PublishingEventDispatcher($this->notificationBus, $this->eventDispatcher)
+            : $this->eventDispatcher;
        $subscriptionManager = $this->subscriptionManager ?? new SessionSubscriptionManager($logger);
        $sessionManager = $this->sessionManager ?? new SessionManager(
            $this->sessionStore ?? new InMemorySessionStore(),
@ -845,7 +901,7 @@ final class Builder
            $chainLoader->load($registry);
            $eagerlyLoaded = true;
        } else {
-            $registry = new Registry($this->eventDispatcher, $logger, loader: $chainLoader);
+            $registry = new Registry($eventDispatcher, $logger, loader: $chainLoader);
            if (!$this->lazyLoading) {
                $registry->load();
            }
@ -854,7 +910,7 @@ final class Builder

        $messageFactory = MessageFactory::make(additional: $this->extensionMessages);

-        $capabilities = $this->serverCapabilities ?? $this->detectCapabilities($registry, $eagerlyLoaded);
+        $capabilities = $this->serverCapabilities ?? $this->detectCapabilities($registry, $eagerlyLoaded, $eventDispatcher);

        // Extensions enabled via enableExtension() are folded into caller-supplied
        // capabilities too, so setCapabilities() does not silently drop them.
@ -895,6 +951,7 @@ final class Builder

        return [
            'logger' => $logger,
+            'eventDispatcher' => $eventDispatcher,
            'registry' => $registry,
            'configuration' => $configuration,
            'messageFactory' => $messageFactory,
@ -909,9 +966,11 @@ final class Builder
     * the load, so they are advertised from the configured sources instead — opaque sources (custom
     * loaders, discovery) advertise all kinds, and over-advertising is harmless per MCP semantics.
     */
-    private function detectCapabilities(RegistryInterface $registry, bool $eagerlyLoaded): ServerCapabilities
+    private function detectCapabilities(RegistryInterface $registry, bool $eagerlyLoaded, ?EventDispatcherInterface $eventDispatcher): ServerCapabilities
    {
-        $listChanged = $this->eventDispatcher instanceof EventDispatcherInterface;
+        // Without a dispatcher the registry announces nothing, so there is no
+        // list-changed notification to advertise.
+        $listChanged = $eventDispatcher instanceof EventDispatcherInterface;

        if ($eagerlyLoaded) {
            $hasResources = $registry->hasResources() || $registry->hasResourceTemplates();
--- a/src/Server/Stateless/NotificationFilter.php
+++ b/src/Server/Stateless/NotificationFilter.php
@ -11,6 +11,11 @@

 namespace Mcp\Server\Stateless;

+use Mcp\Schema\JsonRpc\Notification;
+use Mcp\Schema\Notification\PromptListChangedNotification;
+use Mcp\Schema\Notification\ResourceListChangedNotification;
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
+use Mcp\Schema\Notification\ToolListChangedNotification;
 use Mcp\Schema\ServerCapabilities;

 /**
@ -65,6 +70,24 @@ final class NotificationFilter
        );
    }

+    /**
+     * Whether this filter admits $notification onto the stream.
+     *
+     * An allow-list decision, so an unrecognized notification is declined: the
+     * server MUST NOT send a type the client did not ask for, and "did not ask
+     * for" includes types it has never heard of.
+     */
+    public function carries(Notification $notification): bool
+    {
+        return match (true) {
+            $notification instanceof ToolListChangedNotification => $this->toolsListChanged,
+            $notification instanceof PromptListChangedNotification => $this->promptsListChanged,
+            $notification instanceof ResourceListChangedNotification => $this->resourcesListChanged,
+            $notification instanceof ResourceUpdatedNotification => \in_array($notification->uri, $this->resourceSubscriptions, true),
+            default => false,
+        };
+    }
+
    /**
     * Agreed types only; declined ones are omitted rather than sent as `false`.
     *
--- a/src/Server/Stateless/StatelessProtocol.php
+++ b/src/Server/Stateless/StatelessProtocol.php
@ -31,6 +31,7 @@ use Mcp\Server\Handler\Request\RequestHandlerInterface;
 use Mcp\Server\Protocol;
 use Mcp\Server\Session\InMemorySessionStore;
 use Mcp\Server\Session\Session;
+use Mcp\Server\Subscription\NotificationBusInterface;
 use Mcp\Server\Wire\CachePolicy;
 use Mcp\Server\Wire\Rev2026Codec;
 use Mcp\Server\Wire\WireCodecInterface;
@ -100,6 +101,7 @@ final class StatelessProtocol
        private readonly ?StandardHeaderValidator $headerValidator = null,
        private readonly ?RequestStateCodec $requestStateCodec = null,
        ?CachePolicy $cachePolicy = null,
+        private readonly ?NotificationBusInterface $notificationBus = null,
    ) {
        $this->codec = $codec ?? new Rev2026Codec($configuration->serverInfo, $cachePolicy);

@ -295,8 +297,12 @@ final class StatelessProtocol
        $agreed = NotificationFilter::fromParams($notifications)->intersect($this->configuration->capabilities);

        $lifetime = $this->subscriptionLifetime;
+        $bus = $this->notificationBus;
+        $codec = $this->codec;

-        return StatelessResult::stream(static function () use ($agreed, $id, $lifetime): \Generator {
+        return StatelessResult::stream(static function () use ($agreed, $id, $lifetime, $bus, $codec): \Generator {
+            // MUST be the first message carrying this subscription's id, and
+            // MUST precede any notification on it.
            yield [
                'jsonrpc' => '2.0',
                'method' => self::ACKNOWLEDGED_NOTIFICATION,
@ -306,12 +312,27 @@ final class StatelessProtocol
                ],
            ];

-            // Cross-request notifications need a channel this SDK does not
-            // define yet, so the stream only waits. The tick is not optional:
-            // PHP spots a dropped peer by writing, and a sleeping loop would
-            // pin an FPM worker for the full lifetime.
-            $deadline = microtime(true) + $lifetime;
+            // From now, not from the beginning: a subscriber wants what happens
+            // next, not a replay of the server's history.
+            $cursor = $bus?->cursor() ?? 0;
+
+            // The tick is not optional: PHP spots a dropped peer by writing,
+            // and a sleeping loop would pin an FPM worker for the full lifetime.
+            $deadline = 0.0 >= $lifetime ? \INF : microtime(true) + $lifetime;
+
            while (microtime(true) < $deadline) {
+                if (null !== $bus) {
+                    [$notifications, $cursor] = $bus->since($cursor);
+
+                    foreach ($notifications as $notification) {
+                        if (!$agreed->carries($notification)) {
+                            continue;
+                        }
+
+                        yield self::tagWithSubscription($notification, $id);
+                    }
+                }
+
                yield null;

                if (connection_aborted()) {
@ -326,14 +347,36 @@ final class StatelessProtocol
            yield [
                'jsonrpc' => '2.0',
                'id' => $id,
-                'result' => [
+                'result' => $codec->encodeResult(self::LISTEN_METHOD, [
                    'resultType' => 'complete',
                    '_meta' => [RequestMeta::SUBSCRIPTION_ID => $id],
-                ],
+                ], false),
            ];
        });
    }

+    /**
+     * Every message on a listen stream carries the id of the subscription it
+     * belongs to, which is how a client demultiplexes them on stdio — where
+     * they all share one channel.
+     *
+     * @return array<string, mixed>
+     */
+    private static function tagWithSubscription(Notification $notification, string|int $id): array
+    {
+        /** @var array<string, mixed> $frame */
+        $frame = $notification->jsonSerialize();
+
+        $params = \is_array($frame['params'] ?? null) ? $frame['params'] : [];
+        $meta = \is_array($params['_meta'] ?? null) ? $params['_meta'] : [];
+
+        $meta[RequestMeta::SUBSCRIPTION_ID] = $id;
+        $params['_meta'] = $meta;
+        $frame['params'] = $params;
+
+        return $frame;
+    }
+
    private function discover(): DiscoverResult
    {
        return new DiscoverResult(
--- a/src/Server/Subscription/InMemoryNotificationBus.php
+++ b/src/Server/Subscription/InMemoryNotificationBus.php
@ -0,0 +1,74 @@
+<?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\Subscription;
+
+use Mcp\Exception\InvalidArgumentException;
+use Mcp\Schema\JsonRpc\Notification;
+
+/**
+ * A notification bus held in one process's memory.
+ *
+ * Correct for stdio and for persistent runtimes where the whole server lives in
+ * one process (Swoole, FrankenPHP worker mode, ReactPHP). Under PHP-FPM it is
+ * not: the worker that publishes and the worker holding the stream open are
+ * different processes, and a notification published in one is invisible to the
+ * other. {@see Psr16NotificationBus} is the answer there.
+ *
+ * @author Christopher Hertel <mail@christopher-hertel.de>
+ */
+final class InMemoryNotificationBus implements NotificationBusInterface
+{
+    /** @var array<int, Notification> */
+    private array $entries = [];
+
+    private int $next = 0;
+
+    /**
+     * @param int $backlog how many notifications to keep before dropping the oldest
+     */
+    public function __construct(
+        private readonly int $backlog = 256,
+    ) {
+        if ($this->backlog < 1) {
+            throw new InvalidArgumentException(\sprintf('The notification backlog must be at least one entry, got %d.', $this->backlog));
+        }
+    }
+
+    public function publish(Notification $notification): void
+    {
+        $this->entries[$this->next] = $notification;
+        ++$this->next;
+
+        // A stream that went away must not make this grow forever.
+        if (\count($this->entries) > $this->backlog) {
+            $this->entries = \array_slice($this->entries, -$this->backlog, preserve_keys: true);
+        }
+    }
+
+    public function cursor(): int
+    {
+        return $this->next;
+    }
+
+    public function since(int $cursor): array
+    {
+        $found = [];
+
+        foreach ($this->entries as $sequence => $notification) {
+            if ($sequence >= $cursor) {
+                $found[] = $notification;
+            }
+        }
+
+        return [$found, $this->next];
+    }
+}
--- a/src/Server/Subscription/NotificationBusInterface.php
+++ b/src/Server/Subscription/NotificationBusInterface.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\Subscription;
+
+use Mcp\Schema\JsonRpc\Notification;
+
+/**
+ * Carries server-initiated notifications to whichever `subscriptions/listen`
+ * streams are open.
+ *
+ * A bus and not a registry of listeners, because the two ends are rarely in the
+ * same process: under PHP-FPM the request that changes the tool list and the
+ * request holding a listen stream open are different workers, so the only thing
+ * they can share is storage. The read side is a cursor rather than a
+ * subscription: a stream that opens at cursor `c` reads forward from `c`, and
+ * nothing has to know a stream exists before it does.
+ *
+ * Implementations are expected to drop old entries — a listen stream that goes
+ * away must not make the backlog grow without bound.
+ *
+ * @author Christopher Hertel <mail@christopher-hertel.de>
+ */
+interface NotificationBusInterface
+{
+    /**
+     * Publishes a notification to every stream currently reading forward.
+     */
+    public function publish(Notification $notification): void;
+
+    /**
+     * The cursor a stream opening now should start from.
+     *
+     * Deliberately "now" and not "the beginning": a client that subscribes
+     * wants what happens next, not a replay of everything the server has done.
+     */
+    public function cursor(): int;
+
+    /**
+     * Notifications published after $cursor, and the cursor to read from next.
+     *
+     * @return array{list<Notification>, int}
+     */
+    public function since(int $cursor): array;
+}
--- a/src/Server/Subscription/Psr16NotificationBus.php
+++ b/src/Server/Subscription/Psr16NotificationBus.php
@ -0,0 +1,113 @@
+<?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\Subscription;
+
+use Mcp\Exception\InvalidArgumentException;
+use Mcp\JsonRpc\MessageFactory;
+use Mcp\Schema\JsonRpc\Notification;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Psr\SimpleCache\CacheInterface;
+
+/**
+ * A notification bus over any PSR-16 cache, so publishers and listen streams in
+ * different processes can see each other.
+ *
+ * The sequence counter and the entries are separate keys: a publisher bumps the
+ * counter and writes its entry under the number it got. That is not atomic
+ * across processes — two publishers can take the same number and one entry is
+ * lost — which is the trade every PSR-16-only design makes, and acceptable
+ * here: a lost `tools/list_changed` costs a client one stale list until the TTL
+ * lapses, not correctness. A backend with atomic increments (Redis `INCR`)
+ * should implement {@see NotificationBusInterface} directly.
+ *
+ * @author Christopher Hertel <mail@christopher-hertel.de>
+ */
+final class Psr16NotificationBus implements NotificationBusInterface
+{
+    private const CURSOR_KEY = 'cursor';
+
+    private readonly MessageFactory $messageFactory;
+
+    /**
+     * @param string $prefix  namespace for this bus's keys, so one cache can carry several
+     * @param int    $ttl     how long an entry stays readable, in seconds
+     * @param int    $backlog how many entries a reader will look back over
+     */
+    public function __construct(
+        private readonly CacheInterface $cache,
+        private readonly string $prefix = 'mcp.notifications.',
+        private readonly int $ttl = 120,
+        private readonly int $backlog = 256,
+        private readonly LoggerInterface $logger = new NullLogger(),
+        ?MessageFactory $messageFactory = null,
+    ) {
+        if ($this->backlog < 1) {
+            throw new InvalidArgumentException(\sprintf('The notification backlog must be at least one entry, got %d.', $this->backlog));
+        }
+
+        $this->messageFactory = $messageFactory ?? MessageFactory::make();
+    }
+
+    public function publish(Notification $notification): void
+    {
+        $sequence = $this->cursor();
+
+        $this->cache->set($this->key((string) $sequence), json_encode($notification, \JSON_THROW_ON_ERROR), $this->ttl);
+        $this->cache->set($this->key(self::CURSOR_KEY), $sequence + 1, $this->ttl);
+    }
+
+    public function cursor(): int
+    {
+        $cursor = $this->cache->get($this->key(self::CURSOR_KEY), 0);
+
+        return \is_int($cursor) ? $cursor : 0;
+    }
+
+    public function since(int $cursor): array
+    {
+        $head = $this->cursor();
+
+        // A reader that fell far behind reads what is still there, not
+        // everything it missed.
+        $from = max($cursor, $head - $this->backlog);
+
+        $found = [];
+
+        for ($sequence = $from; $sequence < $head; ++$sequence) {
+            $raw = $this->cache->get($this->key((string) $sequence));
+
+            if (!\is_string($raw)) {
+                // Expired, or lost to a concurrent publisher taking the same
+                // number. Neither is worth failing the stream over.
+                continue;
+            }
+
+            try {
+                foreach ($this->messageFactory->create($raw) as $message) {
+                    if ($message instanceof Notification) {
+                        $found[] = $message;
+                    }
+                }
+            } catch (\Throwable $e) {
+                $this->logger->warning('Dropped an unreadable notification from the bus.', ['sequence' => $sequence, 'exception' => $e]);
+            }
+        }
+
+        return [$found, $head];
+    }
+
+    private function key(string $suffix): string
+    {
+        return $this->prefix.$suffix;
+    }
+}
--- a/src/Server/Subscription/PublishingEventDispatcher.php
+++ b/src/Server/Subscription/PublishingEventDispatcher.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\Subscription;
+
+use Psr\EventDispatcher\EventDispatcherInterface;
+
+/**
+ * Publishes registry change events onto a notification bus, then passes them on.
+ *
+ * PSR-14 splits dispatching from listener registration, and the SDK is only
+ * handed the dispatcher half — so it cannot register a listener on an
+ * application's own dispatcher. This decorator is the way round that: configure
+ * a bus and the SDK wraps whatever dispatcher it was given (or stands in for
+ * one), so a runtime `registerTool()` reaches a listening client without the
+ * caller wiring anything.
+ *
+ * An application with its own listener provider can skip this and register
+ * {@see RegistryChangePublisher::listeners()} itself.
+ *
+ * @author Christopher Hertel <mail@christopher-hertel.de>
+ */
+final class PublishingEventDispatcher implements EventDispatcherInterface
+{
+    /** @var array<class-string, callable(object): void> */
+    private readonly array $listeners;
+
+    public function __construct(
+        NotificationBusInterface $bus,
+        private readonly ?EventDispatcherInterface $inner = null,
+    ) {
+        $this->listeners = (new RegistryChangePublisher($bus))->listeners();
+    }
+
+    public function dispatch(object $event): object
+    {
+        $listener = $this->listeners[$event::class] ?? null;
+
+        if (null !== $listener) {
+            $listener($event);
+        }
+
+        return $this->inner?->dispatch($event) ?? $event;
+    }
+}
--- a/src/Server/Subscription/RegistryChangePublisher.php
+++ b/src/Server/Subscription/RegistryChangePublisher.php
@ -0,0 +1,81 @@
+<?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\Subscription;
+
+use Mcp\Event\PromptListChangedEvent;
+use Mcp\Event\ResourceListChangedEvent;
+use Mcp\Event\ResourceTemplateListChangedEvent;
+use Mcp\Event\ToolListChangedEvent;
+use Mcp\Schema\Notification\PromptListChangedNotification;
+use Mcp\Schema\Notification\ResourceListChangedNotification;
+use Mcp\Schema\Notification\ToolListChangedNotification;
+
+/**
+ * Turns the registry's own change events into the notifications a
+ * `subscriptions/listen` stream carries.
+ *
+ * The registry already announces every mutation to the PSR-14 dispatcher; this
+ * is the adapter that puts them where a stream can read them, so a runtime
+ * `registerTool()` reaches a listening client without the caller knowing a bus
+ * exists.
+ *
+ * There is no template-list notification in the protocol, so a changed
+ * resource-template list is announced as a changed resource list — which is
+ * what a client would refetch anyway.
+ *
+ * @author Christopher Hertel <mail@christopher-hertel.de>
+ */
+final class RegistryChangePublisher
+{
+    public function __construct(
+        private readonly NotificationBusInterface $bus,
+    ) {
+    }
+
+    public function onToolListChanged(ToolListChangedEvent $event): void
+    {
+        $this->bus->publish(new ToolListChangedNotification());
+    }
+
+    public function onPromptListChanged(PromptListChangedEvent $event): void
+    {
+        $this->bus->publish(new PromptListChangedNotification());
+    }
+
+    public function onResourceListChanged(ResourceListChangedEvent $event): void
+    {
+        $this->bus->publish(new ResourceListChangedNotification());
+    }
+
+    public function onResourceTemplateListChanged(ResourceTemplateListChangedEvent $event): void
+    {
+        $this->bus->publish(new ResourceListChangedNotification());
+    }
+
+    /**
+     * Every event this publisher handles, keyed by event class.
+     *
+     * Shaped for a dispatcher that wants a map; a framework's own subscriber
+     * conventions can read it too rather than restating the list.
+     *
+     * @return array<class-string, callable(object): void>
+     */
+    public function listeners(): array
+    {
+        return [
+            ToolListChangedEvent::class => $this->onToolListChanged(...),
+            PromptListChangedEvent::class => $this->onPromptListChanged(...),
+            ResourceListChangedEvent::class => $this->onResourceListChanged(...),
+            ResourceTemplateListChangedEvent::class => $this->onResourceTemplateListChanged(...),
+        ];
+    }
+}
--- a/tests/Conformance/server-stateless.php
+++ b/tests/Conformance/server-stateless.php
@ -22,6 +22,7 @@ require_once dirname(__DIR__, 2).'/vendor/autoload.php';

 use Http\Discovery\Psr17Factory;
 use Laminas\HttpHandlerRunner\Emitter\SapiEmitter;
+use Mcp\Capability\Registry;
 use Mcp\Exception\MissingRequiredClientCapabilityException;
 use Mcp\Schema\ClientCapabilities;
 use Mcp\Schema\Content\AudioContent;
@ -32,15 +33,21 @@ use Mcp\Schema\Elicitation\ElicitationSchema;
 use Mcp\Schema\Elicitation\StringSchemaDefinition;
 use Mcp\Schema\Enum\CacheScope;
 use Mcp\Schema\Enum\ProtocolVersion;
+use Mcp\Schema\Prompt;
 use Mcp\Schema\Request\ElicitRequest;
 use Mcp\Schema\Result\CallToolResult;
 use Mcp\Schema\Result\InputRequiredResult;
+use Mcp\Schema\Tool;
 use Mcp\Server;
+use Mcp\Server\Subscription\Psr16NotificationBus;
+use Mcp\Server\Subscription\PublishingEventDispatcher;
 use Mcp\Server\Transport\StatelessHttpTransport;
 use Mcp\Server\Wire\CachePolicy;
 use Mcp\Tests\Conformance\Elements;
 use Mcp\Tests\Conformance\FileLogger;
 use Mcp\Tests\Conformance\MrtrElements;
+use Symfony\Component\Cache\Adapter\FilesystemAdapter;
+use Symfony\Component\Cache\Psr16Cache;

 chdir(__DIR__);

@ -49,9 +56,22 @@ $logger = new FileLogger(__DIR__.'/logs/conformance-stateless.log', true);
 $psr17Factory = new Psr17Factory();
 $request = $psr17Factory->createServerRequestFromGlobals();

+// Explicit rather than builder-built, so the diagnostic hooks below can mutate
+// the live registry and the change reaches an open subscription.
+//
+// Filesystem-backed and not in-memory: under php-fpm the worker holding the
+// listen stream open and the worker serving the tools/call that mutates the
+// registry are different processes, so the only thing they share is storage.
+$bus = new Psr16NotificationBus(
+    new Psr16Cache(new FilesystemAdapter('mcp-conformance-notifications', 120, __DIR__.'/sessions')),
+    logger: $logger,
+);
+$registry = new Registry(new PublishingEventDispatcher($bus), $logger);
+
 $protocol = Server::builder()
    ->setServerInfo('mcp-conformance-test-server', '1.0.0')
    ->setLogger($logger)
+    ->setRegistry($registry)
    // Tools
    ->addTool(static fn () => 'This is a simple text response for testing.', name: 'test_simple_text', description: 'Tests simple text content response')
    ->addTool(static fn () => new ImageContent(Elements::TEST_IMAGE_BASE64, 'image/png'), name: 'test_image_content', description: 'Tests image content response')
@ -106,6 +126,38 @@ $protocol = Server::builder()
        ],
    )
    ->addTool([Elements::class, 'toolWithProgress'], name: 'test_tool_with_progress', description: 'Tests tool that reports progress notifications')
+    // Diagnostic hooks the subscription scenarios call to make the lists change
+    // while a listen stream is open.
+    ->addTool(
+        static function () use ($registry): string {
+            $registry->registerTool(
+                new Tool(
+                    name: 'test_ephemeral_tool_'.bin2hex(random_bytes(4)),
+                    title: null,
+                    inputSchema: ['type' => 'object', 'properties' => new stdClass(), 'required' => null],
+                    description: 'Registered to trigger a list change',
+                    annotations: null,
+                ),
+                static fn (): string => 'ephemeral',
+            );
+
+            return 'Tool list mutated.';
+        },
+        name: 'test_trigger_tool_change',
+        description: 'Registers a tool so the tool list changes',
+    )
+    ->addTool(
+        static function () use ($registry): string {
+            $registry->registerPrompt(
+                new Prompt('test_ephemeral_prompt_'.bin2hex(random_bytes(4)), null, 'Registered to trigger a list change'),
+                static fn (): array => [['role' => 'user', 'content' => 'ephemeral']],
+            );
+
+            return 'Prompt list mutated.';
+        },
+        name: 'test_trigger_prompt_change',
+        description: 'Registers a prompt so the prompt list changes',
+    )
    // Multi round-trip request tools (SEP-2322).
    ->addTool([MrtrElements::class, 'elicitation'], name: 'test_input_required_result_elicitation', description: 'MRTR: asks for a name via elicitation')
    ->addTool([MrtrElements::class, 'sampling'], name: 'test_input_required_result_sampling', description: 'MRTR: asks for a sampling completion')
@ -127,6 +179,13 @@ $protocol = Server::builder()
    ->addPrompt([MrtrElements::class, 'prompt'], name: 'test_input_required_result_prompt', description: 'MRTR: a prompt that asks for input first')
    // Fixed so a retry landing on another process still verifies.
    ->setRequestState(str_repeat('conformance-fixture-key-', 2))
+    // So a listen stream carries the registry's changes rather than only
+    // acknowledging. In-memory is right here: the conformance server is one
+    // FrankenPHP-less php-fpm pool, and the scenarios publish within a request.
+    ->setNotificationBus($bus)
+    // Short, so a listen stream cannot tie up an fpm worker for the length of
+    // a whole run.
+    ->setSubscriptionLifetime(5.0)
    // Lists are the same for everyone here; a read is not.
    ->setCachePolicy(
        CachePolicy::default(60_000)
--- a/tests/Unit/Server/Stateless/StatelessProtocolTest.php
+++ b/tests/Unit/Server/Stateless/StatelessProtocolTest.php
@ -20,14 +20,19 @@ use Mcp\Schema\Enum\CacheScope;
 use Mcp\Schema\Enum\LoggingLevel;
 use Mcp\Schema\Enum\ProtocolVersion;
 use Mcp\Schema\JsonRpc\Error;
+use Mcp\Schema\Notification\PromptListChangedNotification;
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
+use Mcp\Schema\Notification\ToolListChangedNotification;
 use Mcp\Schema\Request\ElicitRequest;
 use Mcp\Schema\Result\InputRequiredResult;
 use Mcp\Schema\Result\ReadResourceResult;
+use Mcp\Schema\ServerCapabilities;
 use Mcp\Server;
 use Mcp\Server\RequestContext;
 use Mcp\Server\Stateless\RequestMeta;
 use Mcp\Server\Stateless\StatelessProtocol;
 use Mcp\Server\Stateless\StatelessResult;
+use Mcp\Server\Subscription\InMemoryNotificationBus;
 use Mcp\Server\Wire\CachePolicy;
 use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\TestDox;
@ -610,6 +615,119 @@ class StatelessProtocolTest extends TestCase
        $this->assertArrayNotHasKey('cacheScope', $answer['body']['result']);
    }

+    #[TestDox('a listen stream acknowledges, then carries what the client asked for')]
+    public function testListenStreamDeliversSubscribedNotifications(): void
+    {
+        $bus = new InMemoryNotificationBus();
+
+        $protocol = Server::builder()
+            ->setServerInfo('test-server', '1.0.0')
+            ->setCapabilities(new ServerCapabilities(toolsListChanged: true, resourcesListChanged: true, resourcesSubscribe: true))
+            ->setNotificationBus($bus)
+            ->setSubscriptionLifetime(0.8)
+            ->addTool(static fn (): string => 'ok', name: 'a_tool', description: 'x')
+            ->buildStateless([ProtocolVersion::V2026_07_28]);
+
+        $result = $protocol->handle(json_encode([
+            'jsonrpc' => '2.0',
+            'id' => 42,
+            'method' => 'subscriptions/listen',
+            'params' => [
+                'notifications' => ['toolsListChanged' => true, 'resourceSubscriptions' => ['file:///watched']],
+                '_meta' => [
+                    RequestMeta::PROTOCOL_VERSION => ProtocolVersion::V2026_07_28->value,
+                    RequestMeta::CLIENT_CAPABILITIES => new \stdClass(),
+                ],
+            ],
+        ], \JSON_THROW_ON_ERROR), [
+            'MCP-Protocol-Version' => ProtocolVersion::V2026_07_28->value,
+            'Mcp-Method' => 'subscriptions/listen',
+        ]);
+
+        $this->assertTrue($result->isStream());
+
+        $frames = [];
+        $published = false;
+
+        foreach (($result->frames)() as $frame) {
+            if (null === $frame) {
+                // Publish once the stream is established, so the notifications
+                // arrive the way a concurrent request would deliver them.
+                if (!$published) {
+                    $bus->publish(new ToolListChangedNotification());
+                    $bus->publish(new PromptListChangedNotification());
+                    $bus->publish(new ResourceUpdatedNotification('file:///watched'));
+                    $bus->publish(new ResourceUpdatedNotification('file:///ignored'));
+                    $published = true;
+                }
+
+                continue;
+            }
+
+            $frames[] = $frame;
+        }
+
+        $methods = array_map(static fn (array $frame): string => $frame['method'] ?? 'response', $frames);
+
+        // The acknowledgment MUST come first, the declined types must not come
+        // at all, and the graceful closure ends it.
+        $this->assertSame([
+            'notifications/subscriptions/acknowledged',
+            'notifications/tools/list_changed',
+            'notifications/resources/updated',
+            'response',
+        ], $methods);
+
+        $this->assertSame(['toolsListChanged' => true, 'resourceSubscriptions' => ['file:///watched']], (array) $frames[0]['params']['notifications']);
+        $this->assertSame('file:///watched', $frames[2]['params']['uri']);
+
+        // Every message on the stream names the subscription it belongs to.
+        foreach ([$frames[0], $frames[1], $frames[2]] as $frame) {
+            $this->assertSame(42, $frame['params']['_meta'][RequestMeta::SUBSCRIPTION_ID]);
+        }
+
+        $this->assertSame(42, $frames[3]['id']);
+        $this->assertSame(42, $frames[3]['result']['_meta'][RequestMeta::SUBSCRIPTION_ID]);
+        $this->assertSame('complete', $frames[3]['result']['resultType']);
+    }
+
+    #[TestDox('the acknowledgment drops types the server cannot honour')]
+    public function testAcknowledgmentReflectsWhatTheServerCanDo(): void
+    {
+        $protocol = Server::builder()
+            ->setServerInfo('test-server', '1.0.0')
+            ->setCapabilities(new ServerCapabilities(toolsListChanged: true, promptsListChanged: false))
+            ->setNotificationBus(new InMemoryNotificationBus())
+            ->setSubscriptionLifetime(0.05)
+            ->buildStateless([ProtocolVersion::V2026_07_28]);
+
+        $result = $protocol->handle(json_encode([
+            'jsonrpc' => '2.0',
+            'id' => 1,
+            'method' => 'subscriptions/listen',
+            'params' => [
+                'notifications' => ['toolsListChanged' => true, 'promptsListChanged' => true],
+                '_meta' => [
+                    RequestMeta::PROTOCOL_VERSION => ProtocolVersion::V2026_07_28->value,
+                    RequestMeta::CLIENT_CAPABILITIES => new \stdClass(),
+                ],
+            ],
+        ], \JSON_THROW_ON_ERROR), [
+            'MCP-Protocol-Version' => ProtocolVersion::V2026_07_28->value,
+            'Mcp-Method' => 'subscriptions/listen',
+        ]);
+
+        $first = null;
+        foreach (($result->frames)() as $frame) {
+            if (null !== $frame) {
+                $first = $frame;
+                break;
+            }
+        }
+
+        $this->assertSame(['toolsListChanged' => true], (array) $first['params']['notifications']);
+    }
+
    #[TestDox('a notification is acknowledged with no body, never answered')]
    public function testNotificationIsAcknowledged(): void
    {
--- a/tests/Unit/Server/Subscription/NotificationBusTest.php
+++ b/tests/Unit/Server/Subscription/NotificationBusTest.php
@ -0,0 +1,217 @@
+<?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\Tests\Unit\Server\Subscription;
+
+use Mcp\Event\ToolListChangedEvent;
+use Mcp\Exception\InvalidArgumentException;
+use Mcp\Schema\Notification\PromptListChangedNotification;
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
+use Mcp\Schema\Notification\ToolListChangedNotification;
+use Mcp\Server\Subscription\InMemoryNotificationBus;
+use Mcp\Server\Subscription\NotificationBusInterface;
+use Mcp\Server\Subscription\Psr16NotificationBus;
+use Mcp\Server\Subscription\PublishingEventDispatcher;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\Attributes\TestDox;
+use PHPUnit\Framework\TestCase;
+use Psr\SimpleCache\CacheInterface;
+
+class NotificationBusTest extends TestCase
+{
+    /**
+     * @return iterable<string, array{NotificationBusInterface}>
+     */
+    public static function buses(): iterable
+    {
+        yield 'in memory' => [new InMemoryNotificationBus()];
+        yield 'psr-16' => [new Psr16NotificationBus(self::arrayCache())];
+    }
+
+    #[DataProvider('buses')]
+    #[TestDox('a subscriber reads what is published after it opened')]
+    public function testReadsForwardFromItsCursor(NotificationBusInterface $bus): void
+    {
+        $bus->publish(new ToolListChangedNotification());
+
+        // Opened after the first notification, so it must not see it.
+        $cursor = $bus->cursor();
+
+        $bus->publish(new PromptListChangedNotification());
+        $bus->publish(new ResourceUpdatedNotification('file:///a'));
+
+        [$found, $next] = $bus->since($cursor);
+
+        $this->assertCount(2, $found);
+        $this->assertInstanceOf(PromptListChangedNotification::class, $found[0]);
+        $this->assertInstanceOf(ResourceUpdatedNotification::class, $found[1]);
+        $this->assertSame('file:///a', $found[1]->uri);
+
+        // Reading again from the returned cursor yields nothing new.
+        [$again] = $bus->since($next);
+        $this->assertSame([], $again);
+    }
+
+    #[DataProvider('buses')]
+    #[TestDox('a quiet bus hands back nothing')]
+    public function testQuietBusIsEmpty(NotificationBusInterface $bus): void
+    {
+        [$found] = $bus->since($bus->cursor());
+
+        $this->assertSame([], $found);
+    }
+
+    #[DataProvider('buses')]
+    #[TestDox('two subscribers at different cursors each get their own view')]
+    public function testCursorsAreIndependent(NotificationBusInterface $bus): void
+    {
+        $early = $bus->cursor();
+        $bus->publish(new ToolListChangedNotification());
+        $late = $bus->cursor();
+        $bus->publish(new PromptListChangedNotification());
+
+        $this->assertCount(2, $bus->since($early)[0]);
+        $this->assertCount(1, $bus->since($late)[0]);
+    }
+
+    #[TestDox('the backlog is bounded, so a stream that went away cannot grow it forever')]
+    public function testBacklogIsBounded(): void
+    {
+        $bus = new InMemoryNotificationBus(backlog: 3);
+
+        $cursor = $bus->cursor();
+        for ($i = 0; $i < 10; ++$i) {
+            $bus->publish(new ToolListChangedNotification());
+        }
+
+        $this->assertCount(3, $bus->since($cursor)[0]);
+    }
+
+    #[TestDox('a backlog below one entry is refused')]
+    public function testZeroBacklogIsRefused(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        new InMemoryNotificationBus(backlog: 0);
+    }
+
+    #[TestDox('a registry change event becomes a notification on the bus')]
+    public function testRegistryEventsArePublished(): void
+    {
+        $bus = new InMemoryNotificationBus();
+        $cursor = $bus->cursor();
+
+        (new PublishingEventDispatcher($bus))->dispatch(new ToolListChangedEvent());
+
+        [$found] = $bus->since($cursor);
+
+        $this->assertCount(1, $found);
+        $this->assertInstanceOf(ToolListChangedNotification::class, $found[0]);
+    }
+
+    #[TestDox('an event the publisher does not know is passed on untouched')]
+    public function testUnknownEventIsPassedThrough(): void
+    {
+        $bus = new InMemoryNotificationBus();
+        $cursor = $bus->cursor();
+        $event = new \stdClass();
+
+        $this->assertSame($event, (new PublishingEventDispatcher($bus))->dispatch($event));
+        $this->assertSame([], $bus->since($cursor)[0]);
+    }
+
+    #[TestDox('an inner dispatcher still sees every event')]
+    public function testInnerDispatcherStillRuns(): void
+    {
+        $seen = [];
+        $inner = new class($seen) implements \Psr\EventDispatcher\EventDispatcherInterface {
+            /** @param array<int, object> $seen */
+            public function __construct(public array &$seen)
+            {
+            }
+
+            public function dispatch(object $event): object
+            {
+                $this->seen[] = $event;
+
+                return $event;
+            }
+        };
+
+        (new PublishingEventDispatcher(new InMemoryNotificationBus(), $inner))->dispatch(new ToolListChangedEvent());
+
+        $this->assertCount(1, $inner->seen);
+    }
+
+    private static function arrayCache(): CacheInterface
+    {
+        return new class implements CacheInterface {
+            /** @var array<string, mixed> */
+            private array $values = [];
+
+            public function get(string $key, mixed $default = null): mixed
+            {
+                return $this->values[$key] ?? $default;
+            }
+
+            public function set(string $key, mixed $value, int|\DateInterval|null $ttl = null): bool
+            {
+                $this->values[$key] = $value;
+
+                return true;
+            }
+
+            public function delete(string $key): bool
+            {
+                unset($this->values[$key]);
+
+                return true;
+            }
+
+            public function clear(): bool
+            {
+                $this->values = [];
+
+                return true;
+            }
+
+            public function getMultiple(iterable $keys, mixed $default = null): iterable
+            {
+                foreach ($keys as $key) {
+                    yield $key => $this->get($key, $default);
+                }
+            }
+
+            public function setMultiple(iterable $values, int|\DateInterval|null $ttl = null): bool
+            {
+                foreach ($values as $key => $value) {
+                    $this->set((string) $key, $value, $ttl);
+                }
+
+                return true;
+            }
+
+            public function deleteMultiple(iterable $keys): bool
+            {
+                foreach ($keys as $key) {
+                    $this->delete($key);
+                }
+
+                return true;
+            }
+
+            public function has(string $key): bool
+            {
+                return \array_key_exists($key, $this->values);
+            }
+        };
+    }
+}