Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ All notable changes to `mcp/sdk` will be documented in this file.
* Dispatch `ResponseEvent` and `ErrorEvent` when a suspended Fiber completes.
* Fix a handler throwing after its fiber resumed (e.g. after elicitation or sampling) escaping to the transport: it is now answered with an error response like one throwing before suspending.
* Drop a client response to a request ID the server is not waiting on, or that timed out, instead of storing it in the session, where nothing would ever consume it.
* Fix a client request that timed out (elicitation, sampling) staying in the session's pending requests: `Protocol::checkResponse()` now expires it, answering with the `Request timed out` error, so the transports no longer implement the timeout themselves.

0.8.0
-----
Expand Down
39 changes: 35 additions & 4 deletions src/Server/Protocol.php
Original file line number Diff line number Diff line change
Expand Up @@ -593,9 +593,10 @@ public function consumeOutgoingMessages(Uuid $sessionId): array
* Check for a response to a specific request ID.
*
* When a response is found, it is removed from the session, and the
* corresponding pending request is also cleared.
*/
/**
* corresponding pending request is also cleared. A request that got no
* answer within its timeout is cleared the same way and reported as an
* internal error, so its pending entry does not outlive the wait.
*
* @return Response<array<string, mixed>>|Error|null
*/
public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|null
Expand All @@ -604,7 +605,9 @@ public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|n
$responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}");

if (null === $responseData) {
return null;
$pending = $session->get(self::SESSION_PENDING_REQUESTS, [])[$requestId] ?? null;

return \is_array($pending) && $this->hasTimedOut($pending) ? $this->expireRequest($requestId, $sessionId) : null;
}

$this->logger->debug('Found and consuming client response.', [
Expand Down Expand Up @@ -635,6 +638,34 @@ public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|n
}
}

/**
* Drops a timed out request from the pending ones and reports the timeout.
*
* The session is loaded again as the caller's copy may be as old as the last
* poll, and saving it back would undo what other streams stored since.
*/
private function expireRequest(int $requestId, Uuid $sessionId): ?Error
{
$session = $this->sessionManager->createWithId($sessionId);
$pending = $session->get(self::SESSION_PENDING_REQUESTS, []);

// Answered or expired by another stream in the meantime: the next poll sorts it out.
if (!isset($pending[$requestId]) || null !== $session->get(self::SESSION_RESPONSES.".{$requestId}")) {
return null;
}

$this->logger->warning('Client request timed out.', [
'request_id' => $requestId,
'session_id' => $sessionId->toRfc4122(),
]);

unset($pending[$requestId]);
$session->set(self::SESSION_PENDING_REQUESTS, $pending);
$session->save();

return Error::forInternalError('Request timed out', $requestId);
}

/**
* @param array<mixed> $pending
*/
Expand Down
10 changes: 0 additions & 10 deletions src/Server/Transport/StdioTransport.php
Original file line number Diff line number Diff line change
Expand Up @@ -368,8 +368,6 @@ private function processFiber(): void

foreach ($pendingRequests as $pending) {
$requestId = $pending['request_id'];
$timestamp = $pending['timestamp'];
$timeout = $pending['timeout'] ?? 120;

$response = $this->checkForResponse($requestId, $this->sessionId);

Expand All @@ -379,14 +377,6 @@ private function processFiber(): void

return;
}

if (time() - $timestamp >= $timeout) {
$error = Error::forInternalError('Request timed out', $requestId);
$yielded = $this->sessionFiber->resume($error);
$this->handleFiberYield($yielded, $this->sessionId);

return;
}
}
}

Expand Down
13 changes: 0 additions & 13 deletions src/Server/Transport/StreamableHttpTransport.php
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,13 @@
use Mcp\Schema\Enum\ProtocolVersion;
use Mcp\Schema\JsonRpc\Error;
use Mcp\Server\Authorization\AccessToken;
use Mcp\Server\NativeClock;
use Mcp\Server\Stateless\StatelessProtocol;
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\Http\MiddlewareRequestHandler;
use Mcp\Server\Transport\Http\StatelessResponder;
use Mcp\Server\Wire\InboundClassifier;
use Psr\Clock\ClockInterface;
use Psr\Http\Message\ResponseFactoryInterface;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
Expand Down Expand Up @@ -101,7 +99,6 @@ public function __construct(
?LoggerInterface $logger = null,
?iterable $middleware = null,
private readonly int $maxBodyBytes = self::DEFAULT_MAX_BODY_BYTES,
private readonly ClockInterface $clock = new NativeClock(),
) {
parent::__construct($logger);

Expand Down Expand Up @@ -310,8 +307,6 @@ protected function createStreamedResponse(): ResponseInterface
$resumed = false;
foreach ($pendingRequests as $pending) {
$requestId = $pending['request_id'];
$timestamp = $pending['timestamp'];
$timeout = $pending['timeout'] ?? 120;

$response = $this->checkForResponse($requestId, $this->sessionId);

Expand All @@ -321,14 +316,6 @@ protected function createStreamedResponse(): ResponseInterface
$resumed = true;
break;
}

if ($this->clock->now()->getTimestamp() - $timestamp >= $timeout) {
$error = Error::forInternalError('Request timed out', $requestId);
$yielded = $fiber->resume($error);
$this->handleFiberYield($yielded, $this->sessionId);
$resumed = true;
break;
}
}

if (!$resumed) {
Expand Down
5 changes: 4 additions & 1 deletion tests/Unit/Server/ProtocolSessionRaceTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
namespace Mcp\Tests\Unit\Server;

use Mcp\JsonRpc\MessageFactory;
use Mcp\Schema\JsonRpc\Error;
use Mcp\Schema\JsonRpc\Response;
use Mcp\Server\Protocol;
use Mcp\Server\Session\SessionManager;
Expand Down Expand Up @@ -76,7 +77,9 @@ public function testPollingATimedOutRequestDoesNotOverwriteAConcurrentResponse()
$answering->processInput($transport, '{"jsonrpc": "2.0", "id": 8, "result": {"ok": true}}', $sessionId);
});

$this->assertNull($waiting->checkResponse(7, $sessionId));
$timeout = $waiting->checkResponse(7, $sessionId);
$this->assertInstanceOf(Error::class, $timeout);
$this->assertSame(7, $timeout->getId());

$this->assertInstanceOf(Response::class, $waiting->checkResponse(8, $sessionId));
}
Expand Down
118 changes: 118 additions & 0 deletions tests/Unit/Server/ProtocolTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -978,6 +978,41 @@ public function testStreamSuspendedOnNotificationPollsNoPendingRequest(): void
$this->assertSame([1001], $secondStream->getPendingRequestIds());
}

#[TestDox('A stream stops polling a client request once it timed out')]
public function testStreamStopsPollingTimedOutRequest(): void
{
$handler = $this->createMock(RequestHandlerInterface::class);
$handler->method('supports')->willReturn(true);
$handler->method('handle')->willReturnCallback(static function (Request $request, SessionInterface $session): Response {
// A timeout of 0 seconds has expired by the first poll.
\Fiber::suspend(new RequestSuspension(new PingRequest(), $session->getId()->toRfc4122(), 0));

return new Response($request->getId(), []);
});

$sessionManager = new SessionManager(new InMemorySessionStore());
$protocol = new Protocol(
requestHandlers: [$handler],
notificationHandlers: [],
messageFactory: MessageFactory::make(),
sessionManager: $sessionManager,
);

$session = $sessionManager->create();
$session->save();
$sessionId = $session->getId();

$stream = new PollingLoopTransport();
$protocol->connect($stream);
$protocol->processInput($stream, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $sessionId);

$this->assertSame([1000], $stream->getPendingRequestIds());

$this->assertInstanceOf(Error::class, $protocol->checkResponse(1000, $sessionId));

$this->assertSame([], $stream->getPendingRequestIds());
}

/**
* Two tool calls on one session, each suspended on a request to the client, as with elicitation.
* A further call with ID 3 suspends on a notification instead.
Expand Down Expand Up @@ -2017,6 +2052,67 @@ public function testLateResponseToTimedOutRequestIsDropped(): void
$this->assertNull($sessionManager->createWithId($session->getId())->get('_mcp.responses.1000'));
}

#[TestDox('A timed out client request is reported as an error and its pending entry is dropped')]
public function testCheckResponseReportsTimedOutRequestAndDropsItsPendingEntry(): void
{
[$protocol, $sessionManager, $session] = $this->createProtocolWithPendingRequests([
1000 => ['request_id' => 1000, 'timeout' => 5, 'timestamp' => time() - 10],
1001 => ['request_id' => 1001, 'timeout' => 120, 'timestamp' => time()],
]);

$result = $protocol->checkResponse(1000, $session->getId());

$this->assertInstanceOf(Error::class, $result);
$this->assertSame(1000, $result->getId());
$this->assertSame(Error::INTERNAL_ERROR, $result->code);
$this->assertSame('Request timed out', $result->message);
$this->assertSame([1001], array_keys($protocol->getPendingRequests($session->getId())));
$this->assertSame([1001], array_keys($sessionManager->createWithId($session->getId())->get('_mcp.pending_requests')));

$this->assertNull($protocol->checkResponse(1000, $session->getId()), 'The timeout is reported once.');
}

#[TestDox('A client request within its timeout stays pending')]
public function testCheckResponseKeepsRequestWithinItsTimeout(): void
{
$pending = [1000 => ['request_id' => 1000, 'timeout' => 120, 'timestamp' => time() - 10]];
[$protocol, , $session] = $this->createProtocolWithPendingRequests($pending);

$this->assertNull($protocol->checkResponse(1000, $session->getId()));
$this->assertSame($pending, $protocol->getPendingRequests($session->getId()));
}

#[TestDox('An answer that arrived before the timeout was noticed wins over the timeout')]
public function testCheckResponseReturnsAnswerOfExpiredRequest(): void
{
[$protocol, , $session] = $this->createProtocolWithPendingRequests([
1000 => ['request_id' => 1000, 'timeout' => 5, 'timestamp' => time() - 10],
]);
$session->set('_mcp.responses.1000', ['jsonrpc' => '2.0', 'id' => 1000, 'result' => ['action' => 'accept']]);
$session->save();

$result = $protocol->checkResponse(1000, $session->getId());

$this->assertInstanceOf(Response::class, $result);
$this->assertSame(1000, $result->getId());
$this->assertSame([], $protocol->getPendingRequests($session->getId()));
}

#[TestDox('An answer arriving after the timeout was reported is dropped')]
public function testAnswerAfterReportedTimeoutIsDropped(): void
{
[$protocol, $sessionManager, $session] = $this->createProtocolWithPendingRequests([
1000 => ['request_id' => 1000, 'timeout' => 5, 'timestamp' => time() - 10],
]);

$this->assertInstanceOf(Error::class, $protocol->checkResponse(1000, $session->getId()));

$protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 1000, "result": {}}', $session->getId());

$this->assertNull($sessionManager->createWithId($session->getId())->get('_mcp.responses.1000'));
$this->assertNull($protocol->checkResponse(1000, $session->getId()));
}

#[TestDox('ClientResponseEvent::getId() is null when the error has no id')]
public function testClientResponseEventGetIdCanBeNull(): void
{
Expand Down Expand Up @@ -2399,6 +2495,28 @@ public function testFiberYieldedUnexpectedPayloadIsIgnored(): void
$protocol->handleFiberYield(['type' => 'notification'], Uuid::v4());
}

/**
* @param array<int, array{request_id: int, timeout: int, timestamp: int}> $pending
*
* @return array{Protocol, SessionManager, SessionInterface}
*/
private function createProtocolWithPendingRequests(array $pending): array
{
$sessionManager = new SessionManager(new InMemorySessionStore());
$protocol = new Protocol(
requestHandlers: [],
notificationHandlers: [],
messageFactory: MessageFactory::make(),
sessionManager: $sessionManager,
);

$session = $sessionManager->create();
$session->set('_mcp.pending_requests', $pending);
$session->save();

return [$protocol, $sessionManager, $session];
}

/**
* @param list<int> $requestIds
*/
Expand Down
30 changes: 5 additions & 25 deletions tests/Unit/Server/Transport/StreamableHttpTransportTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
use PHPUnit\Framework\Attributes\DataProvider;
use PHPUnit\Framework\Attributes\TestDox;
use PHPUnit\Framework\TestCase;
use Psr\Clock\ClockInterface;
use Psr\Http\Message\ResponseFactoryInterface;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
Expand Down Expand Up @@ -411,34 +410,15 @@ public function testImmediateResponseIsNotReplayedOnSecondPost(): void
$this->assertSame('', (string) $second->getBody());
}

#[TestDox('the polling loop times out a pending request via the injected clock')]
public function testPollingLoopTimesOutPendingRequestViaInjectedClock(): void
#[TestDox('the polling loop resumes the fiber with the timeout error the response finder reports')]
public function testPollingLoopResumesFiberWithTimeoutErrorFromResponseFinder(): void
{
$request = $this->factory
->createServerRequest('POST', 'http://localhost/')
->withHeader('Host', 'localhost')
->withBody($this->factory->createStream('{"jsonrpc":"2.0","id":1,"method":"ping"}'));

$requestedAt = 1_000_000;

// Frozen 121s after the pending request was issued — past its 120s timeout.
$clock = new class($requestedAt + 121) implements ClockInterface {
public function __construct(private readonly int $timestamp)
{
}

public function now(): \DateTimeImmutable
{
return (new \DateTimeImmutable())->setTimestamp($this->timestamp);
}
};

$transport = new StreamableHttpTransport(
$request,
$this->factory,
$this->factory,
clock: $clock,
);
$transport = new StreamableHttpTransport($request, $this->factory, $this->factory);

$received = null;
$fiber = new \Fiber(static function () use (&$received) {
Expand All @@ -452,9 +432,9 @@ public function now(): \DateTimeImmutable
$transport->attachFiberToSession($fiber, Uuid::v4());
});
$transport->setOutgoingMessagesProvider(static fn (): array => []);
$transport->setResponseFinder(static fn () => null);
$transport->setResponseFinder(static fn (int $requestId): Error => Error::forInternalError('Request timed out', $requestId));
$transport->setPendingRequestsProvider(static fn (): array => [
['request_id' => 1, 'timestamp' => $requestedAt, 'timeout' => 120],
['request_id' => 1, 'timestamp' => 1_000_000, 'timeout' => 120],
]);

$response = $transport->listen();
Expand Down
Loading