From 3e3e2d85e43b95ffeb5420fa665974248bb8db08 Mon Sep 17 00:00:00 2001 From: Illia Vasylevskyi Date: Sun, 27 Sep 2026 20:26:36 -0400 Subject: [PATCH] Support in-flight MCP tool call cancellation and deadlines --- docs/client/transports.md | 46 +++ src/Client.php | 35 +- src/Client/CancellationTokenInterface.php | 20 ++ src/Client/Protocol.php | 82 ++++- src/Client/Transport/HttpTransport.php | 76 +++- src/Client/Transport/StdioTransport.php | 46 ++- src/Client/Transport/TransportInterface.php | 2 +- src/Exception/RequestCancelledException.php | 19 + tests/Integration/CancellationTest.php | 111 ++++++ tests/Integration/Fixture/cancellation.php | 55 +++ tests/Unit/Client/ProtocolTest.php | 300 ++++++++++++++++ .../Client/Transport/HttpTransportTest.php | 336 ++++++++++++++++++ 12 files changed, 1093 insertions(+), 35 deletions(-) create mode 100644 src/Client/CancellationTokenInterface.php create mode 100644 src/Exception/RequestCancelledException.php create mode 100644 tests/Integration/CancellationTest.php create mode 100644 tests/Integration/Fixture/cancellation.php diff --git a/docs/client/transports.md b/docs/client/transports.md index 84e1500b..6d0b6a5d 100644 --- a/docs/client/transports.md +++ b/docs/client/transports.md @@ -62,3 +62,49 @@ The transport automatically discovers PSR-18 HTTP clients from: # Install any PSR-18 client - discovery works automatically composer require php-http/guzzle7-adapter ``` + +## Cancellation and deadlines + +`callTool()` accepts optional `cancellation: ?CancellationTokenInterface` and +`timeoutSeconds: ?float` arguments. The token's `isCancellationRequested()` +method must return without blocking. The timeout must be finite and positive +and replaces the default request timeout for this call. +An observed cancellation throws `RequestCancelledException`. An observed +per-call deadline expiry throws `TimeoutException`. + +Interruption is checked before a request goes out and again once the send +returns. The second check is what covers a synchronous `application/json` +answer: the reply is buffered while `send()` is still on the stack, and a call +the caller has given up on must not come back as successful. Its buffered reply +is dropped with the pending request, and the connection stays available for +later calls. + +STDIO checks for interruption while polling the server and sends +`notifications/cancelled` for an interrupted pending request. Late responses +are ignored. + +HTTP cancellation is cooperative, and what it signals depends on the revision: + +- up to `2025-11-25` a disconnect is not a cancellation, so the client sends + `notifications/cancelled` for the abandoned request. That notification is + another request on the same connection and can therefore block; it is best + effort, and a failure to send it is logged rather than reported in place of + the interruption. +- from `2026-07-28` closing the request's response stream is the signal, so no + separate notification goes out. + +Either way the transport closes the active response body on interruption and +clears the pending request without closing the MCP session. Closing a response +does not guarantee that server-side work stops, and physical socket cleanup +depends on the HTTP client. + +PSR-18 requests and PSR-7 body reads can block. Cancellation and deadlines +cannot interrupt those operations, including waiting for headers, reading a +JSON body, or waiting for the next SSE chunk. They take effect only after +control returns to the transport. A per-call deadline is therefore not a hard +HTTP wall-clock limit. Configure network timeouts on the underlying HTTP client +to bound blocking I/O. + +This keeps HTTP transport compatible with PSR-18 clients and uses the existing +SSE parser. It requires no framework-specific asynchronous client, at the cost +of delayed cancellation during blocking I/O. diff --git a/src/Client.php b/src/Client.php index 532f60a2..143ab206 100644 --- a/src/Client.php +++ b/src/Client.php @@ -12,10 +12,13 @@ namespace Mcp; use Mcp\Client\Builder; +use Mcp\Client\CancellationTokenInterface; use Mcp\Client\Configuration; use Mcp\Client\Protocol; use Mcp\Client\Transport\TransportInterface; use Mcp\Exception\ConnectionException; +use Mcp\Exception\InvalidArgumentException; +use Mcp\Exception\RequestCancelledException; use Mcp\Exception\RequestException; use Mcp\Exception\RuntimeException; use Mcp\Schema\Enum\LoggingLevel; @@ -190,13 +193,31 @@ public function listTools(?string $cursor = null): ListToolsResult /** * Call a tool on the server. * - * @param string $name Tool name - * @param array $arguments Tool arguments + * Cancellation and deadlines are cooperative: on HTTP, blocking I/O has to + * return before the transport can observe the interruption. An answer that + * arrived for an interrupted call is discarded rather than returned. + * + * @param string $name Tool name + * @param array $arguments Tool arguments * @param (callable(float $progress, ?float $total, ?string $message): void)|null $onProgress - * Optional callback for progress updates + * Optional callback for progress updates + * @param CancellationTokenInterface|null $cancellation Non-blocking cancellation signal + * @param float|null $timeoutSeconds Finite positive timeout replacing the default for this call + * + * @throws RequestCancelledException When cancellation is observed + * @throws Exception\TimeoutException When the per-call deadline is observed to have expired + * @throws RequestException|ConnectionException|InvalidArgumentException|RuntimeException */ - public function callTool(string $name, array $arguments = [], ?callable $onProgress = null): CallToolResult + public function callTool(string $name, array $arguments = [], ?callable $onProgress = null, ?CancellationTokenInterface $cancellation = null, ?float $timeoutSeconds = null): CallToolResult { + if ($cancellation?->isCancellationRequested()) { + throw new RequestCancelledException('The client cancelled the request.'); + } + + if (null !== $timeoutSeconds && (!is_finite($timeoutSeconds) || $timeoutSeconds <= 0)) { + throw new InvalidArgumentException('The per-call timeout must be a finite positive number of seconds.'); + } + $catalog = $this->protocol->getToolCatalog(); // A tool the listing showed to be malformed is refused here rather than @@ -208,7 +229,7 @@ public function callTool(string $name, array $arguments = [], ?callable $onProgr $request = new CallToolRequest($name, $arguments); - $response = $this->sendRequest($request, $onProgress); + $response = $this->sendRequest($request, $onProgress, $cancellation, $timeoutSeconds); return CallToolResult::fromArray($response->result); } @@ -337,14 +358,14 @@ public function sendRootsListChanged(): void * * @throws RequestException|ConnectionException */ - private function sendRequest(Request $request, ?callable $onProgress = null): Response + private function sendRequest(Request $request, ?callable $onProgress = null, ?CancellationTokenInterface $cancellation = null, ?float $timeoutSeconds = null): Response { if (!$this->isConnected()) { throw new ConnectionException('Client is not connected. Call connect() first.'); } $withProgress = null !== $onProgress; - $fiber = new \Fiber(fn () => $this->protocol->request($request, $this->config->requestTimeout, $withProgress)); + $fiber = new \Fiber(fn () => $this->protocol->request($request, $this->config->requestTimeout, $withProgress, $cancellation, $timeoutSeconds)); $response = $this->transport->runRequest($fiber, $onProgress); if ($response instanceof Error) { diff --git a/src/Client/CancellationTokenInterface.php b/src/Client/CancellationTokenInterface.php new file mode 100644 index 00000000..6790e718 --- /dev/null +++ b/src/Client/CancellationTokenInterface.php @@ -0,0 +1,20 @@ +>|Error */ - public function request(Request $request, int $timeout, bool $withProgress = false): Response|Error + public function request(Request $request, int $timeout, bool $withProgress = false, ?CancellationTokenInterface $cancellation = null, ?float $callTimeout = null): Response|Error { + $deadline = null !== $callTimeout ? microtime(true) + $callTimeout : null; $payload = $request->withId(0)->jsonSerialize(); unset($payload['id']); @@ -374,11 +379,11 @@ public function request(Request $request, int $timeout, bool $withProgress = fal } if (null === $this->envelope) { - return $this->exchange($payload, $timeout); + return $this->exchange($payload, $timeout, $cancellation, $deadline); } for ($attempt = 0; $attempt < self::MAX_ROUND_TRIPS; ++$attempt) { - $response = $this->exchange($payload, $timeout); + $response = $this->exchange($payload, $timeout, $cancellation, $deadline); if ($response instanceof Error) { $retry = $this->withAcceptedVersion($response); @@ -477,8 +482,12 @@ private function withAcceptedVersion(Error $error): ?ProtocolVersion * * @return Response>|Error */ - private function exchange(array $payload, int $timeout): Response|Error + private function exchange(array $payload, int $timeout, ?CancellationTokenInterface $cancellation = null, ?float $deadline = null): Response|Error { + if (null !== ($interruption = self::interruption($cancellation, $deadline))) { + throw $interruption; + } + $requestId = $this->state->nextRequestId(); $payload['id'] = $requestId; @@ -487,6 +496,12 @@ private function exchange(array $payload, int $timeout): Response|Error try { $this->send($payload, 'request'); + // send() can block and leave a JSON answer already buffered: drop it + // and report the interruption instead of a success nobody awaits. + if (null !== ($interruption = self::interruption($cancellation, $deadline))) { + throw $interruption; + } + $immediate = $this->state->consumeResponse($requestId); if (null !== $immediate) { $this->logger->debug('Received immediate response', ['id' => $requestId]); @@ -496,11 +511,25 @@ private function exchange(array $payload, int $timeout): Response|Error $this->logger->debug('Suspending fiber for response', ['id' => $requestId]); - return \Fiber::suspend([ + $response = \Fiber::suspend([ 'type' => 'await_response', 'request_id' => $requestId, 'timeout' => $timeout, + 'cancellation' => $cancellation, + 'deadline' => $deadline, ]); + + // A transport may resume with a buffered reply before checking interruption. + if (null !== ($interruption = self::interruption($cancellation, $deadline))) { + throw $interruption; + } + + return $response; + } catch (RequestCancelledException|TimeoutException $e) { + $this->state->consumeResponse($requestId); + $this->notifyCancellation($requestId, $e->getMessage()); + + throw $e; } finally { // Only the response path clears it, so a request that timed out or // whose send() threw would stay pending and fail every later one. @@ -508,6 +537,43 @@ private function exchange(array $payload, int $timeout): Response|Error } } + /** + * The interruption an in-flight request is subject to, if any. Checked on + * both sides of a send and after the transport resumes a suspended request. + * + * @phpstan-impure + */ + private static function interruption(?CancellationTokenInterface $cancellation, ?float $deadline): RequestCancelledException|TimeoutException|null + { + if ($cancellation?->isCancellationRequested()) { + return new RequestCancelledException('The client cancelled the request.'); + } + + if (null !== $deadline && microtime(true) >= $deadline) { + return new TimeoutException('The request deadline expired.'); + } + + return null; + } + + /** + * Tell the server an abandoned request's result will go unused. Only stdio and + * handshake-era HTTP need it: a modern connection signals by closing the + * response stream. Best effort — a send failure is logged, not raised. + */ + private function notifyCancellation(int $requestId, string $reason): void + { + if ($this->transport instanceof HttpTransport && true === $this->state->getProtocolVersion()?->isModern()) { + return; + } + + try { + $this->sendNotification(new CancelledNotification($requestId, $reason)); + } catch (\Throwable $notificationError) { + $this->logger->warning('Could not send request cancellation notification.', ['request_id' => $requestId, 'exception' => $notificationError]); + } + } + /** * Send a notification to the server (fire and forget). */ @@ -611,6 +677,12 @@ private function handleResponse(Response|Error $response): void return; } + if (!\array_key_exists($requestId, $this->state->getPendingRequests())) { + $this->logger->debug('Ignoring response for a request that is no longer pending.', ['id' => $requestId]); + + return; + } + $this->logger->debug('Handling response', ['id' => $requestId]); $this->state->storeResponse($requestId, $response->jsonSerialize()); diff --git a/src/Client/Transport/HttpTransport.php b/src/Client/Transport/HttpTransport.php index 3c12ea74..c0e455f9 100644 --- a/src/Client/Transport/HttpTransport.php +++ b/src/Client/Transport/HttpTransport.php @@ -15,6 +15,8 @@ use Http\Discovery\Psr18ClientDiscovery; use Mcp\Exception\ConnectionException; use Mcp\Exception\InvalidArgumentException; +use Mcp\Exception\RequestCancelledException; +use Mcp\Exception\TimeoutException; use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\JsonRpc\Response; use Psr\Http\Client\ClientInterface; @@ -29,6 +31,7 @@ * PSR-18 HTTP clients are auto-discovered if not provided. * * @phpstan-import-type McpFiber from TransportInterface + * @phpstan-import-type FiberSuspend from TransportInterface * * @author Kyrian Obikwelu */ @@ -46,6 +49,9 @@ class HttpTransport extends BaseTransport implements HeaderAwareTransportInterfa /** @var McpFiber|null */ private ?\Fiber $activeFiber = null; + /** @var FiberSuspend|null */ + private ?array $activeSuspend = null; + /** @var (callable(float, ?float, ?string): void)|null */ private $activeProgressCallback; @@ -173,6 +179,14 @@ public function send(string $data): void $this->logger->debug('Received session ID', ['session_id' => $this->sessionId]); } + // A notification carries no id, so nobody is waiting for this body: it is + // discarded rather than read as the answer to some other request. + if (self::isNotification($data)) { + $response->getBody()->close(); + + return; + } + $contentType = strtolower($response->getHeaderLine('Content-Type')); if (str_contains($contentType, 'text/event-stream')) { @@ -186,6 +200,17 @@ public function send(string $data): void } } + /** + * Whether the outgoing message is a notification: a `method` and no `id`, so + * it asks the server for nothing and no answer of its own can exist. + */ + private static function isNotification(string $data): bool + { + $payload = json_decode($data, true); + + return \is_array($payload) && \array_key_exists('method', $payload) && !\array_key_exists('id', $payload); + } + /** * @param McpFiber $fiber * @param (callable(float $progress, ?float $total, ?string $message): void)|null $onProgress @@ -194,17 +219,21 @@ public function runRequest(\Fiber $fiber, ?callable $onProgress = null): Respons { $this->activeFiber = $fiber; $this->activeProgressCallback = $onProgress; - $fiber->start(); + try { + $this->activeSuspend = $fiber->start(); + while (!$fiber->isTerminated()) { + $this->tick(); + } - while (!$fiber->isTerminated()) { - $this->tick(); + return $fiber->getReturn(); + } finally { + $this->activeFiber = null; + $this->activeSuspend = null; + $this->activeProgressCallback = null; + $this->activeStream?->close(); + $this->activeStream = null; + $this->sseBuffer = ''; } - - $this->activeFiber = null; - $this->activeProgressCallback = null; - $this->activeStream = null; - - return $fiber->getReturn(); } public function close(): void @@ -253,13 +282,35 @@ private function protocolHeaders(string $payload): array private function tick(): void { + $this->checkInterruption(); $this->processSSEStream(); $this->processProgress(); + $this->checkInterruption(); $this->processFiber(); usleep(1000); // 1ms } + private function checkInterruption(): void + { + if (null === $this->activeSuspend || !$this->activeFiber?->isSuspended()) { + return; + } + + $error = null; + if (($this->activeSuspend['cancellation'] ?? null)?->isCancellationRequested()) { + $error = new RequestCancelledException('The client cancelled the request.'); + } elseif (null !== ($deadline = $this->activeSuspend['deadline'] ?? null) && microtime(true) >= $deadline) { + $error = new TimeoutException('The request deadline expired.'); + } + + if (null !== $error) { + // A blocking read may have delivered a response after cancellation. + $this->state?->consumeResponse($this->activeSuspend['request_id']); + $this->activeFiber->throw($error); + } + } + /** * Read SSE data incrementally from active stream. */ @@ -422,15 +473,16 @@ private function processFiber(): void if (null !== $response) { $this->logger->debug('Resuming fiber with response', ['request_id' => $requestId]); - $this->activeFiber->resume($response); + $this->activeSuspend = $this->activeFiber->resume($response); return; } - if (time() - $timestamp >= $timeout) { + // The explicit per-call deadline replaces the default request timeout. + if (null === ($this->activeSuspend['deadline'] ?? null) && time() - $timestamp >= $timeout) { $this->logger->warning('Request timed out', ['request_id' => $requestId]); $error = Error::forInternalError('Request timed out', $requestId); - $this->activeFiber->resume($error); + $this->activeSuspend = $this->activeFiber->resume($error); return; } diff --git a/src/Client/Transport/StdioTransport.php b/src/Client/Transport/StdioTransport.php index f1029619..314da718 100644 --- a/src/Client/Transport/StdioTransport.php +++ b/src/Client/Transport/StdioTransport.php @@ -13,6 +13,8 @@ use Mcp\Exception\ConnectionException; use Mcp\Exception\InvalidArgumentException; +use Mcp\Exception\RequestCancelledException; +use Mcp\Exception\TimeoutException; use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\JsonRpc\Response; use Psr\Log\LoggerInterface; @@ -27,6 +29,7 @@ * - Managing Fibers waiting for responses * * @phpstan-import-type McpFiber from TransportInterface + * @phpstan-import-type FiberSuspend from TransportInterface * * @author Kyrian Obikwelu */ @@ -54,6 +57,9 @@ class StdioTransport extends BaseTransport /** @var McpFiber|null */ private ?\Fiber $activeFiber = null; + /** @var FiberSuspend|null */ + private ?array $activeSuspend = null; + /** @var (callable(float, ?float, ?string): void)|null */ private $activeProgressCallback; @@ -127,16 +133,19 @@ public function runRequest(\Fiber $fiber, ?callable $onProgress = null): Respons { $this->activeFiber = $fiber; $this->activeProgressCallback = $onProgress; - $fiber->start(); - - while (!$fiber->isTerminated()) { - $this->tick(); - } + try { + $this->activeSuspend = $fiber->start(); - $this->activeFiber = null; - $this->activeProgressCallback = null; + while (!$fiber->isTerminated()) { + $this->tick(); + } - return $fiber->getReturn(); + return $fiber->getReturn(); + } finally { + $this->activeFiber = null; + $this->activeProgressCallback = null; + $this->activeSuspend = null; + } } public function close(): void @@ -309,16 +318,33 @@ private function processFiber(): void if (null !== $response) { $this->logger->debug('Resuming fiber with response', ['request_id' => $requestId]); - $this->activeFiber->resume($response); + $this->activeSuspend = $this->activeFiber->resume($response); return; } + if (($this->activeSuspend['request_id'] ?? null) === $requestId) { + if (($this->activeSuspend['cancellation'] ?? null)?->isCancellationRequested()) { + $this->activeFiber->throw(new RequestCancelledException('The client cancelled the request.')); + + return; + } + + $deadline = $this->activeSuspend['deadline'] ?? null; + if (null !== $deadline) { + if (microtime(true) >= $deadline) { + $this->activeFiber->throw(new TimeoutException('The request deadline expired.')); + } + + return; + } + } + // Check timeout if (time() - $timestamp >= $timeout) { $this->logger->warning('Request timed out', ['request_id' => $requestId]); $error = Error::forInternalError('Request timed out', $requestId); - $this->activeFiber->resume($error); + $this->activeSuspend = $this->activeFiber->resume($error); return; } diff --git a/src/Client/Transport/TransportInterface.php b/src/Client/Transport/TransportInterface.php index 4860128e..b554c8f4 100644 --- a/src/Client/Transport/TransportInterface.php +++ b/src/Client/Transport/TransportInterface.php @@ -23,7 +23,7 @@ * * @phpstan-type FiberReturn (Response|Error) * @phpstan-type FiberResume (Response|Error) - * @phpstan-type FiberSuspend array{type: 'await_response', request_id: int, timeout: int} + * @phpstan-type FiberSuspend array{type: 'await_response', request_id: int, timeout: int, cancellation?: \Mcp\Client\CancellationTokenInterface|null, deadline?: float|null} * @phpstan-type McpFiber \Fiber * * @author Kyrian Obikwelu diff --git a/src/Exception/RequestCancelledException.php b/src/Exception/RequestCancelledException.php new file mode 100644 index 00000000..f1dd6155 --- /dev/null +++ b/src/Exception/RequestCancelledException.php @@ -0,0 +1,19 @@ +connect('cancellation', env: ['MCP_FIXTURE_LOG' => $log]); + $token = new class implements CancellationTokenInterface { + public bool $cancelled = false; + + public function isCancellationRequested(): bool + { + return $this->cancelled; + } + }; + + try { + $client->callTool('slow', onProgress: static function () use ($token): void { + $token->cancelled = true; + }, cancellation: $token); + $this->fail('The pending request should have been cancelled.'); + } catch (RequestCancelledException) { + } + + // The next response is a barrier: the server processes the cancellation + // and its late reply before it can answer this request. + $this->assertSame('quick', $client->callTool('fast')->content[0]->text ?? null); + $events = $this->events($log); + $this->assertSame(['call', 'cancelled', 'call'], array_column($events, 'event')); + $this->assertSame($events[0]['id'], $events[1]['id']); + } finally { + unlink($log); + } + } + + #[TestDox('a per-call deadline cancels an unanswered request without breaking the connection')] + public function testDeadline(): void + { + $log = tempnam(sys_get_temp_dir(), 'mcp-cancel-'); + try { + $client = $this->connect('cancellation', env: ['MCP_FIXTURE_LOG' => $log]); + try { + $client->callTool('slow', timeoutSeconds: 0.05); + $this->fail('The pending request should have timed out.'); + } catch (TimeoutException) { + } + + $this->assertSame('quick', $client->callTool('fast')->content[0]->text ?? null); + $events = $this->events($log); + $this->assertSame(['call', 'cancelled', 'call'], array_column($events, 'event')); + $this->assertSame($events[0]['id'], $events[1]['id']); + } finally { + unlink($log); + } + } + + #[TestDox('a pre-cancelled call sends no request')] + public function testPreCancelledCall(): void + { + $log = tempnam(sys_get_temp_dir(), 'mcp-cancel-'); + try { + $client = $this->connect('cancellation', env: ['MCP_FIXTURE_LOG' => $log]); + $token = new class implements CancellationTokenInterface { + public function isCancellationRequested(): bool + { + return true; + } + }; + + try { + $client->callTool('slow', cancellation: $token); + $this->fail('A pre-cancelled call must not be sent.'); + } catch (RequestCancelledException) { + } + + $this->assertSame('quick', $client->callTool('fast')->content[0]->text ?? null); + $events = $this->events($log); + $this->assertSame(['call'], array_column($events, 'event')); + $this->assertSame('fast', $events[0]['name']); + } finally { + unlink($log); + } + } + + /** + * @return list> + */ + private function events(string $log): array + { + return array_map(static fn (string $line): array => json_decode($line, true, flags: \JSON_THROW_ON_ERROR), file($log, \FILE_IGNORE_NEW_LINES | \FILE_SKIP_EMPTY_LINES)); + } +} diff --git a/tests/Integration/Fixture/cancellation.php b/tests/Integration/Fixture/cancellation.php new file mode 100644 index 00000000..249ac40b --- /dev/null +++ b/tests/Integration/Fixture/cancellation.php @@ -0,0 +1,55 @@ + '2.0', 'id' => $id, + 'result' => [ + 'protocolVersion' => '2025-11-25', + 'capabilities' => (object) [], + 'serverInfo' => ['name' => 'cancellation-fixture', 'version' => '1.0.0'], + ], + ], \JSON_THROW_ON_ERROR)."\n"; + } elseif ('tools/call' === $method) { + $name = $message['params']['name']; + file_put_contents($log, json_encode(['event' => 'call', 'name' => $name, 'id' => $id], \JSON_THROW_ON_ERROR)."\n", \FILE_APPEND); + + if ('slow' === $name) { + if (isset($message['params']['_meta']['progressToken'])) { + echo json_encode([ + 'jsonrpc' => '2.0', 'method' => 'notifications/progress', + 'params' => ['progressToken' => $message['params']['_meta']['progressToken'], 'progress' => 1], + ], \JSON_THROW_ON_ERROR)."\n"; + } + } else { + echo json_encode([ + 'jsonrpc' => '2.0', 'id' => $id, + 'result' => ['content' => [['type' => 'text', 'text' => 'quick']]], + ], \JSON_THROW_ON_ERROR)."\n"; + } + } elseif ('notifications/cancelled' === $method) { + $abandonedId = $message['params']['requestId']; + file_put_contents($log, json_encode(['event' => 'cancelled', 'id' => $abandonedId], \JSON_THROW_ON_ERROR)."\n", \FILE_APPEND); + echo json_encode([ + 'jsonrpc' => '2.0', 'id' => $abandonedId, + 'result' => ['content' => [['type' => 'text', 'text' => 'late']]], + ], \JSON_THROW_ON_ERROR)."\n"; + } +} diff --git a/tests/Unit/Client/ProtocolTest.php b/tests/Unit/Client/ProtocolTest.php index ee421258..d2611c02 100644 --- a/tests/Unit/Client/ProtocolTest.php +++ b/tests/Unit/Client/ProtocolTest.php @@ -11,19 +11,24 @@ namespace Mcp\Tests\Unit\Client; +use Mcp\Client\CancellationTokenInterface; use Mcp\Client\Configuration; use Mcp\Client\Protocol; use Mcp\Client\State\ClientStateInterface; use Mcp\Client\Transport\TransportInterface; use Mcp\Exception\ConnectionException; use Mcp\Exception\LogicException; +use Mcp\Exception\RequestCancelledException; +use Mcp\Exception\TimeoutException; use Mcp\Schema\ClientCapabilities; use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\Implementation; use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\JsonRpc\MessageInterface; use Mcp\Schema\JsonRpc\Response; +use Mcp\Schema\Request\CallToolRequest; use Mcp\Schema\Request\PingRequest; +use Mcp\Schema\Result\CallToolResult; use Mcp\Server\Stateless\RequestMeta; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\TestDox; @@ -163,6 +168,7 @@ public function testIgnoresIdLessErrorResponse(): void public function testErrorResponseWithIdIsStoredForItsPendingRequest(): void { $protocol = new Protocol(); + $protocol->getState()->addPendingRequest(7, 30); $protocol->processMessage('{"jsonrpc": "2.0", "id": 7, "error": {"code": -32601, "message": "Method not found"}}'); @@ -173,6 +179,18 @@ public function testErrorResponseWithIdIsStoredForItsPendingRequest(): void $this->assertSame(Error::METHOD_NOT_FOUND, $response->code); } + #[TestDox('a response for a cancelled request cannot accumulate in state')] + public function testIgnoresResponseForNoLongerPendingRequest(): void + { + $protocol = new Protocol(); + $protocol->getState()->addPendingRequest(7, 30); + $protocol->getState()->removePendingRequest(7); + + $protocol->processMessage('{"jsonrpc": "2.0", "id": 7, "result": {}}'); + + $this->assertNull($protocol->getState()->consumeResponse(7)); + } + #[TestDox('reconnecting starts with a fresh tool catalog, not the previous server\'s verdicts')] public function testReconnectResetsToolCatalog(): void { @@ -208,6 +226,171 @@ public function testEmptyInputResponsesEncodesAsJsonObject(): void $this->assertStringNotContainsString('"inputResponses":[]', $transport->retryBody); } + /** @return iterable */ + public static function suspendedInterruptionProvider(): iterable + { + yield 'cancelled token' => [false]; + yield 'expired deadline' => [true]; + } + + #[DataProvider('suspendedInterruptionProvider')] + public function testInterruptedSuspendedRequestRejectsBufferedResponse(bool $expireDeadline): void + { + $messages = []; + $transport = $this->createMock(TransportInterface::class); + $transport->method('send')->willReturnCallback(static function (string $data) use (&$messages): void { + $messages[] = json_decode($data, true, flags: \JSON_THROW_ON_ERROR); + }); + $protocol = new Protocol(); + $protocol->connect($transport, $this->createConfiguration(ProtocolVersion::V2025_11_25)); + $token = new TestCancellationToken(); + $fiber = new \Fiber(static fn () => $protocol->request(new CallToolRequest('slow', []), 5, false, $token, $expireDeadline ? 0.02 : null)); + $suspend = $fiber->start(); + $this->assertTrue($fiber->isSuspended()); + + if ($expireDeadline) { + // Wait for the actual suspended deadline, not an assumed timing window. + for ($attempt = 0; $attempt < 1000 && microtime(true) < $suspend['deadline']; ++$attempt) { + usleep(100); + } + $this->assertGreaterThanOrEqual($suspend['deadline'], microtime(true)); + } else { + $token->cancelled = true; + } + + $state = $protocol->getState(); + $requestId = $suspend['request_id']; + $state->storeResponse($requestId, ['jsonrpc' => '2.0', 'id' => $requestId, 'result' => []]); + // Match STDIO's response-first ordering: consume the reply, then resume. + try { + $fiber->resume($state->consumeResponse($requestId)); + $this->fail('An interrupted suspended request must reject its buffered reply.'); + } catch (RequestCancelledException|TimeoutException $e) { + $this->assertInstanceOf($expireDeadline ? TimeoutException::class : RequestCancelledException::class, $e); + } + + $this->assertSame([], $state->getPendingRequests()); + $this->assertNull($state->consumeResponse($requestId)); + $this->assertSame('notifications/cancelled', $messages[1]['method']); + $this->assertSame($requestId, $messages[1]['params']['requestId']); + + $next = new \Fiber(static fn () => $protocol->request(new CallToolRequest('fast', []), 5)); + $nextSuspend = $next->start(); + $reply = new Response($nextSuspend['request_id'], ['content' => []]); + $next->resume($reply); + $this->assertSame($reply, $next->getReturn()); + $this->assertSame([], $state->getPendingRequests()); + } + + #[TestDox('a token that flips while the answer is on the wire cancels the request')] + public function testCancellationDuringTheSendIsReported(): void + { + $token = new TestCancellationToken(); + $transport = new InterruptingTransport(static function (array $message) use ($token): void { + if ('slow' === ($message['params']['name'] ?? null)) { + $token->cancelled = true; + } + }); + + $protocol = new Protocol(); + $protocol->connect($transport, $this->createConfiguration(ProtocolVersion::V2025_11_25)); + + try { + $protocol->request(new CallToolRequest('slow', []), 5, false, $token); + $this->fail('A cancelled request must not return the answer that arrived for it.'); + } catch (RequestCancelledException $e) { + $this->assertSame('The client cancelled the request.', $e->getMessage()); + } + + // The buffered answer and the pending entry go together: neither may + // survive to confuse the next request. + $this->assertNull($protocol->getState()->consumeResponse($transport->requestId('slow'))); + $this->assertSame([], $protocol->getState()->getPendingRequests()); + $this->assertSame('fast', $this->toolText($protocol->request(new CallToolRequest('fast', []), 5))); + } + + #[TestDox('a deadline that passes while the answer is on the wire cancels the request')] + public function testDeadlineDuringTheSendIsReported(): void + { + $transport = new InterruptingTransport(function (array $message): void { + if ('slow' === ($message['params']['name'] ?? null)) { + $this->waitPastDeadline(); + } + }); + + $protocol = new Protocol(); + $protocol->connect($transport, $this->createConfiguration(ProtocolVersion::V2025_11_25)); + + try { + $protocol->request(new CallToolRequest('slow', []), 5, false, null, self::DEADLINE_SECONDS); + $this->fail('A request past its deadline must not return the answer that arrived for it.'); + } catch (TimeoutException $e) { + $this->assertSame('The request deadline expired.', $e->getMessage()); + } + + $this->assertNull($protocol->getState()->consumeResponse($transport->requestId('slow'))); + $this->assertSame([], $protocol->getState()->getPendingRequests()); + $this->assertSame('fast', $this->toolText($protocol->request(new CallToolRequest('fast', []), 5))); + } + + #[TestDox('a cancellation notification that cannot be sent is logged, and the interruption still propagates')] + public function testNotificationFailureKeepsTheInterruption(): void + { + $token = new TestCancellationToken(); + $transport = new InterruptingTransport(static function (array $message) use ($token): void { + if ('slow' === ($message['params']['name'] ?? null)) { + $token->cancelled = true; + } + }); + $transport->failsNotifications = true; + + $protocol = new Protocol(logger: $logger = new CollectingLogger()); + $protocol->connect($transport, $this->createConfiguration(ProtocolVersion::V2025_11_25)); + + try { + $protocol->request(new CallToolRequest('slow', []), 5, false, $token); + $this->fail('A cancelled request must not return the answer that arrived for it.'); + } catch (RequestCancelledException $e) { + $this->assertSame('The client cancelled the request.', $e->getMessage()); + } + + $this->assertCount(1, $logger->warnings); + $this->assertInstanceOf(ConnectionException::class, $logger->warnings[0]['exception'] ?? null); + $this->assertSame('fast', $this->toolText($protocol->request(new CallToolRequest('fast', []), 5))); + } + + /** + * @param Response>|Error $response + */ + private function toolText(Response|Error $response): mixed + { + $this->assertInstanceOf(Response::class, $response); + + return CallToolResult::fromArray($response->result)->content[0]->text ?? null; + } + + /** + * A per-call timeout small enough that waiting it out costs microseconds. + */ + private const DEADLINE_SECONDS = 0.001; + + /** + * Wait until the per-call deadline has certainly passed. The deadline is + * stamped as `microtime(true) + DEADLINE_SECONDS` before the send, so the + * clock reaching twice that is a condition the test checks rather than + * assumes. Bounded, so a clock that cannot advance fails instead of hanging. + */ + private function waitPastDeadline(): void + { + $boundary = microtime(true) + 2 * self::DEADLINE_SECONDS; + + for ($attempt = 0; $attempt < 1000 && microtime(true) < $boundary; ++$attempt) { + usleep(100); + } + + $this->assertGreaterThanOrEqual($boundary, microtime(true), 'The clock must pass the per-call deadline for this assertion to be about the deadline.'); + } + private function createConfiguration(ProtocolVersion $protocolVersion): Configuration { return new Configuration( @@ -435,3 +618,120 @@ public function log($level, $message, array $context = []): void } } } + +/** + * Token the tests flip from inside a transport, standing in for whatever asked + * for a request to stop while it was already on the wire. + */ +final class TestCancellationToken implements CancellationTokenInterface +{ + public bool $cancelled = false; + + public function isCancellationRequested(): bool + { + return $this->cancelled; + } +} + +/** + * Transport that runs a hook while a request is on the wire and then answers it + * inline — the shape of a send whose reply is already buffered by the time the + * caller can observe a token flip or a spent deadline. + */ +final class InterruptingTransport implements TransportInterface +{ + /** @var list> */ + public array $messages = []; + + /** Fail notification delivery, as a transport that has died would. */ + public bool $failsNotifications = false; + + /** @var \Closure(array): void */ + private \Closure $duringRequest; + + private ClientStateInterface $state; + + /** + * @param (\Closure(array): void)|null $duringRequest + */ + public function __construct(?\Closure $duringRequest = null) + { + $this->duringRequest = $duringRequest ?? static function (): void { + }; + } + + public function setState(ClientStateInterface $state): void + { + $this->state = $state; + } + + public function send(string $data): void + { + /** @var array $message */ + $message = json_decode($data, true, flags: \JSON_THROW_ON_ERROR); + $this->messages[] = $message; + + if (!\array_key_exists('id', $message)) { + if ($this->failsNotifications) { + throw new ConnectionException('The transport is gone.'); + } + + return; + } + + ($this->duringRequest)($message); + + $this->state->storeResponse($message['id'], [ + 'jsonrpc' => MessageInterface::JSONRPC_VERSION, + 'id' => $message['id'], + 'result' => ['content' => [['type' => 'text', 'text' => $message['params']['name'] ?? '']]], + ]); + } + + /** + * The request id recorded for this tool name. + */ + public function requestId(string $name): int|string + { + foreach ($this->messages as $message) { + if ($name === ($message['params']['name'] ?? null)) { + $id = $message['id'] ?? null; + + if (\is_int($id) || \is_string($id)) { + return $id; + } + } + } + + throw new \RuntimeException(\sprintf('No request recorded for tool "%s".', $name)); + } + + public function connect(): void + { + } + + public function close(): void + { + } + + public function runRequest(\Fiber $fiber, ?callable $onProgress = null): Response|Error + { + throw new LogicException('Not used in these tests.'); + } + + public function onInitialize(callable $callback): void + { + } + + public function onMessage(callable $callback): void + { + } + + public function onError(callable $callback): void + { + } + + public function onClose(callable $callback): void + { + } +} diff --git a/tests/Unit/Client/Transport/HttpTransportTest.php b/tests/Unit/Client/Transport/HttpTransportTest.php index 83a77e0b..c4f354b0 100644 --- a/tests/Unit/Client/Transport/HttpTransportTest.php +++ b/tests/Unit/Client/Transport/HttpTransportTest.php @@ -12,9 +12,13 @@ namespace Mcp\Tests\Unit\Client\Transport; use Mcp\Client; +use Mcp\Client\CancellationTokenInterface; use Mcp\Client\State\ClientState; use Mcp\Client\Transport\HttpTransport; use Mcp\Exception\InvalidArgumentException; +use Mcp\Exception\RequestCancelledException; +use Mcp\Exception\TimeoutException; +use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\JsonRpc\Error; use Nyholm\Psr7\Factory\Psr17Factory; use Nyholm\Psr7\Response; @@ -200,6 +204,245 @@ public function testRejectsNonPositiveCap(): void $this->createTransport(maxSseBufferBytes: 0); } + /** @return iterable */ + public static function cancellationPointProvider(): iterable + { + yield 'before first SSE read' => [true]; + yield 'between SSE reads' => [false]; + } + + #[DataProvider('cancellationPointProvider')] + public function testCancellationClosesSseBodyAndAllowsNextCall(bool $beforeRead): void + { + $token = new class implements CancellationTokenInterface { + public bool $cancelled = false; + + public function isCancellationRequested(): bool + { + return $this->cancelled; + } + }; + $body = $this->createMock(StreamInterface::class); + $body->method('eof')->willReturn(false); + $body->expects($beforeRead ? $this->never() : $this->once())->method('read')->willReturnCallback(static function () use ($token): string { + $token->cancelled = true; + + return ": keepalive\n\n"; + }); + $body->expects($this->once())->method('close'); + + $httpClient = $this->createMock(ClientInterface::class); + $httpClient->method('sendRequest')->willReturnCallback(static function (RequestInterface $request) use ($token, $body, $beforeRead): ResponseInterface { + $payload = json_decode((string) $request->getBody(), true, flags: \JSON_THROW_ON_ERROR); + if ('initialize' === $payload['method']) { + $result = [ + 'protocolVersion' => '2025-11-25', + 'capabilities' => ['tools' => []], + 'serverInfo' => ['name' => 'test', 'version' => '1'], + ]; + } elseif ('tools/call' === $payload['method']) { + if ('slow' === $payload['params']['name']) { + $token->cancelled = $beforeRead; + + return new Response(200, ['Content-Type' => 'text/event-stream'], $body); + } + $result = ['content' => [['type' => 'text', 'text' => 'next call']]]; + } else { + return new Response(202); + } + + return new Response(200, ['Content-Type' => 'application/json'], json_encode([ + 'jsonrpc' => '2.0', 'id' => $payload['id'], 'result' => $result, + ], \JSON_THROW_ON_ERROR)); + }); + $client = Client::builder()->setClientInfo('test', '1')->build(); + $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory)); + + try { + $client->callTool('slow', cancellation: $token); + $this->fail('Expected cancellation.'); + } catch (RequestCancelledException) { + $this->assertTrue($client->isConnected()); + } + $this->assertSame('next call', $client->callTool('fast')->content[0]->text ?? null); + $client->disconnect(); + } + + public function testExpiredDeadlineClosesBodyBeforeReading(): void + { + $transport = $this->createTransport(); + $state = new ClientState(); + $transport->setState($state); + $state->addPendingRequest(1, 30); + $body = $this->createMock(StreamInterface::class); + $body->expects($this->never())->method('read'); + $body->expects($this->once())->method('close'); + $this->setActiveStream($transport, $body); + + $fiber = new \Fiber(static fn () => \Fiber::suspend([ + 'type' => 'await_response', + 'request_id' => 1, + 'timeout' => 30, + 'deadline' => microtime(true) - 1, + ])); + + $this->expectException(TimeoutException::class); + $transport->runRequest($fiber); + } + + /** @return iterable */ + public static function notificationAnswerProvider(): iterable + { + yield 'JSON answer' => ['application/json']; + yield 'SSE answer' => ['text/event-stream']; + } + + #[DataProvider('notificationAnswerProvider')] + #[TestDox('a notification answer is discarded rather than parked in the stream a request is waiting on: $_dataName')] + public function testNotificationAnswerIsDiscarded(string $contentType): void + { + $original = $this->createMock(StreamInterface::class); + $original->expects($this->never())->method('close'); + + $answer = $this->createMock(StreamInterface::class); + $answer->expects($this->once())->method('close'); + $answer->expects($this->never())->method('getContents'); + + $httpClient = new class($answer, $contentType) implements ClientInterface { + public function __construct( + private readonly StreamInterface $answer, + private readonly string $contentType, + ) { + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + return new Response(200, ['Content-Type' => $this->contentType], $this->answer); + } + }; + + $transport = new HttpTransport('https://example.test/mcp', [], $httpClient, $this->factory, $this->factory); + $dispatched = []; + $transport->onMessage(static function (string $message) use (&$dispatched): void { + $dispatched[] = $message; + }); + $this->setActiveStream($transport, $original); + + $transport->send(json_encode([ + 'jsonrpc' => '2.0', + 'method' => 'notifications/cancelled', + 'params' => ['requestId' => 1], + ], \JSON_THROW_ON_ERROR)); + + $this->assertSame($original, $this->readPrivate($transport, 'activeStream'), 'the interrupted request keeps its own stream'); + $this->assertSame([], $dispatched, 'nothing may be dispatched as an answer to a notification'); + } + + #[TestDox('a token that flips while the JSON body is read cancels the call, and a handshake-era server is told')] + public function testCancellationDuringJsonBodyReadIsReported(): void + { + $token = new class implements CancellationTokenInterface { + public bool $cancelled = false; + + public function isCancellationRequested(): bool + { + return $this->cancelled; + } + }; + $body = new class { + public string $contents = ''; + }; + $stream = $this->createMock(StreamInterface::class); + $stream->method('getContents')->willReturnCallback(static function () use ($body, $token): string { + $token->cancelled = true; + + return $body->contents; + }); + + $httpClient = new RecordingHttpClient(static function (array $message) use ($body, $stream): ?ResponseInterface { + if ('slow' !== ($message['params']['name'] ?? null)) { + return null; + } + + // The server answered in full; the caller stopped caring mid-read. + $body->contents = json_encode([ + 'jsonrpc' => '2.0', + 'id' => $message['id'], + 'result' => ['content' => [['type' => 'text', 'text' => 'abandoned']]], + ], \JSON_THROW_ON_ERROR); + + return new Response(200, ['Content-Type' => 'application/json'], $stream); + }); + + $client = Client::builder()->setClientInfo('test', '1')->build(); + $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory)); + + try { + $client->callTool('slow', cancellation: $token); + $this->fail('Expected cancellation.'); + } catch (RequestCancelledException $e) { + $this->assertSame('The client cancelled the request.', $e->getMessage()); + } + + $abandoned = $httpClient->callId('slow'); + $cancellations = $httpClient->messagesOfMethod('notifications/cancelled'); + $this->assertCount(1, $cancellations); + $this->assertSame($abandoned, $cancellations[0]['params']['requestId'] ?? null); + $this->assertSame('next call', $client->callTool('fast')->content[0]->text ?? null); + $client->disconnect(); + } + + /** @return iterable */ + public static function revisionProvider(): iterable + { + yield 'handshake-era HTTP reports the cancellation' => [ProtocolVersion::V2025_11_25, true]; + yield 'modern HTTP leaves it to the closed response stream' => [ProtocolVersion::V2026_07_28, false]; + } + + #[DataProvider('revisionProvider')] + #[TestDox('cancellation signalling: $_dataName')] + public function testCancellationSignallingDependsOnTheRevision(ProtocolVersion $version, bool $expectsNotification): void + { + $token = new class implements CancellationTokenInterface { + public bool $cancelled = false; + + public function isCancellationRequested(): bool + { + return $this->cancelled; + } + }; + $httpClient = new RecordingHttpClient(static function (array $message) use ($token): ?ResponseInterface { + if ('slow' === ($message['params']['name'] ?? null)) { + $token->cancelled = true; + } + + return null; + }); + + $client = Client::builder()->setClientInfo('test', '1')->setProtocolVersion($version)->build(); + $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory)); + + try { + $client->callTool('slow', cancellation: $token); + $this->fail('Expected cancellation.'); + } catch (RequestCancelledException $e) { + $this->assertSame('The client cancelled the request.', $e->getMessage()); + } + + $cancellations = $httpClient->messagesOfMethod('notifications/cancelled'); + + if (!$expectsNotification) { + $this->assertSame([], $cancellations); + } else { + $this->assertCount(1, $cancellations); + $this->assertArrayNotHasKey('id', $cancellations[0], 'the cancellation goes out as a notification'); + $this->assertSame($httpClient->callId('slow'), $cancellations[0]['params']['requestId'] ?? null); + } + + $this->assertSame('next call', $client->callTool('fast')->content[0]->text ?? null); + $client->disconnect(); + } + private function createTransport(int $maxSseBufferBytes = 8 * 1024 * 1024): HttpTransport { return new HttpTransport( @@ -226,3 +469,96 @@ private function readPrivate(HttpTransport $transport, string $property): mixed return (new \ReflectionProperty($transport, $property))->getValue($transport); } } + +/** + * PSR-18 client that answers both wire formats this SDK speaks and records every + * message it was handed, so a test can inspect what went out. + */ +final class RecordingHttpClient implements ClientInterface +{ + /** @var list> */ + public array $messages = []; + + /** + * The hook sees every request first and may return its own answer; a null + * return falls through to the default one. + * + * @param (\Closure(array): ?ResponseInterface)|null $intercept + */ + public function __construct(private readonly ?\Closure $intercept = null) + { + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + /** @var array $message */ + $message = json_decode((string) $request->getBody(), true, flags: \JSON_THROW_ON_ERROR); + $this->messages[] = $message; + + if (null !== $this->intercept) { + $answer = ($this->intercept)($message); + if (null !== $answer) { + return $answer; + } + } + + return match ($message['method'] ?? null) { + 'initialize' => $this->json($message, [ + 'protocolVersion' => ProtocolVersion::V2025_11_25->value, + 'capabilities' => ['tools' => []], + 'serverInfo' => ['name' => 'test-server', 'version' => '1.0.0'], + ]), + 'server/discover' => $this->json($message, [ + 'resultType' => 'complete', + 'supportedVersions' => [ProtocolVersion::V2026_07_28->value], + 'capabilities' => ['tools' => []], + 'serverInfo' => ['name' => 'test-server', 'version' => '1.0.0'], + ]), + 'tools/call' => $this->json($message, [ + 'content' => [['type' => 'text', 'text' => 'fast' === ($message['params']['name'] ?? null) ? 'next call' : 'abandoned']], + ]), + default => new Response(202), + }; + } + + /** + * Every recorded message carrying this method. + * + * @return list> + */ + public function messagesOfMethod(string $method): array + { + return array_values(array_filter($this->messages, static fn (array $message): bool => $method === ($message['method'] ?? null))); + } + + /** + * The request id recorded for this tool name. + */ + public function callId(string $name): int|string + { + foreach ($this->messages as $message) { + if ('tools/call' === ($message['method'] ?? null) && $name === ($message['params']['name'] ?? null)) { + $id = $message['id'] ?? null; + + if (\is_int($id) || \is_string($id)) { + return $id; + } + } + } + + throw new \RuntimeException(\sprintf('No recorded "tools/call" for tool "%s".', $name)); + } + + /** + * @param array $message + * @param array $result + */ + private function json(array $message, array $result): Response + { + return new Response(200, ['Content-Type' => 'application/json'], json_encode([ + 'jsonrpc' => '2.0', + 'id' => $message['id'], + 'result' => $result, + ], \JSON_THROW_ON_ERROR)); + } +}