From 85aef2e718b6d84fb18c21d4aa01467c54dfa1fd Mon Sep 17 00:00:00 2001 From: Guillaume Sainthillier Date: Tue, 6 Oct 2026 09:24:55 +0200 Subject: [PATCH 1/3] [Server] Fix lost responses on concurrent requests of the same session (Streamable HTTP) Fixes #275 and #467. A response to a POST went through the session's outgoing queue, which concurrent requests of the same session read and write back whole. One request could then answer with another's response, and the other with an empty 202. StreamableHttpTransport now implements InlineResponseTransportInterface: Protocol hands it the responses through send() and the transport answers the POST with them. The queue keeps server-initiated requests and notifications. --- CHANGELOG.md | 1 + src/Server/Protocol.php | 18 +++- .../InlineResponseTransportInterface.php | 27 ++++++ .../Transport/StreamableHttpTransport.php | 32 ++++++- src/Server/Transport/TransportInterface.php | 3 +- .../Fixture/InterleavingSessionStore.php | 68 +++++++++++++ .../Transport/StreamableHttpTransportTest.php | 95 +++++++++++++++++++ 7 files changed, 235 insertions(+), 9 deletions(-) create mode 100644 src/Server/Transport/InlineResponseTransportInterface.php create mode 100644 tests/Unit/Server/Transport/Fixture/InterleavingSessionStore.php diff --git a/CHANGELOG.md b/CHANGELOG.md index a07ed6f7..de033c9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * 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 OIDC discovery rejecting issuers with a trailing slash (e.g. Authentik, Auth0). * Fix stateless SSE streams holding back frames until close when PHP output buffering is enabled. +* 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 ----- diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index f460c3c7..301338c0 100644 --- a/src/Server/Protocol.php +++ b/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; } diff --git a/src/Server/Transport/InlineResponseTransportInterface.php b/src/Server/Transport/InlineResponseTransportInterface.php new file mode 100644 index 00000000..ec4c2277 --- /dev/null +++ b/src/Server/Transport/InlineResponseTransportInterface.php @@ -0,0 +1,27 @@ + */ -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 responses to the requests of the current POST, see {@see InlineResponseTransportInterface} */ + private array $inlineResponses = []; + /** @var list|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); diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php index 58d09789..451ab85d 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/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 $context Context about this message: diff --git a/tests/Unit/Server/Transport/Fixture/InterleavingSessionStore.php b/tests/Unit/Server/Transport/Fixture/InterleavingSessionStore.php new file mode 100644 index 00000000..53a6a246 --- /dev/null +++ b/tests/Unit/Server/Transport/Fixture/InterleavingSessionStore.php @@ -0,0 +1,68 @@ +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; + } +} diff --git a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php index 186432fc..4dc50ba8 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/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 + */ + 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 { From 3effbcc549663f9d73c0bc23886c2f471618d62c Mon Sep 17 00:00:00 2001 From: Guillaume Sainthillier Date: Tue, 6 Oct 2026 09:32:56 +0200 Subject: [PATCH 2/3] [QA] Ignore the PHPStan 2.3.0 findings in ElicitationSchema PHPStan 2.3.0 reports the runtime checks of fromArray() as redundant with its phpdoc shape. They guard input from the wire, so keep them. --- src/Schema/Elicitation/ElicitationSchema.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Schema/Elicitation/ElicitationSchema.php b/src/Schema/Elicitation/ElicitationSchema.php index 36d130af..eb46637f 100644 --- a/src/Schema/Elicitation/ElicitationSchema.php +++ b/src/Schema/Elicitation/ElicitationSchema.php @@ -59,13 +59,13 @@ public static function fromArray(array $data): self throw new InvalidArgumentException('ElicitationSchema type must be "object".'); } - if (!isset($data['properties']) || !\is_array($data['properties'])) { + if (!isset($data['properties']) || !\is_array($data['properties'])) { /* @phpstan-ignore isset.offset */ throw new InvalidArgumentException('Missing or invalid "properties" for elicitation schema.'); } $properties = []; foreach ($data['properties'] as $name => $propertyData) { - if (!\is_array($propertyData)) { + if (!\is_array($propertyData)) { /* @phpstan-ignore function.alreadyNarrowedType */ throw new InvalidArgumentException(\sprintf('Property "%s" must be an array.', $name)); } $properties[$name] = self::createSchemaDefinition($propertyData); From 9e10e9273f04f05ac1b72805913bf50010fcd33c Mon Sep 17 00:00:00 2001 From: Guillaume Sainthillier Date: Tue, 6 Oct 2026 23:10:38 +0200 Subject: [PATCH 3/3] Revert "[QA] Ignore the PHPStan 2.3.0 findings in ElicitationSchema" This reverts commit 3effbcc549663f9d73c0bc23886c2f471618d62c. --- src/Schema/Elicitation/ElicitationSchema.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Schema/Elicitation/ElicitationSchema.php b/src/Schema/Elicitation/ElicitationSchema.php index eb46637f..36d130af 100644 --- a/src/Schema/Elicitation/ElicitationSchema.php +++ b/src/Schema/Elicitation/ElicitationSchema.php @@ -59,13 +59,13 @@ public static function fromArray(array $data): self throw new InvalidArgumentException('ElicitationSchema type must be "object".'); } - if (!isset($data['properties']) || !\is_array($data['properties'])) { /* @phpstan-ignore isset.offset */ + if (!isset($data['properties']) || !\is_array($data['properties'])) { throw new InvalidArgumentException('Missing or invalid "properties" for elicitation schema.'); } $properties = []; foreach ($data['properties'] as $name => $propertyData) { - if (!\is_array($propertyData)) { /* @phpstan-ignore function.alreadyNarrowedType */ + if (!\is_array($propertyData)) { throw new InvalidArgumentException(\sprintf('Property "%s" must be an array.', $name)); } $properties[$name] = self::createSchemaDefinition($propertyData);