diff --git a/CHANGELOG.md b/CHANGELOG.md index 64fb49a6..43ffa2e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * Fix the client request timeout firing up to a second early and throwing a generic `RequestException`: it now throws `TimeoutException`, which extends `RequestException`, and sends `notifications/cancelled` for the abandoned request. * [BC Break] `TimeoutException` extends `RequestException` instead of `Exception`, so a `catch (RequestException)` now also catches timeouts. `ClientStateInterface::getPendingRequests()` reports `timestamp` as a `float` from `microtime(true)` instead of an `int`. * Fix the client's `HttpTransport` waiting out the request timeout when a non-streamed response to a request ends without its answer (truncated, empty, unreadable or not JSON): the call now fails at once with a `RequestException`, also when the body of an error status fails to read. +* Fix the client's `HttpTransport` failing to connect through an amphp-based PSR-18 client, like Symfony's `Psr18Client(new AmpHttpClient())`: the request never left the process and `connect()` failed at the init timeout with `FutureIterator::consume(): Return value must be of type ?array`. Requests are now sent from outside the request fiber, where the event loop runs. * Fix an unreachable `Psr16SessionStore` cache being answered as "session not found" (`404`): the failure is logged and answered with `503`, so clients retry instead of re-initializing. `SessionStoreInterface::exists()` may throw the new `SessionStoreException`. * Serve both protocol eras over stdio: `StdioTransport` settles the era on the client's first request and serves `2026-07-28` requests, `subscriptions/listen` and `notifications/cancelled` on the one channel. * Fix `ClientGateway::log()` ignoring the client's log level: like `ClientLogger`, it drops messages below the level set with `logging/setLevel`, and below `warning` while the client has set none. diff --git a/src/Client/Transport/HttpTransport.php b/src/Client/Transport/HttpTransport.php index 4435113b..a74b2024 100644 --- a/src/Client/Transport/HttpTransport.php +++ b/src/Client/Transport/HttpTransport.php @@ -37,6 +37,8 @@ * @phpstan-import-type FiberSuspend from TransportInterface * @phpstan-import-type SseEvent from SseParser * + * @phpstan-type SendSuspend array{type: 'send', data: string} + * * @author Kyrian Obikwelu */ class HttpTransport extends BaseTransport implements HeaderAwareTransportInterface @@ -53,7 +55,7 @@ class HttpTransport extends BaseTransport implements HeaderAwareTransportInterfa /** @var McpFiber|null */ private ?\Fiber $activeFiber = null; - /** @var FiberSuspend|null */ + /** @var FiberSuspend|SendSuspend|null */ private ?array $activeSuspend = null; /** @var StreamInterface|null Active SSE stream being read */ @@ -141,7 +143,7 @@ public function connect(): void $fiber = new \Fiber(fn () => $this->handleInitialize()); $this->activeFiber = $fiber; - $fiber->start(); + $this->activeSuspend = $fiber->start(); while (!$fiber->isTerminated()) { $this->tick(); @@ -149,6 +151,7 @@ public function connect(): void $result = $fiber->getReturn(); $this->activeFiber = null; + $this->activeSuspend = null; $this->resetSseState(); if ($result instanceof Error) { @@ -185,6 +188,18 @@ public function getSessionId(): ?string public function send(string $data): void { + // An asynchronous PSR-18 client (amphp, through Symfony's AmpHttpClient + // or its own adapter) suspends the fiber it is called from, for its event + // loop to resume; only from {main} does it run that loop itself. Inside + // the request fiber, the one thing that would resume it is this + // transport, with an answer meant for the protocol. So the request fiber + // steps out and the next tick sends from {main}, body reads included. + if (null !== $this->activeFiber && \Fiber::getCurrent() === $this->activeFiber) { + \Fiber::suspend(['type' => 'send', 'data' => $data]); + + return; + } + $request = $this->requestFactory->createRequest('POST', $this->endpoint) ->withHeader('Content-Type', 'application/json') ->withHeader('Accept', 'application/json, text/event-stream') @@ -275,6 +290,30 @@ public function send(string $data): void } } + /** + * Send what the request fiber stepped out for, from {main}. + * + * A failure is thrown back into the fiber, so it surfaces from send() as it + * would have had the message been sent in place. + */ + private function processSends(): void + { + while (null !== ($fiber = $this->activeFiber) && $fiber->isSuspended() && 'send' === ($this->activeSuspend['type'] ?? null)) { + /** @var SendSuspend $suspend */ + $suspend = $this->activeSuspend; + + try { + $this->send($suspend['data']); + } catch (\Throwable $e) { + $this->activeSuspend = $fiber->throw($e); + + continue; + } + + $this->activeSuspend = $fiber->resume(); + } + } + /** * Fails a request whose response ended without its answer at once, rather than at its timeout. */ @@ -544,6 +583,7 @@ private function nonBlocking(StreamInterface $body, ?bool &$switched = null): St private function tick(): void { + $this->processSends(); $this->checkInterruption(); $this->processSSEStream(); $this->processListenStream(); @@ -570,7 +610,8 @@ private function checkInterruption(): void if (null !== $error) { // A blocking read may have delivered a response after cancellation. $this->state?->consumeResponse($this->activeSuspend['request_id']); - $this->activeFiber->throw($error); + // The fiber may step out again to send the cancellation notification. + $this->activeSuspend = $this->activeFiber->throw($error); } } @@ -1000,6 +1041,11 @@ private function processFiber(): void return; } + // Waiting to have a message sent, which the next tick does. + if ('send' === ($this->activeSuspend['type'] ?? null)) { + return; + } + if (null === $this->state) { return; } diff --git a/tests/Unit/Client/Transport/HttpTransportTest.php b/tests/Unit/Client/Transport/HttpTransportTest.php index 3ed9af57..6f2907db 100644 --- a/tests/Unit/Client/Transport/HttpTransportTest.php +++ b/tests/Unit/Client/Transport/HttpTransportTest.php @@ -1206,6 +1206,39 @@ public function testHandshakeEraRequestsCarryTheProtocolVersion(): void ], $sent); } + #[TestDox('an HTTP client that suspends the calling fiber, as amphp does, is not called inside the request fiber')] + public function testHttpClientIsCalledOutsideTheRequestFiber(): void + { + $httpClient = new class(new RecordingHttpClient()) implements ClientInterface { + public function __construct(private readonly RecordingHttpClient $server) + { + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + FiberSuspendingStream::awaitEventLoop(); + + $response = $this->server->sendRequest($request); + + // The body of an amphp response is read through the event loop too. + return $response->withBody(new FiberSuspendingStream($response->getBody())); + } + }; + + $client = Client::builder() + ->setClientInfo('test-client', '1.0.0') + ->setInitTimeout(1) + ->setMaxRetries(0) + ->build(); + + $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory)); + + $this->assertTrue($client->isConnected()); + $this->assertSame('test-server', $client->getServerInfo()?->name); + $this->assertSame('next call', $client->callTool('fast')->content[0]->text ?? null); + $client->disconnect(); + } + /** * @param array $headers */ @@ -1529,3 +1562,110 @@ private static function events(string $eol): array return $events; } } + +/** + * A response body that, like one read through Revolt's event loop, suspends the + * fiber it is read from and expects only that loop to resume it. + */ +final class FiberSuspendingStream implements StreamInterface +{ + public function __construct(private readonly StreamInterface $inner) + { + } + + /** + * Inside a fiber, a Revolt-based client suspends and waits for its event loop + * to resume it; only from {main} does it run the loop itself and carry on. + */ + public static function awaitEventLoop(): void + { + if (null === \Fiber::getCurrent()) { + return; + } + + $resumedWith = \Fiber::suspend(); + + throw new \LogicException(\sprintf('Resumed by something other than the event loop, with %s.', get_debug_type($resumedWith))); + } + + public function __toString(): string + { + self::awaitEventLoop(); + + return (string) $this->inner; + } + + public function close(): void + { + $this->inner->close(); + } + + public function detach() + { + return $this->inner->detach(); + } + + public function getSize(): ?int + { + return $this->inner->getSize(); + } + + public function tell(): int + { + return $this->inner->tell(); + } + + public function eof(): bool + { + return $this->inner->eof(); + } + + public function isSeekable(): bool + { + return $this->inner->isSeekable(); + } + + public function seek(int $offset, int $whence = \SEEK_SET): void + { + $this->inner->seek($offset, $whence); + } + + public function rewind(): void + { + $this->inner->rewind(); + } + + public function isWritable(): bool + { + return false; + } + + public function write(string $string): int + { + throw new \LogicException('A response body is not writable.'); + } + + public function isReadable(): bool + { + return true; + } + + public function read(int $length): string + { + self::awaitEventLoop(); + + return $this->inner->read($length); + } + + public function getContents(): string + { + self::awaitEventLoop(); + + return $this->inner->getContents(); + } + + public function getMetadata(?string $key = null) + { + return null === $key ? [] : null; + } +}