[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,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;
         }

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<string, string> $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;
+    }
+}