[Server] Stop bundling queued notifications into the JSON answer of a request

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).

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());

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<string, array{string, list<int>}>
+     */
+    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<int> $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')]