From 474793e246123e2e65e54360cf2da3fc4c1d9124 Mon Sep 17 00:00:00 2001 From: Ajendra Date: Sat, 14 Nov 2026 12:13:03 +0530 Subject: [PATCH] Handle notifications/cancelled on the handshake era Fixes #732 --- src/Server/Fiber/HandlerFibers.php | 9 ++++ src/Server/Protocol.php | 53 ++++++++++++++++++- src/Server/Wire/EraClassification.php | 8 +++ src/Server/Wire/InboundClassifier.php | 4 ++ tests/Integration/CancellationTest.php | 2 +- .../Transport/StdioSuspendedFibersTest.php | 31 +++++++++-- 6 files changed, 100 insertions(+), 7 deletions(-) diff --git a/src/Server/Fiber/HandlerFibers.php b/src/Server/Fiber/HandlerFibers.php index c83eb62aa..ae007a368 100644 --- a/src/Server/Fiber/HandlerFibers.php +++ b/src/Server/Fiber/HandlerFibers.php @@ -115,6 +115,15 @@ public static function isAbandoned(): bool return (self::states()[$fiber] ?? null)?->isAbandoned() ?? false; } + /** + * Marks a suspended handler fiber as abandoned so it can be terminated + * without sending a response, e.g. on notifications/cancelled. + */ + public static function abandon(\Fiber $fiber): void + { + (self::states()[$fiber] ?? null)?->abandoned(); + } + /** * Suspends the current handler fiber for the server, until the server resumes it: how a handler waits on the client. * diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 90c40e81a..5dd4e8d34 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -23,6 +23,7 @@ use Mcp\Schema\Request\ResourceSubscribeRequest; use Mcp\Schema\Request\ResourceUnsubscribeRequest; use Mcp\Schema\Request\SetLogLevelRequest; +use Mcp\Schema\Notification\CancelledNotification; use Mcp\Server\Authorization\AccessToken; use Mcp\Server\Event\ClientResponseEvent; use Mcp\Server\Event\ErrorEvent; @@ -135,6 +136,7 @@ class Protocol private array $droppedAwaitedRequestIds = []; private readonly ?SessionLockInterface $sessionLock; + private readonly array $notificationHandlers; /** * @param array>> $requestHandlers @@ -144,7 +146,7 @@ class Protocol */ public function __construct( private readonly array $requestHandlers, - private readonly array $notificationHandlers, + array $notificationHandlers, private readonly MessageFactory $messageFactory, private readonly SessionManagerInterface $sessionManager, private readonly LoggerInterface $logger = new NullLogger(), @@ -156,6 +158,24 @@ public function __construct( private readonly ?FiberContextInterface $fiberContext = null, private readonly bool $sessionless = false, ) { + $this->notificationHandlers = [ + ...$notificationHandlers, + new class($this) implements NotificationHandlerInterface { + public function __construct(private readonly Protocol $protocol) + { + } + + public function supports(Notification $notification): bool + { + return $notification instanceof CancelledNotification; + } + + public function handle(Notification $notification, SessionInterface $session): void + { + $this->protocol->handleCancelledNotification($notification, $session); + } + }, + ]; $this->sessionLock = null !== $sessionLock && null !== $fiberScheduler ? new EventLoopSessionLock($sessionLock, $fiberScheduler) : $sessionLock; $this->awaitedRequestIds = new \WeakMap(); $this->fiberOutgoingMessages = new \WeakMap(); @@ -616,6 +636,37 @@ private function handleNotification(Notification $notification, SessionInterface } } + private function handleCancelledNotification(CancelledNotification $notification, SessionInterface $session): void + { + $requestId = $notification->requestId; + + $this->logger->info('Handling cancelled notification.', ['request_id' => $requestId]); + + foreach ($this->awaitedRequestIds as $fiber => $awaitedId) { + if ($awaitedId->toRfc4122() === $requestId) { + HandlerFibers::abandon($fiber); + + if ($fiber->isSuspended() && !$fiber->isTerminated()) { + try { + $fiber->throw(new \RuntimeException('The request has been cancelled.')); + } catch (\Throwable) { + // Fiber terminated without producing a response, per the cancellation utility. + } + } + + $this->settleAwaitedRequest($requestId); + + if (!$session instanceof EphemeralSessionInterface) { + $session->forget(self::SESSION_PENDING_REQUESTS . ".{$requestId}"); + } + + return; + } + } + + $this->logger->debug('No matching suspended fiber for cancelled notification.', ['request_id' => $requestId]); + } + /** * Sends a request to the client and returns the request ID. */ diff --git a/src/Server/Wire/EraClassification.php b/src/Server/Wire/EraClassification.php index 20c0aae68..026480fc7 100644 --- a/src/Server/Wire/EraClassification.php +++ b/src/Server/Wire/EraClassification.php @@ -39,6 +39,14 @@ public static function legacy(?string $claimedVersion = null): self return new self(false, $claimedVersion, null, 200); } + /** + * notifications/cancelled before initialize to ensure no response is sent. + */ + public static function cancelled(): self + { + return new self(false, null, null, 204); + } + /** * The modern era, claiming $version — which may be one this server does not * serve. Routing and support are separate questions, and the dispatcher diff --git a/src/Server/Wire/InboundClassifier.php b/src/Server/Wire/InboundClassifier.php index 3482220d7..98b3a97f2 100644 --- a/src/Server/Wire/InboundClassifier.php +++ b/src/Server/Wire/InboundClassifier.php @@ -104,6 +104,10 @@ public static function crossCheckVersion(?string $headerVersion, string $claimed */ private function classifyMessage(array $message, ?string $headerVersion): EraClassification { + if ('notifications/cancelled' === ($message['method'] ?? null)) { + return EraClassification::cancelled(); + } + $id = $message['id'] ?? null; $isNotification = !\array_key_exists('id', $message) || null === $id; diff --git a/tests/Integration/CancellationTest.php b/tests/Integration/CancellationTest.php index 7aaa4ea26..d4288e29d 100644 --- a/tests/Integration/CancellationTest.php +++ b/tests/Integration/CancellationTest.php @@ -18,7 +18,7 @@ final class CancellationTest extends IntegrationTestCase { - #[TestDox('cancelling a pending stdio call notifies the server and leaves the connection usable')] + #[TestDox('cancelling a pending stdio call notifies the handshake-era server and leaves the connection usable')] public function testCancelsPendingCall(): void { $log = tempnam(sys_get_temp_dir(), 'mcp-cancel-'); diff --git a/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php b/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php index dd9c28f81..6bf4b15a4 100644 --- a/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php +++ b/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php @@ -54,10 +54,16 @@ public function testConcurrentElicitationsResumeWithTheirOwnAnswer(): void $lines = $this->exchange([ self::call(2, 'ask'), self::call(3, 'ask'), - ], static fn (int|string $first, int|string $second): array => [ - ['jsonrpc' => '2.0', 'id' => $second, 'result' => ['action' => 'accept', 'content' => ['name' => 'second']]], - ['jsonrpc' => '2.0', 'id' => $first, 'result' => ['action' => 'accept', 'content' => ['name' => 'first']]], - ]); + ], static fn (int|string $first, ?int|string $second = null): array => { + if (null === $second) { + return []; + } + + return [ + ['jsonrpc' => '2.0', 'id' => $second, 'result' => ['action' => 'accept', 'content' => ['name' => 'second']]], + ['jsonrpc' => '2.0', 'id' => $first, 'result' => ['action' => 'accept', 'content' => ['name' => 'first']]], + ]; + }); $elicitations = array_values(array_filter($lines, static fn (array $line): bool => 'elicitation/create' === ($line['method'] ?? null))); $this->assertCount(2, $elicitations); @@ -66,6 +72,21 @@ public function testConcurrentElicitationsResumeWithTheirOwnAnswer(): void $this->assertSame('second', $this->answer($lines, 3)['result']['content'][0]['text'] ?? null); } + #[TestDox('handshake-era cancellation drops the suspended fiber and sends no response')] + public function testHandshakeEraCancellation(): void + { + $lines = $this->exchange([ + self::call(2, 'ask'), + ], static fn (int|string $elicitationId): array => [ + ['jsonrpc' => '2.0', 'method' => 'notifications/cancelled', 'params' => ['requestId' => $elicitationId]], + ]); + + $elicitations = array_filter($lines, static fn (array $line): bool => 'elicitation/create' === ($line['method'] ?? null)); + $this->assertCount(1, $elicitations); + + $this->assertEmpty($this->answer($lines, 2), 'should send no response for cancelled request'); + } + #[TestDox('notifications a call sent reach the client before its answer')] public function testNotificationsPrecedeTheAnswer(): void { @@ -120,7 +141,7 @@ private function exchange(array $messages, ?\Closure $answer = null): array $requestIds = []; $dispatcher = new class(static function (ServerRequestEvent $event) use ($input, $answer, &$requestIds): void { $requestIds[] = $event->getRequest()->getId(); - if (null === $answer || 2 !== \count($requestIds)) { + if (null === $answer) { return; }