diff --git a/CHANGELOG.md b/CHANGELOG.md index eb1ea471..645d5f29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * [BC Break] `ProtectedResourceMetadata` requires `$resource`, serves at the path derived from it (RFC 9728 ยง3.1) and requires https except for loopback hosts; drops localized, policy, ToS, extra fields and `$metadataPaths`. * [BC Break] Add `ScopePolicy` as third argument of `AuthorizationMiddleware`, answering `403 insufficient_scope` per method and tool, with scope hierarchies; the `resource_metadata` challenge URL comes from the configured resource instead of the `Host` header. * Expose `WWW-Authenticate` in the default `CorsMiddleware`. +* 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 19e88852..25c13c01 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -30,6 +30,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; @@ -473,7 +474,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(), ]); @@ -496,6 +500,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', [ @@ -541,8 +549,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; @@ -77,6 +77,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; @@ -170,6 +173,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; } @@ -205,6 +214,8 @@ protected function handlePostRequest(string $body, ?AccessToken $accessToken = n $this->immediateStatusCode = null; if (null !== $immediateResponse) { + $this->inlineResponses = []; + return $this->responseFactory->createResponse($immediateStatusCode ?? 200) ->withHeader('Content-Type', 'application/json') ->withBody($this->streamFactory->createStream($immediateResponse)); @@ -232,14 +243,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) @@ -257,7 +268,11 @@ protected function createStreamedResponse(): ResponseInterface { $fiber = $this->sessionFiber; - $callback = function () use ($fiber): void { + // The other requests of a batch whose handler did not suspend. + $inlineResponses = $this->inlineResponses; + $this->inlineResponses = []; + + $callback = function () use ($fiber, $inlineResponses): void { if (null === $fiber) { return; } @@ -265,6 +280,13 @@ protected function createStreamedResponse(): ResponseInterface 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 ($fiber->isSuspended()) { $this->flushOutgoingMessages($this->sessionId); diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php index 6e9a5730..8a8601cf 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/src/Server/Transport/TransportInterface.php @@ -51,7 +51,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/ProtocolSessionRaceTest.php b/tests/Unit/Server/ProtocolSessionRaceTest.php new file mode 100644 index 00000000..73f9e268 --- /dev/null +++ b/tests/Unit/Server/ProtocolSessionRaceTest.php @@ -0,0 +1,54 @@ +createWithId($sessionId)->save(); + + $waiting = new Protocol([], [], MessageFactory::make(), $sessions); + $answering = new Protocol([], [], MessageFactory::make(), $sessions); + $transport = $this->createMock(TransportInterface::class); + + // The answer lands right after the waiting worker read the session, + // before anything it does next could write the session back. + $store->interleaveAfterNextRead(static function () use ($answering, $transport, $sessionId): void { + $answering->processInput($transport, '{"jsonrpc": "2.0", "id": 7, "result": {"ok": true}}', $sessionId); + }); + + // One turn of the waiting worker's loop, with nothing queued to send. + $this->assertSame([], $waiting->consumeOutgoingMessages($sessionId)); + + $this->assertInstanceOf(Response::class, $waiting->checkResponse(7, $sessionId)); + } +} diff --git a/tests/Unit/Server/Session/Fixture/InterleavingSessionStore.php b/tests/Unit/Server/Session/Fixture/InterleavingSessionStore.php new file mode 100644 index 00000000..e464ed44 --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/InterleavingSessionStore.php @@ -0,0 +1,84 @@ +afterNextRead = $interleaved; + } + + /** + * 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->afterNextWrite = $interleaved; + $this->readBeforeWrite = $readBeforeWrite; + } + + public function read(Uuid $id): string|false + { + if (null !== $data = $this->staleRead) { + $this->staleRead = null; + + return $data; + } + + $data = parent::read($id); + + if (null !== $interleaved = $this->afterNextRead) { + $this->afterNextRead = null; + $interleaved(); + } + + return $data; + } + + public function write(Uuid $id, string $data): bool + { + $before = parent::read($id); + $written = parent::write($id, $data); + + if (null !== $interleaved = $this->afterNextWrite) { + $this->afterNextWrite = 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 9905be5e..9716b335 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php @@ -13,13 +13,17 @@ 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\PassthroughMiddleware; use Mcp\Server\Transport\Http\Middleware\ProtocolVersionMiddleware; use Mcp\Server\Transport\StreamableHttpTransport; use Mcp\Server\Transport\TransportInterface; +use Mcp\Tests\Unit\Server\Session\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; @@ -463,6 +467,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 {