From e0fced59f66e1e083613ef6508bdff5cd5d3b68f Mon Sep 17 00:00:00 2001 From: Vitalii Cherepanov Date: Mon, 14 Sep 2026 14:38:51 +0200 Subject: [PATCH] [Server] Answer a JSON POST with only its own responses --- CHANGELOG.md | 1 + src/Server/Protocol.php | 55 ++++++- src/Server/Transport/BaseTransport.php | 14 ++ .../Transport/ManagesTransportCallbacks.php | 2 +- .../Transport/StreamableHttpTransport.php | 44 +++++- src/Server/Transport/TransportInterface.php | 7 +- .../Transport/StreamableHttpTransportTest.php | 139 ++++++++++++++++++ 7 files changed, 251 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 22ad2b37..794cb217 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ 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. +* 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. 0.8.0 ----- diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index f460c3c7..3ced4baa 100644 --- a/src/Server/Protocol.php +++ b/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|null $responseIds * * @return array}> */ - public function consumeOutgoingMessages(Uuid $sessionId): array + public function consumeOutgoingMessages(Uuid $sessionId, ?array $responseIds = null): array { $session = $this->sessionManager->createWithId($sessionId); + /** @var array}> $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} $message + * @param list $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); } /** diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php index 58172352..d28926d5 100644 --- a/src/Server/Transport/BaseTransport.php +++ b/src/Server/Transport/BaseTransport.php @@ -84,6 +84,20 @@ protected function getOutgoingMessages(?Uuid $sessionId): array return []; } + /** + * @param list $responseIds + * + * @return array}> + */ + protected function getOutgoingResponses(?Uuid $sessionId, array $responseIds): array + { + if ($sessionId && \is_callable($this->outgoingMessagesProvider)) { + return ($this->outgoingMessagesProvider)($sessionId, $responseIds); + } + + return []; + } + /** * @return array> */ diff --git a/src/Server/Transport/ManagesTransportCallbacks.php b/src/Server/Transport/ManagesTransportCallbacks.php index 072d3f0e..00e720e3 100644 --- a/src/Server/Transport/ManagesTransportCallbacks.php +++ b/src/Server/Transport/ManagesTransportCallbacks.php @@ -32,7 +32,7 @@ trait ManagesTransportCallbacks /** @var callable(Uuid): void */ protected $sessionEndListener; - /** @var callable(Uuid): array}> */ + /** @var callable(Uuid, list|null=): array}> */ protected $outgoingMessagesProvider; /** @var callable(Uuid): array> */ diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php index c01503eb..12d78ff0 100644 --- a/src/Server/Transport/StreamableHttpTransport.php +++ b/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 */ + private array $expectedResponseIds = []; + private bool $batchRequest = false; + /** @var list|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,7 +230,7 @@ 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) @@ -231,7 +238,7 @@ protected function createJsonResponse(): ResponseInterface } $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, 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 */ diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php index 58d09789..8167f1c3 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/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}> $provider + * @param callable(Uuid $sessionId, list|null $responseIds=): array}> $provider */ public function setOutgoingMessagesProvider(callable $provider): void; diff --git a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php index 186432fc..fca89eaa 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/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 {