From c6f3b0a76f40194cfc4daba400eae57900b71923 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Sat, 10 Oct 2026 03:58:31 +0200 Subject: [PATCH] [Server] Resume every suspended fiber on stdio Concurrent tool calls that report progress or elicit replaced each other's fiber, so only the last one finished. Protocol tracks the awaited request per fiber, and the transport callbacks get the fiber. --- CHANGELOG.md | 1 + src/Server/Protocol.php | 35 ++-- src/Server/Transport/BaseTransport.php | 11 +- .../Transport/ManagesTransportCallbacks.php | 7 +- src/Server/Transport/StdioTransport.php | 65 +++++--- .../Transport/StreamableHttpTransport.php | 8 +- src/Server/Transport/TransportInterface.php | 6 +- tests/Unit/Fixtures/PollingLoopTransport.php | 8 +- .../Transport/StdioSuspendedFibersTest.php | 152 ++++++++++++++++++ 9 files changed, 234 insertions(+), 59 deletions(-) create mode 100644 tests/Unit/Server/Transport/StdioSuspendedFibersTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index e34852a9..ba56e532 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * On a `2026-07-28` connection, `Client::setLoggingLevel()` stamps the level on every following request, `Client::ping()` sends `server/discover` and `Client::sendRootsListChanged()` sends nothing. * Fail a client request at once when the HTTP server refuses it with an error status or the stdio server process exits, instead of waiting out the timeout. * [BC Break] Bump `MessageInterface::PROTOCOL_VERSION` to `2026-07-28`. Use `ProtocolVersion::latestHandshake()` where a handshake revision is needed, e.g. in an `initialize` answer. +* [BC Break] Fix concurrent stdio tool calls that report progress, log, elicit or sample never finishing: `StdioTransport` keeps every suspended fiber and resumes each with the answer to its own request. A transport passes the suspended `\Fiber` as last argument to the pending-requests provider and the fiber yield handler. 0.8.0 ----- diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index bbd25019..84538fa6 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -74,10 +74,10 @@ class Protocol private const INTERNAL_ERROR_MESSAGE = 'Internal server error.'; /** - * The client request each transport's fiber is suspended on. Pending requests are stored in the - * session, which concurrent streams share, so a stream must only poll the one its fiber sent. + * The client request each suspended fiber waits on. Pending requests are stored in the session, + * which concurrent calls share, so a fiber must only be resumed with the answer to the one it sent. * - * @var \WeakMap, int> + * @var \WeakMap */ private \WeakMap $awaitedRequestIds; @@ -113,19 +113,12 @@ public function connect(TransportInterface $transport): void $transport->setOutgoingMessagesProvider($this->consumeOutgoingMessages(...)); - // The transport keeps these callbacks, so they reference it weakly to not keep it alive. - $transportRef = \WeakReference::create($transport); - - $transport->setPendingRequestsProvider(fn (Uuid $sessionId): array => $this->getAwaitedPendingRequests($transportRef->get(), $sessionId)); + $transport->setPendingRequestsProvider(fn (Uuid $sessionId, \Fiber $fiber): array => $this->getAwaitedPendingRequests($fiber, $sessionId)); $transport->setResponseFinder($this->checkResponse(...)); - $transport->setFiberYieldHandler(function (mixed $yieldedValue, ?Uuid $sessionId) use ($transportRef): void { - $requestId = $this->handleFiberYield($yieldedValue, $sessionId); - - if (null !== $transport = $transportRef->get()) { - $this->trackAwaitedRequest($transport, $requestId); - } + $transport->setFiberYieldHandler(function (mixed $yieldedValue, ?Uuid $sessionId, \Fiber $fiber): void { + $this->trackAwaitedRequest($fiber, $this->handleFiberYield($yieldedValue, $sessionId)); }); $this->logger->info('Protocol connected to transport', ['transport' => $transport::class]); @@ -374,7 +367,7 @@ private function handleRequest(TransportInterface $transport, Request $request, throw new RuntimeException('Failed to save the session of a suspended request.'); } - $this->trackAwaitedRequest($transport, $awaitedRequestId); + $this->trackAwaitedRequest($fiber, $awaitedRequestId); $transport->attachFiberToSession($fiber, $session->getId()); return; @@ -695,27 +688,27 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int } /** - * @param TransportInterface $transport + * @param McpFiber $fiber */ - private function trackAwaitedRequest(TransportInterface $transport, ?int $requestId): void + private function trackAwaitedRequest(\Fiber $fiber, ?int $requestId): void { if (null === $requestId) { - unset($this->awaitedRequestIds[$transport]); + unset($this->awaitedRequestIds[$fiber]); return; } - $this->awaitedRequestIds[$transport] = $requestId; + $this->awaitedRequestIds[$fiber] = $requestId; } /** - * @param TransportInterface|null $transport + * @param McpFiber $fiber * * @return array */ - private function getAwaitedPendingRequests(?TransportInterface $transport, Uuid $sessionId): array + private function getAwaitedPendingRequests(\Fiber $fiber, Uuid $sessionId): array { - $requestId = null !== $transport ? $this->awaitedRequestIds[$transport] ?? null : null; + $requestId = $this->awaitedRequestIds[$fiber] ?? null; if (null === $requestId) { return []; } diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php index dcf18c36..f80b1046 100644 --- a/src/Server/Transport/BaseTransport.php +++ b/src/Server/Transport/BaseTransport.php @@ -86,12 +86,14 @@ protected function getOutgoingMessages(?Uuid $sessionId): array } /** + * @param McpFiber $fiber + * * @return array> */ - protected function getPendingRequests(?Uuid $sessionId): array + protected function getPendingRequests(?Uuid $sessionId, \Fiber $fiber): array { if ($sessionId && \is_callable($this->pendingRequestsProvider)) { - return ($this->pendingRequestsProvider)($sessionId); + return ($this->pendingRequestsProvider)($sessionId, $fiber); } return []; @@ -111,15 +113,16 @@ protected function checkForResponse(int $requestId, ?Uuid $sessionId): Response| /** * @param FiberSuspend|null $yielded + * @param McpFiber $fiber the fiber that yielded */ - protected function handleFiberYield(mixed $yielded, ?Uuid $sessionId): void + protected function handleFiberYield(mixed $yielded, ?Uuid $sessionId, \Fiber $fiber): void { if (null === $yielded || !\is_callable($this->fiberYieldHandler)) { return; } try { - ($this->fiberYieldHandler)($yielded, $sessionId); + ($this->fiberYieldHandler)($yielded, $sessionId, $fiber); } catch (\Throwable $e) { $this->logger->error('Fiber yield handler failed.', [ 'exception' => $e, diff --git a/src/Server/Transport/ManagesTransportCallbacks.php b/src/Server/Transport/ManagesTransportCallbacks.php index 765e2ef3..b5939fe3 100644 --- a/src/Server/Transport/ManagesTransportCallbacks.php +++ b/src/Server/Transport/ManagesTransportCallbacks.php @@ -22,6 +22,7 @@ * @phpstan-import-type FiberReturn from \Mcp\Server\Transport\TransportInterface * @phpstan-import-type FiberResume from \Mcp\Server\Transport\TransportInterface * @phpstan-import-type FiberSuspend from \Mcp\Server\Transport\TransportInterface + * @phpstan-import-type McpFiber from \Mcp\Server\Transport\TransportInterface * * @author Kyrian Obikwelu * */ @@ -36,13 +37,13 @@ trait ManagesTransportCallbacks /** @var callable(Uuid): array}> */ protected $outgoingMessagesProvider; - /** @var callable(Uuid): array> */ + /** @var callable(Uuid, McpFiber): array> */ protected $pendingRequestsProvider; /** @var (callable(int, Uuid): (Response>|Error|null))|null */ protected $responseFinder; - /** @var callable(FiberSuspend|null, ?Uuid): void */ + /** @var callable(FiberSuspend|null, ?Uuid, McpFiber): void */ protected $fiberYieldHandler; public function onMessage(callable $listener): void @@ -74,7 +75,7 @@ public function setResponseFinder(callable $finder): void } /** - * @param callable(FiberSuspend|null, ?Uuid): void $handler + * @param callable(FiberSuspend|null, ?Uuid, McpFiber): void $handler */ public function setFiberYieldHandler(callable $handler): void { diff --git a/src/Server/Transport/StdioTransport.php b/src/Server/Transport/StdioTransport.php index 4f25d5e9..7d9d142e 100644 --- a/src/Server/Transport/StdioTransport.php +++ b/src/Server/Transport/StdioTransport.php @@ -22,6 +22,7 @@ use Mcp\Server\Transport\Stdio\RunnerState; use Mcp\Server\Wire\InboundClassifier; use Psr\Log\LoggerInterface; +use Symfony\Component\Uid\Uuid; /** * Serves one client over the standard streams, in whichever protocol era it @@ -79,6 +80,14 @@ class StdioTransport extends BaseTransport implements StatelessAwareTransportInt */ private array $streams = []; + /** + * Handshake-era handlers suspended on the client, by object id: the + * client pipelines its requests, so several can be in flight at once. + * + * @var array + */ + private array $fibers = []; + /** * @param resource $input * @param resource $output @@ -124,6 +133,12 @@ public function send(string $data, array $context): void $this->writeLine($data); } + public function attachFiberToSession(\Fiber $fiber, Uuid $sessionId): void + { + $this->fibers[spl_object_id($fiber)] = $fiber; + $this->sessionId = $sessionId; + } + public function listen(): int { $this->logger->info('StdioTransport is listening for messages on STDIN...'); @@ -131,7 +146,7 @@ public function listen(): int while (!feof($this->input) && RunnerState::RUNNING === $this->runnerControl->getState()) { $this->processInput(); - $this->processFiber(); + $this->processFibers(); $this->processStreams(); $this->flushOutgoingMessages(); } @@ -341,27 +356,35 @@ private function writeError(Error $error): void $this->writeLine(json_encode($error, \JSON_THROW_ON_ERROR | \JSON_UNESCAPED_SLASHES)); } - private function processFiber(): void + /** + * Resumes every suspended handler that can go on: one that sent a + * notification right away, one that sent a request once its answer is in + * or it timed out. + */ + private function processFibers(): void { - if (null === $this->sessionFiber) { - return; - } - - if ($this->sessionFiber->isTerminated()) { - $this->handleFiberTermination($this->sessionFiber); - - return; - } + foreach ($this->fibers as $key => $fiber) { + if ($fiber->isSuspended()) { + $this->resumeFiber($fiber); + } - if (!$this->sessionFiber->isSuspended()) { - return; + if ($fiber->isTerminated()) { + unset($this->fibers[$key]); + $this->handleFiberTermination($fiber); + } } + } - $pendingRequests = $this->getPendingRequests($this->sessionId); + /** + * @param McpFiber $fiber + */ + private function resumeFiber(\Fiber $fiber): void + { + $pendingRequests = $this->getPendingRequests($this->sessionId, $fiber); if (empty($pendingRequests)) { - $yielded = $this->sessionFiber->resume(); - $this->handleFiberYield($yielded, $this->sessionId); + $yielded = $fiber->resume(); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); return; } @@ -374,16 +397,16 @@ private function processFiber(): void $response = $this->checkForResponse($requestId, $this->sessionId); if (null !== $response) { - $yielded = $this->sessionFiber->resume($response); - $this->handleFiberYield($yielded, $this->sessionId); + $yielded = $fiber->resume($response); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); return; } if (time() - $timestamp >= $timeout) { $error = Error::forInternalError('Request timed out', $requestId); - $yielded = $this->sessionFiber->resume($error); - $this->handleFiberYield($yielded, $this->sessionId); + $yielded = $fiber->resume($error); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); return; } @@ -405,8 +428,6 @@ private function handleFiberTermination(\Fiber $fiber): void $this->logger->error('STDIO: Failed to encode final Fiber result.', ['exception' => $e]); } } - - $this->sessionFiber = null; } private function flushOutgoingMessages(): void diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php index 847524e1..c52fab50 100644 --- a/src/Server/Transport/StreamableHttpTransport.php +++ b/src/Server/Transport/StreamableHttpTransport.php @@ -299,11 +299,11 @@ protected function createStreamedResponse(): ResponseInterface while ($fiber->isSuspended()) { $this->flushOutgoingMessages($this->sessionId); - $pendingRequests = $this->getPendingRequests($this->sessionId); + $pendingRequests = $this->getPendingRequests($this->sessionId, $fiber); if (empty($pendingRequests)) { $yielded = $fiber->resume(); - $this->handleFiberYield($yielded, $this->sessionId); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); continue; } @@ -317,7 +317,7 @@ protected function createStreamedResponse(): ResponseInterface if (null !== $response) { $yielded = $fiber->resume($response); - $this->handleFiberYield($yielded, $this->sessionId); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); $resumed = true; break; } @@ -325,7 +325,7 @@ protected function createStreamedResponse(): ResponseInterface if ($this->clock->now()->getTimestamp() - $timestamp >= $timeout) { $error = Error::forInternalError('Request timed out', $requestId); $yielded = $fiber->resume($error); - $this->handleFiberYield($yielded, $this->sessionId); + $this->handleFiberYield($yielded, $this->sessionId, $fiber); $resumed = true; break; } diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php index 3066e293..40e37418 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/src/Server/Transport/TransportInterface.php @@ -97,11 +97,11 @@ public function onSessionEnd(callable $listener): void; public function setOutgoingMessagesProvider(callable $provider): void; /** - * Set a provider function to retrieve all pending server-initiated requests. + * Set a provider function to retrieve the pending server-initiated request a suspended Fiber waits on. * * The transport calls this to decide if it should wait for a client response before resuming a Fiber. * - * @param callable(Uuid $sessionId): array> $provider + * @param callable(Uuid $sessionId, McpFiber $fiber): array> $provider */ public function setPendingRequestsProvider(callable $provider): void; @@ -118,7 +118,7 @@ public function setResponseFinder(callable $finder): void; * The transport calls this to let the Protocol handle new requests/notifications * that are yielded from a Fiber's execution. * - * @param callable(FiberSuspend|null, ?Uuid $sessionId): void $handler + * @param callable(FiberSuspend|null, ?Uuid $sessionId, McpFiber $fiber): void $handler */ public function setFiberYieldHandler(callable $handler): void; diff --git a/tests/Unit/Fixtures/PollingLoopTransport.php b/tests/Unit/Fixtures/PollingLoopTransport.php index 34317a18..ae749305 100644 --- a/tests/Unit/Fixtures/PollingLoopTransport.php +++ b/tests/Unit/Fixtures/PollingLoopTransport.php @@ -29,7 +29,9 @@ final class PollingLoopTransport extends InMemoryTransport */ public function getPendingRequestIds(): array { - return array_keys($this->getPendingRequests($this->sessionId)); + \assert(null !== $this->sessionFiber); + + return array_keys($this->getPendingRequests($this->sessionId, $this->sessionFiber)); } /** @@ -37,6 +39,8 @@ public function getPendingRequestIds(): array */ public function yieldFromFiber(NotificationSuspension|RequestSuspension $yielded): void { - $this->handleFiberYield($yielded, $this->sessionId); + \assert(null !== $this->sessionFiber); + + $this->handleFiberYield($yielded, $this->sessionId, $this->sessionFiber); } } diff --git a/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php b/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php new file mode 100644 index 00000000..646c05f4 --- /dev/null +++ b/tests/Unit/Server/Transport/StdioSuspendedFibersTest.php @@ -0,0 +1,152 @@ +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 + { + // The server numbers its requests to the client from 1000 on, in the order the calls ask. + $lines = $this->exchange([ + self::call(2, 'ask'), + self::call(3, 'ask'), + ['jsonrpc' => '2.0', 'id' => 1001, 'result' => ['action' => 'accept', 'content' => ['name' => 'second']]], + ['jsonrpc' => '2.0', 'id' => 1000, 'result' => ['action' => 'accept', 'content' => ['name' => 'first']]], + ]); + + $elicitations = array_values(array_filter($lines, static fn (array $line): bool => 'elicitation/create' === ($line['method'] ?? null))); + $this->assertSame([1000, 1001], array_column($elicitations, 'id')); + + $this->assertSame('first', $this->answer($lines, 2)['result']['content'][0]['text'] ?? null); + $this->assertSame('second', $this->answer($lines, 3)['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 + * + * @return list> + */ + private function exchange(array $messages): array + { + $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') + ->build(); + + $input = fopen('php://temp', 'r+'); + $this->assertNotFalse($input); + // 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))), + ); + } +}