[Server] Fix lost responses on concurrent requests of the same session (Streamable HTTP)

CHANGELOG.md

@@ -13,6 +13,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 `JwtTokenValidator` with several issuers always fetching the keys of the first one: keys now come from the issuer the token claims, which must be configured.
+* Fix lost responses on concurrent requests of one session over Streamable HTTP: a POST is answered with its own responses instead of taking them from the session's outgoing queue. Adds `InlineResponseTransportInterface` for transports that answer each request on the exchange that carried it.
 
 0.8.0
 -----

src/Server/Protocol.php

@@ -29,6 +29,7 @@
 use Mcp\Server\Session\SessionManagerInterface;
 use Mcp\Server\Stateless\InputContext;
 use Mcp\Server\Stateless\RequestStateCodec;
+use Mcp\Server\Transport\InlineResponseTransportInterface;
 use Mcp\Server\Transport\TransportInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Psr\Log\LoggerInterface;
@@ -449,7 +450,10 @@ public function sendNotification(Notification $notification, SessionInterface $s
      */
     private function sendResponse(TransportInterface $transport, Response|Error $response, ?SessionInterface $session, array $context = []): void
     {
-        if (null === $session) {
+        // Queued in the session, a response can be overwritten or taken by a concurrent
+        // request of the same session: a transport that can answer on the request's
+        // own exchange gets it directly.
+        if (null === $session || $transport instanceof InlineResponseTransportInterface) {
             $this->logger->info('Sending immediate response', [
                 'response_id' => $response->getId(),
             ]);
@@ -472,6 +476,10 @@ private function sendResponse(TransportInterface $transport, Response|Error $res
             }
 
             $context['type'] = 'response';
+            if (null !== $session) {
+                $context['session_id'] = $session->getId();
+            }
+
             $transport->send($encoded, $context);
         } else {
             $this->logger->info('Queueing server response', [
@@ -517,8 +525,12 @@ public function consumeOutgoingMessages(Uuid $sessionId): array
     {
         $session = $this->sessionManager->createWithId($sessionId);
         $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []);
-        $session->set(self::SESSION_OUTGOING_QUEUE, []);
-        $session->save();
+
+        // Saving an unchanged session would only overwrite what a concurrent request saved in the meantime.
+        if ([] !== $queue) {
+            $session->set(self::SESSION_OUTGOING_QUEUE, []);
+            $session->save();
+        }
 
         return $queue;
     }

src/Server/Transport/InlineResponseTransportInterface.php

@@ -0,0 +1,27 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Server\Transport;
+
+/**
+ * A transport that answers each request on the exchange that carried it.
+ *
+ * {@see \Mcp\Server\Protocol} hands such a transport its responses through
+ * {@see TransportInterface::send()}, with the session in the `session_id`
+ * context key, instead of queueing them in the session. The session is shared
+ * by the concurrent requests of a client and written back whole, so a queued
+ * response can be overwritten by another request, or taken by it.
+ *
+ * Server-initiated requests and notifications still go through the session queue.
+ */
+interface InlineResponseTransportInterface
+{
+}

src/Server/Transport/StreamableHttpTransport.php

@@ -51,7 +51,7 @@
  *
  * @author Kyrian Obikwelu <koshnawaza@gmail.com>
  */
-class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface
+class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface, InlineResponseTransportInterface
 {
     use ReadsBoundedBody;
 
@@ -74,6 +74,9 @@ class StreamableHttpTransport extends BaseTransport implements StatelessAwareTra
     private ?string $immediateResponse = null;
     private ?int $immediateStatusCode = null;
 
+    /** @var list<string> responses to the requests of the current POST, see {@see InlineResponseTransportInterface} */
+    private array $inlineResponses = [];
+
     /** @var list<MiddlewareInterface>|null null until {@see self::listen()} resolves the defaults */
     private ?array $middleware;
 
@@ -162,6 +165,12 @@ public function connectStateless(StatelessProtocol $protocol): void
 
     public function send(string $data, array $context): void
     {
+        if (isset($context['session_id'])) {
+            $this->inlineResponses[] = $data;
+
+            return;
+        }
+
         $this->immediateResponse = $data;
         $this->immediateStatusCode = $context['status_code'] ?? 200;
     }
@@ -196,6 +205,8 @@ protected function handlePostRequest(string $body): ResponseInterface
         $this->immediateStatusCode = null;
 
         if (null !== $immediateResponse) {
+            $this->inlineResponses = [];
+
             return $this->responseFactory->createResponse($immediateStatusCode ?? 200)
                 ->withHeader('Content-Type', 'application/json')
                 ->withBody($this->streamFactory->createStream($immediateResponse));
@@ -223,14 +234,14 @@ protected function handleDeleteRequest(): ResponseInterface
 
     protected function createJsonResponse(): ResponseInterface
     {
-        $outgoingMessages = $this->getOutgoingMessages($this->sessionId);
+        $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->inlineResponses];
+        $this->inlineResponses = [];
 
-        if (empty($outgoingMessages)) {
+        if ([] === $messages) {
             return $this->responseFactory->createResponse(202)
                 ->withHeader('Content-Type', 'application/json');
         }
 
-        $messages = array_column($outgoingMessages, 'message');
         $responseBody = 1 === \count($messages) ? $messages[0] : '['.implode(',', $messages).']';
 
         $response = $this->responseFactory->createResponse(200)
@@ -246,10 +257,21 @@ protected function createJsonResponse(): ResponseInterface
 
     protected function createStreamedResponse(): ResponseInterface
     {
-        $callback = function (): void {
+        // The other requests of a batch whose handler did not suspend.
+        $inlineResponses = $this->inlineResponses;
+        $this->inlineResponses = [];
+
+        $callback = function () use ($inlineResponses): void {
             try {
                 $this->logger->info('SSE: Starting request processing loop');
 
+                foreach ($inlineResponses as $message) {
+                    echo "event: message\n";
+                    echo "data: {$message}\n\n";
+                    @ob_flush();
+                    flush();
+                }
+
                 while ($this->sessionFiber->isSuspended()) {
                     $this->flushOutgoingMessages($this->sessionId);
 

src/Server/Transport/TransportInterface.php

@@ -50,7 +50,8 @@ public function listen(): mixed;
     /**
      * Send a message to the client immediately (bypassing session queue).
      *
-     * Used for session resolution errors when no session is available.
+     * Used for session resolution errors when no session is available, and for
+     * every response on a {@see InlineResponseTransportInterface}.
      * The transport decides HOW to send based on context.
      *
      * @param array<string, mixed> $context Context about this message:

tests/Unit/Server/Transport/Fixture/InterleavingSessionStore.php

@@ -0,0 +1,68 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Tests\Unit\Server\Transport\Fixture;
+
+use Mcp\Server\Session\InMemorySessionStore;
+use Symfony\Component\Uid\Uuid;
+
+/**
+ * A session store that runs a second request while the first one saves its session.
+ *
+ * Replays, in one process and in a fixed order, what two PHP workers serving the
+ * same session do when their requests overlap.
+ */
+final class InterleavingSessionStore extends InMemorySessionStore
+{
+    private ?\Closure $interleaved = null;
+    private bool $readBeforeWrite = false;
+    private string|false|null $staleRead = null;
+
+    /**
+     * Runs $interleaved right after the next write.
+     *
+     * With $readBeforeWrite, the interleaved request reads the session as it was
+     * before that write, as if it had loaded it before the first request saved.
+     */
+    public function interleaveOnNextWrite(\Closure $interleaved, bool $readBeforeWrite = false): void
+    {
+        $this->interleaved = $interleaved;
+        $this->readBeforeWrite = $readBeforeWrite;
+    }
+
+    public function read(Uuid $id): string|false
+    {
+        if (null !== $data = $this->staleRead) {
+            $this->staleRead = null;
+
+            return $data;
+        }
+
+        return parent::read($id);
+    }
+
+    public function write(Uuid $id, string $data): bool
+    {
+        $before = parent::read($id);
+        $written = parent::write($id, $data);
+
+        if (null !== $interleaved = $this->interleaved) {
+            $this->interleaved = null;
+            if ($this->readBeforeWrite) {
+                $this->staleRead = $before;
+            }
+
+            $interleaved();
+        }
+
+        return $written;
+    }
+}

tests/Unit/Server/Transport/StreamableHttpTransportTest.php

@@ -13,12 +13,16 @@
 
 use Mcp\Exception\InvalidArgumentException;
 use Mcp\Schema\JsonRpc\Error;
+use Mcp\Server;
+use Mcp\Server\RequestContext;
 use Mcp\Server\Transport\Http\Middleware\CorsMiddleware;
 use Mcp\Server\Transport\Http\Middleware\DnsRebindingProtectionMiddleware;
 use Mcp\Server\Transport\Http\Middleware\ProtocolVersionMiddleware;
 use Mcp\Server\Transport\StreamableHttpTransport;
 use Mcp\Server\Transport\TransportInterface;
+use Mcp\Tests\Unit\Server\Transport\Fixture\InterleavingSessionStore;
 use Nyholm\Psr7\Factory\Psr17Factory;
+use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\TestDox;
 use PHPUnit\Framework\TestCase;
 use Psr\Clock\ClockInterface;
@@ -437,6 +441,97 @@ public function now(): \DateTimeImmutable
         $this->assertInstanceOf(Error::class, $received);
     }
 
+    /**
+     * @return iterable<string, array{bool}>
+     */
+    public static function provideInterleavings(): iterable
+    {
+        yield 'B runs between A saving its session and A answering' => [false];
+        yield 'B loaded the session before A saved it (lost update)' => [true];
+    }
+
+    #[TestDox('concurrent POSTs of one session each get their own response: $_dataName')]
+    #[DataProvider('provideInterleavings')]
+    public function testConcurrentPostsOfOneSessionEachGetTheirOwnResponse(bool $readBeforeWrite): 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);
+
+        $responseB = null;
+        $store->interleaveOnNextWrite(function () use ($store, $sessionId, &$responseB): void {
+            $responseB = $this->post($store, '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"echo","arguments":{"text":"b"}}}', $sessionId);
+        }, $readBeforeWrite);
+
+        $responseA = $this->post($store, '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}}', $sessionId);
+
+        $this->assertInstanceOf(ResponseInterface::class, $responseB);
+        foreach ([2 => $responseA, 3 => $responseB] as $id => $response) {
+            $this->assertSame(200, $response->getStatusCode(), \sprintf('Request %d was answered %d.', $id, $response->getStatusCode()));
+            $this->assertSame($sessionId, $response->getHeaderLine(StreamableHttpTransport::SESSION_HEADER));
+            $this->assertSame($id, json_decode((string) $response->getBody(), true)['id'] ?? null, \sprintf('Request %d got: %s', $id, $response->getBody()));
+        }
+    }
+
+    #[TestDox('a batch streamed over SSE still carries the responses that did not suspend')]
+    public function testStreamedBatchCarriesInlineResponses(): 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":"tools/call","params":{"name":"progress","arguments":{},"_meta":{"progressToken":"p"}}},{"jsonrpc":"2.0","id":3,"method":"ping"}]', $sessionId);
+
+        $this->assertSame('text/event-stream', $response->getHeaderLine('Content-Type'));
+
+        // The stream calls ob_flush() itself, so the output is captured by a handler, not a plain buffer.
+        $output = '';
+        ob_start(static function (string $chunk) use (&$output): string {
+            $output .= $chunk;
+
+            return '';
+        });
+        try {
+            $response->getBody()->getContents();
+        } finally {
+            ob_end_flush();
+        }
+
+        $this->assertMatchesRegularExpression('/"id":3,"result".*"progressToken":"p".*"id":2,"result"/s', $output);
+    }
+
+    /**
+     * Sends one POST to a fresh server sharing $store, like a PHP worker would.
+     */
+    private function post(InterleavingSessionStore $store, string $body, string $sessionId = ''): ResponseInterface
+    {
+        $request = $this->factory
+            ->createServerRequest('POST', 'http://localhost/')
+            ->withHeader('Host', 'localhost')
+            ->withHeader('Content-Type', 'application/json')
+            ->withHeader('Accept', 'application/json, text/event-stream')
+            ->withBody($this->factory->createStream($body));
+
+        if ('' !== $sessionId) {
+            $request = $request
+                ->withHeader(StreamableHttpTransport::SESSION_HEADER, $sessionId)
+                ->withHeader(StreamableHttpTransport::PROTOCOL_VERSION_HEADER, '2025-06-18');
+        }
+
+        $server = Server::builder()
+            ->setServerInfo('test', '1.0')
+            ->setSession($store)
+            ->addTool(static fn (string $text): string => $text, 'echo')
+            ->addTool(static function (RequestContext $context): string {
+                $context->getClientGateway()->progress(0.5);
+
+                return 'done';
+            }, 'progress')
+            ->build();
+
+        return $server->run(new StreamableHttpTransport($request, $this->factory, $this->factory));
+    }
+
     private function stubAuth401(): MiddlewareInterface
     {
         return new class($this->factory) implements MiddlewareInterface {