From 35e54093160cc0e3b562cf3f82156df1fd495bf1 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Thu, 15 Oct 2026 01:43:45 +0200 Subject: [PATCH] [Server] Answer a request as SSE when the session has queued messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A JSON answer bundled the session's queued messages with the request's response into one array, which clients on 2025-06-18 and later reject. The queued messages now go out on an SSE answer ahead of the responses, and a JSON answer holds the response alone. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- CHANGELOG.md | 1 + .../Transport/StreamableHttpTransport.php | 21 ++++++--- .../Transport/StreamableHttpTransportTest.php | 43 +++++++++++++++---- 3 files changed, 52 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be4ac902..7b18a6bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * Fix parallel elicitations on one session getting each other's answers: requests to the client get UUID ids instead of a session counter, and a request or notification a handler sends goes out on the stream of its own call. * [BC Break] Ids of requests to the client are UUIDs: `Protocol::sendRequest()`, `Protocol::checkResponse()` and `Protocol::handleFiberYield()` take and return a `Uuid` instead of an `int`, as does the response finder of a transport. On the wire and in the session they are RFC 4122 strings. * Fix `StreamableHttpTransport` answering a POST of only responses or notifications with `200` and the session's queued messages: it is now accepted with `202` and no body, and the queued messages stay. +* Fix `StreamableHttpTransport` bundling the session's queued messages (e.g. `notifications/resources/updated`) with a request's response into a JSON array, which clients on `2025-06-18` and later reject: such a request is now answered as SSE, the queued messages first, and a JSON answer holds the response alone. * Fix stateless SSE streams holding back frames until close when PHP output buffering is enabled. * Fix a session deleted while one of its requests still ran coming back when that request saved it. * Reject a recognized `Mcp-Param-*` header whose mirrored argument is absent from the body with `-32020`, instead of accepting the request (SEP-2243). diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php index 867b5534..aee34971 100644 --- a/src/Server/Transport/StreamableHttpTransport.php +++ b/src/Server/Transport/StreamableHttpTransport.php @@ -255,14 +255,25 @@ protected function createJsonResponse(): ResponseInterface return $this->responseFactory->createResponse(202); } - $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->responses]; + $responses = $this->responses; $this->responses = []; - $responseBody = 1 === \count($messages) ? $messages[0] : '['.implode(',', $messages).']'; + // A JSON answer holds the responses alone, so messages queued for the + // session make it a stream that carries them ahead of the responses. + $queued = array_column($this->getOutgoingMessages($this->sessionId), 'message'); - $response = $this->responseFactory->createResponse(200) - ->withHeader('Content-Type', 'application/json') - ->withBody($this->streamFactory->createStream($responseBody)); + if ([] === $queued) { + $response = $this->responseFactory->createResponse(200) + ->withHeader('Content-Type', 'application/json') + ->withBody($this->streamFactory->createStream(1 === \count($responses) ? $responses[0] : '['.implode(',', $responses).']')); + } else { + $events = array_map(static fn (string $message): string => "event: message\ndata: {$message}\n\n", [...$queued, ...$responses]); + + $response = $this->responseFactory->createResponse(200) + ->withHeader('Content-Type', 'text/event-stream') + ->withHeader('Cache-Control', 'no-cache') + ->withBody($this->streamFactory->createStream(implode('', $events))); + } if ($this->sessionId) { $response = $response->withHeader(self::SESSION_HEADER, $this->sessionId->toRfc4122()); diff --git a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php index e113c645..b83c32c5 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php @@ -622,8 +622,21 @@ public function testStreamedBatchCarriesItsOtherResponses(): void $this->assertMatchesRegularExpression('/"id":3,"result".*"progressToken":"p".*"id":2,"result"/s', $output); } - #[TestDox('a batch answered as JSON carries the queued notifications first, then its responses, in one array')] - public function testJsonBatchCarriesQueuedNotificationsAndItsResponses(): void + /** + * @return iterable}> + */ + public static function provideRequestsMeetingQueuedNotifications(): iterable + { + yield 'single request' => ['{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}}', [2]]; + yield 'batch' => ['[{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}},{"jsonrpc":"2.0","id":3,"method":"ping"}]', [2, 3]]; + } + + /** + * @param list $ids + */ + #[DataProvider('provideRequestsMeetingQueuedNotifications')] + #[TestDox('a request meeting queued notifications is answered as SSE with them first, then its responses')] + public function testQueuedNotificationsGoOutOnAStreamBeforeTheResponses(string $body, array $ids): void { $store = new InterleavingSessionStore(); $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}') @@ -637,19 +650,33 @@ public function testJsonBatchCarriesQueuedNotificationsAndItsResponses(): void ]]); $session->save(); - $response = $this->post($store, '[{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}},{"jsonrpc":"2.0","id":3,"method":"ping"}]', $sessionId); + $response = $this->post($store, $body, $sessionId); $this->assertSame(200, $response->getStatusCode()); - $this->assertSame('application/json', $response->getHeaderLine('Content-Type')); + $this->assertSame('text/event-stream', $response->getHeaderLine('Content-Type')); $this->assertSame($sessionId, $response->getHeaderLine(StreamableHttpTransport::SESSION_HEADER)); - $messages = json_decode((string) $response->getBody(), true); - $this->assertIsArray($messages); - $this->assertTrue(array_is_list($messages)); + preg_match_all('/^data: (.+)$/m', (string) $response->getBody(), $matches); + $messages = array_map(static fn (string $data): array => json_decode($data, true), $matches[1]); $this->assertSame( - ['notifications/resources/updated', 2, 3], + ['notifications/resources/updated', ...$ids], array_map(static fn (array $message): string|int => $message['method'] ?? $message['id'], $messages), ); + + $this->assertSame([], (new Session($store, Uuid::fromString($sessionId)))->get('_mcp.outgoing_queue')); + } + + #[TestDox('a request answered as JSON carries its response alone, a single object')] + public function testJsonAnswerIsTheResponseAlone(): void + { + $store = new InterleavingSessionStore(); + $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}') + ->getHeaderLine(StreamableHttpTransport::SESSION_HEADER); + + $response = $this->post($store, '{"jsonrpc":"2.0","id":2,"method":"ping"}', $sessionId); + + $this->assertSame('application/json', $response->getHeaderLine('Content-Type')); + $this->assertSame(['jsonrpc' => '2.0', 'id' => 2, 'result' => []], json_decode((string) $response->getBody(), true)); } #[TestDox('with a session lock, a POST arriving while another of the session runs is held off instead of overwriting it')]