diff --git a/CHANGELOG.md b/CHANGELOG.md index 458177c8..e8cf8f9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * [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`. * [BC Break] Fix concurrent Streamable HTTP streams on one session resuming each other's fibers: each stream now polls only the client request its own fiber sent, so an elicitation answer reaches the tool call that asked for it. `Protocol::handleFiberYield()` returns the ID of the request it sent. +* 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 e31afcc0..c6fea38b 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -32,6 +32,7 @@ use Mcp\Server\Stateless\RequestStateCodec; use Mcp\Server\Suspension\NotificationSuspension; use Mcp\Server\Suspension\RequestSuspension; +use Mcp\Server\Transport\InlineResponseTransportInterface; use Mcp\Server\Transport\TransportInterface; use Psr\EventDispatcher\EventDispatcherInterface; use Psr\Log\LoggerInterface; @@ -488,8 +489,11 @@ public function sendNotification(Notification $notification, SessionInterface $s */ private function sendResponse(TransportInterface $transport, Response|Error $response, ?SessionInterface $session, array $context = []): void { - if (null === $session) { - $this->logger->info('Sending immediate response', [ + // 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->debug('Sending immediate response', [ 'response_id' => $response->getId(), ]); @@ -511,6 +515,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', [ @@ -556,8 +564,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 d35fd3e2..00d8b41f 100644 --- a/src/Server/Transport/TransportInterface.php +++ b/src/Server/Transport/TransportInterface.php @@ -48,7 +48,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/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index 54dd81ad..0a13b9e4 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -34,6 +34,8 @@ use Mcp\Server\Session\SessionManagerInterface; use Mcp\Server\Suspension\NotificationSuspension; use Mcp\Server\Suspension\RequestSuspension; +use Mcp\Server\Transport\InlineResponseTransportInterface; +use Mcp\Server\Transport\InMemoryTransport; use Mcp\Server\Transport\TransportInterface; use Mcp\Tests\Unit\Fixtures\PollingLoopTransport; use Mcp\Tests\Unit\Fixtures\ThrowingRequest; @@ -1034,6 +1036,48 @@ public function testSuccessfulRequestReturnsResponseWithSessionId(): void $this->assertEquals(['status' => 'ok'], $message['result']); } + #[TestDox('An inline response transport gets the response directly, not through the session queue')] + public function testInlineResponseTransportGetsResponseDirectly(): void + { + $handler = $this->createMock(RequestHandlerInterface::class); + $handler->method('supports')->willReturn(true); + $handler->method('handle')->willReturn(new Response(1, ['status' => 'ok'])); + + $sessions = new SessionManager(new InMemorySessionStore(), gcProbability: 0); + $sessionId = Uuid::v4(); + $sessions->createWithId($sessionId)->save(); + + $transport = new class extends InMemoryTransport implements InlineResponseTransportInterface { + /** @var list}> */ + public array $sent = []; + + public function send(string $data, array $context): void + { + $this->sent[] = [$data, $context]; + } + }; + + $protocol = new Protocol( + requestHandlers: [$handler], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $sessions, + ); + + $protocol->processInput( + $transport, + '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}', + $sessionId + ); + + $this->assertCount(1, $transport->sent); + [$data, $context] = $transport->sent[0]; + $this->assertSame(['status' => 'ok'], json_decode($data, true)['result']); + $this->assertSame('response', $context['type']); + $this->assertEquals($sessionId, $context['session_id']); + $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId)); + } + #[TestDox('Batch requests are processed and send multiple responses')] public function testBatchRequestsAreProcessed(): void { 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..2e0a7a28 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php @@ -13,13 +13,18 @@ use Mcp\Exception\InvalidArgumentException; use Mcp\Schema\JsonRpc\Error; +use Mcp\Server; +use Mcp\Server\RequestContext; +use Mcp\Server\Session\Session; 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 +468,127 @@ 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); + } + + #[TestDox('a batch answered as JSON carries the queued notifications first, then its responses, in one array')] + public function testJsonBatchCarriesQueuedNotificationsAndInlineResponses(): 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); + + // A notification another request of the session queued, e.g. a resource update. + $session = new Session($store, Uuid::fromString($sessionId)); + $session->set('_mcp.outgoing_queue', [[ + 'message' => '{"jsonrpc":"2.0","method":"notifications/resources/updated","params":{"uri":"file:///a"}}', + 'context' => ['type' => 'notification'], + ]]); + $session->save(); + + $response = $this->post($store, '[{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}},{"jsonrpc":"2.0","id":3,"method":"ping"}]', $sessionId); + + $this->assertSame(200, $response->getStatusCode()); + $this->assertSame('application/json', $response->getHeaderLine('Content-Type')); + $this->assertSame($sessionId, $response->getHeaderLine(StreamableHttpTransport::SESSION_HEADER)); + + $messages = json_decode((string) $response->getBody(), true); + $this->assertIsArray($messages); + $this->assertTrue(array_is_list($messages)); + $this->assertSame( + ['notifications/resources/updated', 2, 3], + array_map(static fn (array $message): string|int => $message['method'] ?? $message['id'], $messages), + ); + } + + /** + * 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 {