[Server] Answer a JSON POST with only its own responses

CHANGELOG.md

@@ -7,7 +7,11 @@ All notable changes to `mcp/sdk` will be documented in this file.
 
 * [BC Break] Remove the `providerClass` argument of `#[CompletionProvider]`. Use `provider:`, which takes the same class-string and is now the first positional argument.
 * Add `HttpTransport::getSessionId()` to read the server-minted `Mcp-Session-Id`: a request-scoped caller can persist it and pass it back through the constructor's `$headers` on a later transport. Always `null` on `2026-07-28`, which removed protocol-level sessions.
+<<<<<<< HEAD
 * Fix OIDC discovery rejecting issuers with a trailing slash (e.g. Authentik, Auth0).
+=======
+* Fix `StreamableHttpTransport` returning responses of other requests as a JSON array when requests of one session run concurrently (PHP-FPM). `Protocol::consumeOutgoingMessages()` accepts the response ids to consume.
+>>>>>>> e0fced5 ([Server] Answer a JSON POST with only its own responses)
 
 0.8.0
 -----

src/Server/Protocol.php

@@ -509,18 +509,63 @@ private function queueOutgoing(Request|Notification|Response|Error $message, arr
     }
 
     /**
-     * Consume (get and clear) all outgoing messages for a session.
+     * Consume (get and clear) outgoing messages for a session.
+     *
+     * With response ids only the matching responses are taken, the rest stays queued.
+     * A null id matches an error without an id.
+     *
+     * @param list<int|string|null>|null $responseIds
      *
      * @return array<int, array{message: string, context: array<string, mixed>}>
      */
-    public function consumeOutgoingMessages(Uuid $sessionId): array
+    public function consumeOutgoingMessages(Uuid $sessionId, ?array $responseIds = null): array
     {
         $session = $this->sessionManager->createWithId($sessionId);
+        /** @var array<int, array{message: string, context: array<string, mixed>}> $queue */
         $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []);
-        $session->set(self::SESSION_OUTGOING_QUEUE, []);
-        $session->save();
 
-        return $queue;
+        if (null === $responseIds) {
+            $session->set(self::SESSION_OUTGOING_QUEUE, []);
+            $session->save();
+
+            return $queue;
+        }
+
+        $consumed = [];
+        $remaining = [];
+        foreach ($queue as $message) {
+            if (self::isResponseTo($message, $responseIds)) {
+                $consumed[] = $message;
+            } else {
+                $remaining[] = $message;
+            }
+        }
+
+        if ([] !== $consumed) {
+            $session->set(self::SESSION_OUTGOING_QUEUE, $remaining);
+            $session->save();
+        }
+
+        return $consumed;
+    }
+
+    /**
+     * @param array{message: string, context: array<string, mixed>} $message
+     * @param list<int|string|null>                                 $responseIds
+     */
+    private static function isResponseTo(array $message, array $responseIds): bool
+    {
+        if ('response' !== ($message['context']['type'] ?? null)) {
+            return false;
+        }
+
+        try {
+            $decoded = json_decode($message['message'], true, flags: \JSON_THROW_ON_ERROR);
+        } catch (\JsonException) {
+            return false;
+        }
+
+        return \is_array($decoded) && \in_array($decoded['id'] ?? null, $responseIds, true);
     }
 
     /**

src/Server/Transport/BaseTransport.php

@@ -84,6 +84,20 @@ protected function getOutgoingMessages(?Uuid $sessionId): array
         return [];
     }
 
+    /**
+     * @param list<int|string|null> $responseIds
+     *
+     * @return array<int, array{message: string, context: array<string, mixed>}>
+     */
+    protected function getOutgoingResponses(?Uuid $sessionId, array $responseIds): array
+    {
+        if ($sessionId && \is_callable($this->outgoingMessagesProvider)) {
+            return ($this->outgoingMessagesProvider)($sessionId, $responseIds);
+        }
+
+        return [];
+    }
+
     /**
      * @return array<int, array<string, mixed>>
      */

src/Server/Transport/ManagesTransportCallbacks.php

@@ -32,7 +32,7 @@ trait ManagesTransportCallbacks
     /** @var callable(Uuid): void */
     protected $sessionEndListener;
 
-    /** @var callable(Uuid): array<int, array{message: string, context: array<string, mixed>}> */
+    /** @var callable(Uuid, list<int|string|null>|null=): array<int, array{message: string, context: array<string, mixed>}> */
     protected $outgoingMessagesProvider;
 
     /** @var callable(Uuid): array<int, array<string, mixed>> */

src/Server/Transport/StreamableHttpTransport.php

@@ -74,6 +74,10 @@ class StreamableHttpTransport extends BaseTransport implements StatelessAwareTra
     private ?string $immediateResponse = null;
     private ?int $immediateStatusCode = null;
 
+    /** @var list<int|string|null> */
+    private array $expectedResponseIds = [];
+    private bool $batchRequest = false;
+
     /** @var list<MiddlewareInterface>|null null until {@see self::listen()} resolves the defaults */
     private ?array $middleware;
 
@@ -186,6 +190,9 @@ protected function handleOptionsRequest(): ResponseInterface
      */
     protected function handlePostRequest(string $body): ResponseInterface
     {
+        // Concurrent requests of one session share the outgoing queue.
+        [$this->expectedResponseIds, $this->batchRequest] = self::expectedResponses($body);
+
         $this->handleMessage($body, $this->sessionId);
 
         // Consume the immediate response exactly once, so a transport instance
@@ -223,15 +230,15 @@ protected function handleDeleteRequest(): ResponseInterface
 
     protected function createJsonResponse(): ResponseInterface
     {
-        $outgoingMessages = $this->getOutgoingMessages($this->sessionId);
+        $outgoingMessages = $this->getOutgoingResponses($this->sessionId, $this->expectedResponseIds);
 
         if (empty($outgoingMessages)) {
             return $this->responseFactory->createResponse(202)
                 ->withHeader('Content-Type', 'application/json');
         }
 
         $messages = array_column($outgoingMessages, 'message');
-        $responseBody = 1 === \count($messages) ? $messages[0] : '['.implode(',', $messages).']';
+        $responseBody = $this->batchRequest ? '['.implode(',', $messages).']' : $messages[0];
 
         $response = $this->responseFactory->createResponse(200)
             ->withHeader('Content-Type', 'application/json')
@@ -461,6 +468,39 @@ private function handleModernRequest(ServerRequestInterface $request, string $bo
         return $this->responder->respond($this->stateless->handle($body, self::headers($request)));
     }
 
+    /**
+     * Ids the body expects responses for (null for a message without a usable id), and whether it is a batch.
+     *
+     * @return array{list<int|string|null>, bool}
+     */
+    private static function expectedResponses(string $body): array
+    {
+        try {
+            $data = json_decode($body, true, flags: \JSON_THROW_ON_ERROR);
+        } catch (\JsonException) {
+            // Parse errors are sent directly, not queued.
+            return [[], false];
+        }
+
+        if (!\is_array($data) || [] === $data) {
+            return [[null], false];
+        }
+
+        $batch = array_is_list($data);
+        $ids = [];
+
+        foreach ($batch ? $data : [$data] as $message) {
+            if (\is_array($message) && (isset($message['result']) || isset($message['error']))) {
+                continue;
+            }
+
+            $id = \is_array($message) ? ($message['id'] ?? null) : null;
+            $ids[] = \is_int($id) || \is_string($id) ? $id : null;
+        }
+
+        return [$ids, $batch];
+    }
+
     /**
      * @return array<string, string>
      */

src/Server/Transport/TransportInterface.php

@@ -84,11 +84,12 @@ public function onMessage(callable $listener): void;
     public function onSessionEnd(callable $listener): void;
 
     /**
-     * Set a provider function to retrieve all queued outgoing messages.
+     * Set a provider function to retrieve queued outgoing messages.
      *
-     * The transport calls this to retrieve all queued messages for a session.
+     * The transport calls this to retrieve queued messages for a session. When response ids
+     * are passed, only the responses to those requests are returned.
      *
-     * @param callable(Uuid $sessionId): array<int, array{message: string, context: array<string, mixed>}> $provider
+     * @param callable(Uuid $sessionId, list<int|string|null>|null $responseIds=): array<int, array{message: string, context: array<string, mixed>}> $provider
      */
     public function setOutgoingMessagesProvider(callable $provider): void;
 

tests/Unit/Server/Transport/StreamableHttpTransportTest.php

@@ -12,7 +12,12 @@
 namespace Mcp\Tests\Unit\Server\Transport;
 
 use Mcp\Exception\InvalidArgumentException;
+use Mcp\JsonRpc\MessageFactory;
 use Mcp\Schema\JsonRpc\Error;
+use Mcp\Server\Handler\Request\PingHandler;
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\InMemorySessionStore;
+use Mcp\Server\Session\SessionManager;
 use Mcp\Server\Transport\Http\Middleware\CorsMiddleware;
 use Mcp\Server\Transport\Http\Middleware\DnsRebindingProtectionMiddleware;
 use Mcp\Server\Transport\Http\Middleware\ProtocolVersionMiddleware;
@@ -437,6 +442,140 @@ public function now(): \DateTimeImmutable
         $this->assertInstanceOf(Error::class, $received);
     }
 
+    #[TestDox('concurrent POSTs sharing a session each receive only their own response')]
+    public function testConcurrentPostsSharingASessionEachReceiveTheirOwnResponse(): void
+    {
+        [$protocol, $sessionId] = $this->createSessionProtocol();
+
+        $first = $this->createSessionPost('{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId);
+        $second = $this->createSessionPost('{"jsonrpc":"2.0","id":2,"method":"ping"}', $sessionId);
+        $protocol->connect($first);
+        $protocol->connect($second);
+
+        // The second worker finishes its request while the first one is still between queueing and reading.
+        $secondResponse = null;
+        $first->setOutgoingMessagesProvider(static function (Uuid $id, ?array $responseIds = null) use ($protocol, $second, &$secondResponse): array {
+            $secondResponse = $second->listen();
+
+            return $protocol->consumeOutgoingMessages($id, $responseIds);
+        });
+
+        $firstResponse = $first->listen();
+
+        $this->assertInstanceOf(ResponseInterface::class, $secondResponse);
+        $this->assertSingleJsonRpcResponse(2, $secondResponse);
+        $this->assertSingleJsonRpcResponse(1, $firstResponse);
+    }
+
+    #[TestDox('a batch POST is answered with its own responses only')]
+    public function testBatchPostReceivesOnlyItsOwnResponses(): void
+    {
+        [$protocol, $sessionId] = $this->createSessionProtocol();
+        $this->queueForeignResponse($protocol, $sessionId);
+
+        $transport = $this->createSessionPost('[{"jsonrpc":"2.0","id":3,"method":"ping"},{"jsonrpc":"2.0","id":4,"method":"ping"}]', $sessionId);
+        $protocol->connect($transport);
+
+        $response = $transport->listen();
+
+        $this->assertSame(200, $response->getStatusCode());
+        $body = json_decode((string) $response->getBody(), true, flags: \JSON_THROW_ON_ERROR);
+        $this->assertIsArray($body);
+        $this->assertTrue(array_is_list($body));
+        $this->assertSame([3, 4], array_column($body, 'id'));
+        $this->assertForeignResponseStillQueued($protocol, $sessionId);
+    }
+
+    #[TestDox('a POST without requests is accepted without taking responses queued for other requests')]
+    public function testPostWithoutRequestsLeavesOtherResponsesQueued(): void
+    {
+        [$protocol, $sessionId] = $this->createSessionProtocol();
+        $this->queueForeignResponse($protocol, $sessionId);
+
+        $transport = $this->createSessionPost('{"jsonrpc":"2.0","method":"notifications/initialized"}', $sessionId);
+        $protocol->connect($transport);
+
+        $response = $transport->listen();
+
+        $this->assertSame(202, $response->getStatusCode());
+        $this->assertSame('', (string) $response->getBody());
+        $this->assertForeignResponseStillQueued($protocol, $sessionId);
+    }
+
+    #[TestDox('an invalid message without a usable id is still answered with its error')]
+    public function testInvalidMessageWithoutIdIsAnsweredWithItsError(): void
+    {
+        [$protocol, $sessionId] = $this->createSessionProtocol();
+        $this->queueForeignResponse($protocol, $sessionId);
+
+        $transport = $this->createSessionPost('{"jsonrpc":"2.0","method":42}', $sessionId);
+        $protocol->connect($transport);
+
+        $response = $transport->listen();
+
+        $this->assertSame(200, $response->getStatusCode());
+        $body = json_decode((string) $response->getBody(), true, flags: \JSON_THROW_ON_ERROR);
+        $this->assertIsArray($body);
+        $this->assertFalse(array_is_list($body));
+        $this->assertSame(Error::INVALID_REQUEST, $body['error']['code']);
+        $this->assertArrayNotHasKey('id', $body);
+        $this->assertForeignResponseStillQueued($protocol, $sessionId);
+    }
+
+    /**
+     * @return array{Protocol, Uuid}
+     */
+    private function createSessionProtocol(): array
+    {
+        $sessionManager = new SessionManager(new InMemorySessionStore(), gcProbability: 0);
+        $session = $sessionManager->create();
+        $session->save();
+
+        $protocol = new Protocol(
+            requestHandlers: [new PingHandler()],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        return [$protocol, $session->getId()];
+    }
+
+    private function createSessionPost(string $body, Uuid $sessionId): StreamableHttpTransport
+    {
+        $request = $this->factory
+            ->createServerRequest('POST', 'http://localhost/')
+            ->withHeader('Host', 'localhost')
+            ->withHeader(StreamableHttpTransport::SESSION_HEADER, $sessionId->toRfc4122())
+            ->withBody($this->factory->createStream($body));
+
+        return new StreamableHttpTransport($request, $this->factory, $this->factory);
+    }
+
+    private function queueForeignResponse(Protocol $protocol, Uuid $sessionId): void
+    {
+        $protocol->processInput($this->createMock(TransportInterface::class), '{"jsonrpc":"2.0","id":9,"method":"ping"}', $sessionId);
+    }
+
+    private function assertForeignResponseStillQueued(Protocol $protocol, Uuid $sessionId): void
+    {
+        $queued = $protocol->consumeOutgoingMessages($sessionId);
+
+        $this->assertCount(1, $queued);
+        $this->assertSame(9, json_decode($queued[0]['message'], true, flags: \JSON_THROW_ON_ERROR)['id']);
+    }
+
+    private function assertSingleJsonRpcResponse(int $id, ResponseInterface $response): void
+    {
+        $this->assertSame(200, $response->getStatusCode());
+
+        $body = json_decode((string) $response->getBody(), true, flags: \JSON_THROW_ON_ERROR);
+
+        $this->assertIsArray($body);
+        $this->assertFalse(array_is_list($body), 'Expected a single JSON-RPC object.');
+        $this->assertSame($id, $body['id']);
+    }
+
     private function stubAuth401(): MiddlewareInterface
     {
         return new class($this->factory) implements MiddlewareInterface {