From 9630f0966d4452fcc6f524ff39f257cd2c8561dd Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Sat, 10 Oct 2026 03:57:06 +0200 Subject: [PATCH] [Client] Fail a request at once when its response ends unanswered --- CHANGELOG.md | 1 + src/Client/Transport/HttpTransport.php | 94 ++++++++++++++++++- .../Client/Transport/HttpTransportTest.php | 52 ++++++++++ 3 files changed, 145 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e34852a9..88301dab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * Fix stateless SSE streams holding back frames until close when PHP output buffering is enabled. * Reject a recognized `Mcp-Param-*` header whose mirrored argument is absent from the body with `-32020`, instead of accepting the request (SEP-2243). * Fix `RequestEvent`, `ResponseEvent` and `ErrorEvent` not being dispatched for `2026-07-28` requests. +* Fix the client's `HttpTransport` waiting out the request timeout when a response stream or JSON body ends, is cut or fails to read without the answer: the call now fails at once with a `ConnectionException`. * [BC Break] Validate a tool result's `structuredContent` against the tool's `outputSchema`, which the specification requires the server to honour. A mismatch is answered with a `CallToolResult` carrying `isError: true` instead of the non-conforming value, matching the TypeScript, Python and Java SDKs. Skipped when the tool declares no `outputSchema`, when the result carries no `structuredContent`, and when the result is already an error. Return `new \stdClass()` for an empty object, since `[]` is sent as an array. * Stop the server `Protocol` from logging full JSON-RPC payloads (tool arguments, client replies) at info level: info records now carry only the method and id, the raw message is logged at debug level. * Add `PassthroughMiddleware` to opt `StreamableHttpTransport` out of its default middleware without the warning an empty `$middleware` list logs. diff --git a/src/Client/Transport/HttpTransport.php b/src/Client/Transport/HttpTransport.php index a6d777d3..8a285c61 100644 --- a/src/Client/Transport/HttpTransport.php +++ b/src/Client/Transport/HttpTransport.php @@ -58,6 +58,12 @@ class HttpTransport extends BaseTransport implements HeaderAwareTransportInterfa /** @var string Buffer for incomplete SSE data */ private string $sseBuffer = ''; + /** The request whose POST response opened the active SSE stream. */ + private int|string|null $sseRequestId = null; + + /** Whether the active SSE stream delivered the answer to that request. */ + private bool $sseRequestCompleted = false; + /** @var StreamInterface|null The standalone GET stream the server may send on unprompted */ private ?StreamInterface $listenStream = null; @@ -215,11 +221,24 @@ public function send(string $data): void ? $this->nonBlocking($response->getBody()) : $response->getBody(); $this->sseBuffer = ''; + $this->sseRequestId = self::requestId($data); + $this->sseRequestCompleted = false; } elseif (str_contains($contentType, 'application/json')) { - $body = $response->getBody()->getContents(); + try { + $body = $response->getBody()->getContents(); + } catch (\Throwable $e) { + throw self::unanswered($e); + } + if (!empty($body)) { $this->handleMessage($body); } + + // A cut or truncated body would otherwise leave the request waiting + // for an answer that can no longer arrive. + if (null !== ($requestId = self::requestId($data)) && !self::answers($body, $requestId)) { + throw self::unanswered(); + } } } @@ -234,6 +253,42 @@ private static function isNotification(string $data): bool return \is_array($payload) && \array_key_exists('method', $payload) && !\array_key_exists('id', $payload); } + /** + * The id of the outgoing message if it is a request, so an answer is owed. + */ + private static function requestId(string $data): int|string|null + { + $payload = json_decode($data, true); + $id = \is_array($payload) && \array_key_exists('method', $payload) ? ($payload['id'] ?? null) : null; + + return \is_int($id) || \is_string($id) ? $id : null; + } + + /** + * Whether the received message, or a message of the received batch, answers the request. + */ + private static function answers(string $data, int|string $requestId): bool + { + $decoded = json_decode($data, true); + if (!\is_array($decoded)) { + return false; + } + + foreach (array_is_list($decoded) ? $decoded : [$decoded] as $message) { + if (\is_array($message) && !\array_key_exists('method', $message) && ($message['id'] ?? null) === $requestId + && (\array_key_exists('result', $message) || \array_key_exists('error', $message))) { + return true; + } + } + + return false; + } + + private static function unanswered(?\Throwable $previous = null): ConnectionException + { + return new ConnectionException(\sprintf('The response ended without answering the request%s.', null !== $previous ? ': '.$previous->getMessage() : ''), 0, $previous); + } + /** * Fails a request refused at the HTTP level at once, rather than at its timeout. */ @@ -296,6 +351,7 @@ public function runRequest(\Fiber $fiber, ?callable $onProgress = null): Respons $this->activeStream?->close(); $this->activeStream = null; $this->sseBuffer = ''; + $this->sseRequestId = null; } } @@ -486,14 +542,43 @@ private function processSSEStream(): void return; } - $done = $this->pumpSse($this->activeStream, $this->sseBuffer, $this->abortSseStream(...)); + try { + $done = $this->pumpSse($this->activeStream, $this->sseBuffer, $this->abortSseStream(...)); + } catch (\Throwable $e) { + $this->sseBuffer = ''; + $this->activeStream = null; + $this->failUnansweredRequest($e); + + return; + } if ($done) { $this->sseBuffer = ''; $this->activeStream = null; + $this->failUnansweredRequest(); } } + /** + * Fail the request the ended SSE stream was opened for, unless it was answered. + * + * Nothing else can deliver the answer once its stream is gone, so the + * waiting fiber fails now instead of spinning until the request timeout. + */ + private function failUnansweredRequest(?\Throwable $previous = null): void + { + $requestId = $this->sseRequestId; + $this->sseRequestId = null; + + if (null === $requestId || $this->sseRequestCompleted || !isset($this->state?->getPendingRequests()[$requestId]) + || !$this->activeFiber?->isSuspended()) { + return; + } + + $this->logger->warning('Response stream ended without answering the request', ['request_id' => $requestId, 'exception' => $previous]); + $this->activeSuspend = $this->activeFiber->throw(self::unanswered($previous)); + } + /** * Read what arrived on the listening stream, if one is open. * @@ -567,6 +652,7 @@ private function abortSseStream(string $reason): void $bufferedBytes = \strlen($this->sseBuffer); $this->sseBuffer = ''; $this->activeStream = null; + $this->sseRequestId = null; $this->logger->warning('Aborting SSE stream: '.$reason, [ 'session_id' => $this->sessionId, @@ -630,6 +716,10 @@ private function processSSEEvent(string $event): void } if (!empty($data)) { + if (null !== $this->sseRequestId && self::answers($data, $this->sseRequestId)) { + $this->sseRequestCompleted = true; + } + $this->handleMessage($data); } } diff --git a/tests/Unit/Client/Transport/HttpTransportTest.php b/tests/Unit/Client/Transport/HttpTransportTest.php index 5a402896..9c137209 100644 --- a/tests/Unit/Client/Transport/HttpTransportTest.php +++ b/tests/Unit/Client/Transport/HttpTransportTest.php @@ -567,6 +567,58 @@ public function isCancellationRequested(): bool $client->disconnect(); } + /** + * A null body is one whose read fails, like a connection reset. + * + * @return iterable + */ + public static function cutResponseProvider(): iterable + { + yield 'an SSE stream ending before the answer' => ['text/event-stream', 'data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"working"}}'."\n\n"]; + yield 'an SSE stream cut mid-event' => ['text/event-stream', 'data: {"jsonrpc":"2.0","id":%d,"result":{"cont']; + yield 'an SSE stream failing to read' => ['text/event-stream', null]; + yield 'a truncated JSON body' => ['application/json', '{"jsonrpc":"2.0","id":%d,"result":{"cont']; + yield 'an empty JSON body' => ['application/json', '']; + yield 'a JSON body failing to read' => ['application/json', null]; + } + + #[DataProvider('cutResponseProvider')] + #[TestDox('a response that ends without the answer fails the call at once: $_dataName')] + public function testCutResponseFailsTheCallAtOnce(string $contentType, ?string $body): void + { + $httpClient = new RecordingHttpClient(function (array $message) use ($contentType, $body): ?ResponseInterface { + if ('cut' !== ($message['params']['name'] ?? null)) { + return null; + } + + if (null !== $body) { + return new Response(200, ['Content-Type' => $contentType], \sprintf($body, $message['id'])); + } + + $stream = $this->createMock(StreamInterface::class); + $stream->method('eof')->willReturn(false); + $stream->method('read')->willThrowException(new \RuntimeException('Connection reset by peer')); + $stream->method('getContents')->willThrowException(new \RuntimeException('Connection reset by peer')); + + return new Response(200, ['Content-Type' => $contentType], $stream); + }); + + $client = Client::builder()->setClientInfo('test', '1')->setProtocolVersion(ProtocolVersion::V2025_11_25)->setRequestTimeout(30)->build(); + $client->connect(new HttpTransport('http://localhost/mcp', [], $httpClient, $this->factory, $this->factory)); + + $started = microtime(true); + try { + $client->callTool('cut'); + $this->fail('Expected the call to fail.'); + } catch (ConnectionException $e) { + $this->assertStringContainsString('without answering the request', $e->getMessage()); + } + + $this->assertLessThan(1, microtime(true) - $started, 'the cut must not be waited out like a slow answer'); + $this->assertSame('next call', $client->callTool('fast')->content[0]->text ?? null); + $client->disconnect(); + } + /** @return iterable */ public static function revisionProvider(): iterable {