diff --git a/src/Server/Fiber/HandlerFibers.php b/src/Server/Fiber/HandlerFibers.php index c83eb62a..ae007a36 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 90c40e81..5dd4e8d3 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 20c0aae6..026480fc 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 3482220d..98b3a97f 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 7aaa4ea2..d4288e29 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 dd9c28f8..5d03d8c4 100644 --- a/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php +++ b/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php @@ -1,204 +1,225 @@ -exchange([ - self::call(2, 'count', 'a'), - self::call(3, 'count', 'b'), - self::call(4, 'count', 'c'), - ]); - - foreach ([2, 3, 4] as $id) { - $this->assertSame('counted', $this->answer($lines, $id)['result']['content'][0]['text'] ?? null, \sprintf('call %d did not finish', $id)); - } - - $progress = array_filter($lines, static fn (array $line): bool => 'notifications/progress' === ($line['method'] ?? null)); - $this->assertCount(6, $progress); - } - - #[TestDox('concurrent elicitations each resume with their own answer')] - public function testConcurrentElicitationsResumeWithTheirOwnAnswer(): void - { - // Request ids are random, so the client answers once both elicitations went out, the second one first. - $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']]], - ]); - - $elicitations = array_values(array_filter($lines, static fn (array $line): bool => 'elicitation/create' === ($line['method'] ?? null))); - $this->assertCount(2, $elicitations); - - $this->assertSame('first', $this->answer($lines, 2)['result']['content'][0]['text'] ?? null); - $this->assertSame('second', $this->answer($lines, 3)['result']['content'][0]['text'] ?? null); - } - - #[TestDox('notifications a call sent reach the client before its answer')] - public function testNotificationsPrecedeTheAnswer(): void - { - $lines = $this->exchange([self::call(2, 'announce')]); - - $this->assertSame(['notifications/message', null], array_map(static fn (array $line): ?string => $line['method'] ?? null, \array_slice($lines, 1))); - $this->assertSame('announced', $lines[2]['result']['content'][0]['text'] ?? null); - } - - /** - * @param list> $lines - * - * @return array - */ - private function answer(array $lines, int $id): array - { - foreach ($lines as $line) { - if ($id === ($line['id'] ?? null) && !isset($line['method'])) { - return $line; - } - } - - return []; - } - - /** - * @return array - */ - private static function call(int $id, string $tool, ?string $progressToken = null): array - { - $params = ['name' => $tool, 'arguments' => new \stdClass()]; - if (null !== $progressToken) { - $params['_meta'] = ['progressToken' => $progressToken]; - } - - return ['jsonrpc' => '2.0', 'id' => $id, 'method' => 'tools/call', 'params' => $params]; - } - - /** - * Runs a handshake-era session with the given messages after the handshake. - * - * @param list> $messages - * @param (\Closure(int|string, int|string): list>)|null $answer the client's messages once the server sent two requests, given their ids - * - * @return list> - */ - private function exchange(array $messages, ?\Closure $answer = null): array - { - $input = fopen('php://temp', 'r+'); - $this->assertNotFalse($input); - - $requestIds = []; - $dispatcher = new class(static function (ServerRequestEvent $event) use ($input, $answer, &$requestIds): void { - $requestIds[] = $event->getRequest()->getId(); - if (null === $answer || 2 !== \count($requestIds)) { - return; - } - - // Appended behind what the server has yet to read. - $position = ftell($input); - fseek($input, 0, \SEEK_END); - foreach ($answer(...$requestIds) as $message) { - fwrite($input, json_encode($message, \JSON_THROW_ON_ERROR)."\n"); - } - fwrite($input, str_repeat("\n", 5)); - fseek($input, (int) $position); - }) implements EventDispatcherInterface { - public function __construct(private readonly \Closure $onServerRequest) - { - } - - public function dispatch(object $event): object - { - if ($event instanceof ServerRequestEvent) { - ($this->onServerRequest)($event); - } - - return $event; - } - }; - - $server = Server::builder() - ->setServerInfo('test-server', '1.0.0') - ->addTool(static function (RequestContext $context): string { - $context->getClientGateway()->progress(1, 2); - $context->getClientGateway()->progress(2, 2); - - return 'counted'; - }, name: 'count', description: 'Reports progress') - ->addTool(static function (RequestContext $context): string { - $result = $context->getClientGateway()->elicit('Name?', new ElicitationSchema(['name' => new StringSchemaDefinition('Name')], ['name'])); - - return $result->content['name'] ?? 'none'; - }, name: 'ask', description: 'Asks for a name') - ->addTool(static function (RequestContext $context): string { - $context->getClientGateway()->notify(new LoggingMessageNotification(LoggingLevel::Info, 'starting')); - - return 'announced'; - }, name: 'announce', description: 'Notifies, then answers') - ->setEventDispatcher($dispatcher) - ->build(); - - // A file rather than memory: running the server closes its streams. - $outputFile = tempnam(sys_get_temp_dir(), 'mcp-stdio'); - $output = fopen($outputFile, 'w'); - $this->assertNotFalse($output); - - $messages = [ - ['jsonrpc' => '2.0', 'id' => 1, 'method' => 'initialize', 'params' => [ - 'protocolVersion' => ProtocolVersion::V2025_11_25->value, - 'capabilities' => ['elicitation' => new \stdClass()], - 'clientInfo' => ['name' => 'test-client', 'version' => '1.0.0'], - ]], - ['jsonrpc' => '2.0', 'method' => 'notifications/initialized'], - ...$messages, - ]; - - foreach ($messages as $message) { - fwrite($input, json_encode($message, \JSON_THROW_ON_ERROR)."\n"); - } - - // Blank lines give the run loop a few more turns before the input ends. - fwrite($input, str_repeat("\n", 5)); - rewind($input); - - $server->run(new StdioTransport($input, $output)); - - $written = (string) file_get_contents($outputFile); - unlink($outputFile); - - return array_map( - static fn (string $line): array => json_decode($line, true, flags: \JSON_THROW_ON_ERROR), - array_values(array_filter(explode("\n", $written))), - ); - } -} +exchange([ + self::call(2, 'count', 'a'), + self::call(3, 'count', 'b'), + self::call(4, 'count', 'c'), + ]); + + foreach ([2, 3, 4] as $id) { + $this->assertSame('counted', $this->answer($lines, $id)['result']['content'][0]['text'] ?? null, \sprintf('call %d did not finish', $id)); + } + + $progress = array_filter($lines, static fn (array $line): bool => 'notifications/progress' === ($line['method'] ?? null)); + $this->assertCount(6, $progress); + } + + #[TestDox('concurrent elicitations each resume with their own answer')] + public function testConcurrentElicitationsResumeWithTheirOwnAnswer(): void + { + // Request ids are random, so the client answers once both elicitations went out, the second one first. + $lines = $this->exchange([ + self::call(2, 'ask'), + self::call(3, 'ask'), + ], static fn (int|string $first, int|string|null $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); + + $this->assertSame('first', $this->answer($lines, 2)['result']['content'][0]['text'] ?? null); + $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 + { + $lines = $this->exchange([self::call(2, 'announce')]); + + $this->assertSame(['notifications/message', null], array_map(static fn (array $line): ?string => $line['method'] ?? null, \array_slice($lines, 1))); + $this->assertSame('announced', $lines[2]['result']['content'][0]['text'] ?? null); + } + + /** + * @param list> $lines + * + * @return array + */ + private function answer(array $lines, int $id): array + { + foreach ($lines as $line) { + if ($id === ($line['id'] ?? null) && !isset($line['method'])) { + return $line; + } + } + + return []; + } + + /** + * @return array + */ + private static function call(int $id, string $tool, ?string $progressToken = null): array + { + $params = ['name' => $tool, 'arguments' => new \stdClass()]; + if (null !== $progressToken) { + $params['_meta'] = ['progressToken' => $progressToken]; + } + + return ['jsonrpc' => '2.0', 'id' => $id, 'method' => 'tools/call', 'params' => $params]; + } + + /** + * Runs a handshake-era session with the given messages after the handshake. + * + * @param list> $messages + * @param (\Closure(int|string, int|string): list>)|null $answer the client's messages once the server sent two requests, given their ids + * + * @return list> + */ + private function exchange(array $messages, ?\Closure $answer = null): array + { + $input = fopen('php://temp', 'r+'); + $this->assertNotFalse($input); + + $requestIds = []; + $dispatcher = new class(static function (ServerRequestEvent $event) use ($input, $answer, &$requestIds): void { + $requestIds[] = $event->getRequest()->getId(); + if (null === $answer) { + return; + } + + // Appended behind what the server has yet to read. + $position = ftell($input); + fseek($input, 0, \SEEK_END); + foreach ($answer(...$requestIds) as $message) { + fwrite($input, json_encode($message, \JSON_THROW_ON_ERROR)."\n"); + } + fwrite($input, str_repeat("\n", 5)); + fseek($input, (int) $position); + }) implements EventDispatcherInterface { + public function __construct(private readonly \Closure $onServerRequest) + { + } + + public function dispatch(object $event): object + { + if ($event instanceof ServerRequestEvent) { + ($this->onServerRequest)($event); + } + + return $event; + } + }; + + $server = Server::builder() + ->setServerInfo('test-server', '1.0.0') + ->addTool(static function (RequestContext $context): string { + $context->getClientGateway()->progress(1, 2); + $context->getClientGateway()->progress(2, 2); + + return 'counted'; + }, name: 'count', description: 'Reports progress') + ->addTool(static function (RequestContext $context): string { + $result = $context->getClientGateway()->elicit('Name?', new ElicitationSchema(['name' => new StringSchemaDefinition('Name')], ['name'])); + + return $result->content['name'] ?? 'none'; + }, name: 'ask', description: 'Asks for a name') + ->addTool(static function (RequestContext $context): string { + $context->getClientGateway()->notify(new LoggingMessageNotification(LoggingLevel::Info, 'starting')); + + return 'announced'; + }, name: 'announce', description: 'Notifies, then answers') + ->setEventDispatcher($dispatcher) + ->build(); + + // A file rather than memory: running the server closes its streams. + $outputFile = tempnam(sys_get_temp_dir(), 'mcp-stdio'); + $output = fopen($outputFile, 'w'); + $this->assertNotFalse($output); + + $messages = [ + ['jsonrpc' => '2.0', 'id' => 1, 'method' => 'initialize', 'params' => [ + 'protocolVersion' => ProtocolVersion::V2025_11_25->value, + 'capabilities' => ['elicitation' => new \stdClass()], + 'clientInfo' => ['name' => 'test-client', 'version' => '1.0.0'], + ]], + ['jsonrpc' => '2.0', 'method' => 'notifications/initialized'], + ...$messages, + ]; + + foreach ($messages as $message) { + fwrite($input, json_encode($message, \JSON_THROW_ON_ERROR)."\n"); + } + + // Blank lines give the run loop a few more turns before the input ends. + fwrite($input, str_repeat("\n", 5)); + rewind($input); + + $server->run(new StdioTransport($input, $output)); + + $written = (string) file_get_contents($outputFile); + unlink($outputFile); + + return array_map( + static fn (string $line): array => json_decode($line, true, flags: \JSON_THROW_ON_ERROR), + array_values(array_filter(explode("\n", $written))), + ); + } +}