[Client] Send HTTP requests from outside the request fiber

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.

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 <koshnawaza@gmail.com>
  */
 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,14 +143,15 @@ public function connect(): void
         $fiber = new \Fiber(fn () => $this->handleInitialize());
         $this->activeFiber = $fiber;
 
-        $fiber->start();
+        $this->activeSuspend = $fiber->start();
 
         while (!$fiber->isTerminated()) {
             $this->tick();
         }
 
         $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,12 @@ private function nonBlocking(StreamInterface $body, ?bool &$switched = null): St
 
     private function tick(): void
     {
+        $this->processSends();
+        // The send may have been the fiber's last step, leaving nothing to read for.
+        if ($this->activeFiber?->isTerminated()) {
+            return;
+        }
+
         $this->checkInterruption();
         $this->processSSEStream();
         $this->processListenStream();
@@ -570,7 +615,10 @@ 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:
+            // send it now, before any read on the abandoned stream can block.
+            $this->activeSuspend = $this->activeFiber->throw($error);
+            $this->processSends();
         }
     }
 
@@ -1000,6 +1048,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;
         }

tests/Unit/Client/Transport/HttpTransportTest.php

@@ -865,6 +865,73 @@ public function testExpiredDeadlineClosesBodyBeforeReading(): void
         $transport->runRequest($fiber);
     }
 
+    public function testCancellationNotificationIsSentBeforeTheStreamIsRead(): void
+    {
+        $httpClient = RecordingHttpClient::queued([new Response(202)]);
+        $transport = $this->createTransport(httpClient: $httpClient);
+        $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);
+
+        // Like Protocol::exchange(), the fiber notifies the server before it rethrows.
+        $fiber = new \Fiber(static function () use ($transport): never {
+            try {
+                \Fiber::suspend([
+                    'type' => 'await_response',
+                    'request_id' => 1,
+                    'timeout' => 30,
+                    'deadline' => microtime(true) - 1,
+                ]);
+            } catch (TimeoutException $e) {
+                $transport->send('{"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":1}}');
+
+                throw $e;
+            }
+
+            throw new \LogicException('The deadline did not interrupt the request.');
+        });
+
+        try {
+            $transport->runRequest($fiber);
+            $this->fail('Expected a timeout.');
+        } catch (TimeoutException) {
+        }
+
+        $this->assertCount(1, $httpClient->requests);
+        $this->assertStringContainsString('notifications/cancelled', (string) $httpClient->requests[0]->getBody());
+    }
+
+    #[TestDox('an initialize stream the server leaves open is not read again once "initialized" went out')]
+    public function testInitializeStreamIsNotReadAfterHandshake(): void
+    {
+        $body = $this->createMock(StreamInterface::class);
+        $body->method('eof')->willReturn(false);
+        $body->expects($this->once())->method('read')->willReturn('data: '.json_encode([
+            'jsonrpc' => '2.0',
+            'id' => 1,
+            'result' => [
+                'protocolVersion' => '2025-11-25',
+                'capabilities' => [],
+                'serverInfo' => ['name' => 'test', 'version' => '1'],
+            ],
+        ], \JSON_THROW_ON_ERROR)."\n\n");
+
+        $httpClient = RecordingHttpClient::queued([
+            new Response(200, ['Content-Type' => 'text/event-stream'], $body),
+            new Response(202),
+        ]);
+
+        $client = Client::builder()->setClientInfo('test', '1')->setProtocolVersion(ProtocolVersion::V2025_11_25)->build();
+        $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory));
+
+        $this->assertTrue($client->isConnected());
+        $this->assertSame('notifications/initialized', json_decode((string) $httpClient->requests[1]->getBody(), true, flags: \JSON_THROW_ON_ERROR)['method'] ?? null);
+    }
+
     /** @return iterable<string, array{string}> */
     public static function notificationAnswerProvider(): iterable
     {
@@ -1206,6 +1273,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<string, string> $headers
      */
@@ -1529,3 +1629,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;
+    }
+}