diff --git a/docs/client.md b/docs/client.md index 494edf1f..bbc5cf8a 100644 --- a/docs/client.md +++ b/docs/client.md @@ -57,10 +57,32 @@ $client = Client::builder() ->setClientInfo('My Application', '1.0.0', 'Description of my client') ->setInitTimeout(30) // Seconds to wait for initialization ->setRequestTimeout(120) // Seconds to wait for request responses - ->setMaxRetries(3) // Retry attempts for failed connections + ->setMaxRetries(3) // Retries for failed connections ->build(); ``` +### Connection Retries + +`setMaxRetries()` controls how often `connect()` retries a failed connection. It +counts retries rather than attempts, so the default of `3` means one initial +attempt plus up to three retries — four in total — before the `ConnectionException` +of the last attempt is rethrown: + +```php +$client = Client::builder() + ->setMaxRetries(0) // Fail on the first failed attempt + ->build(); +``` + +Between two attempts the transport is closed, so a retry never reuses a +half-established connection: a `StdioTransport` spawns a fresh server process and +an `HttpTransport` discards the session ID of the failed attempt. Each retry is +preceded by a short, linearly growing delay (100ms, 200ms, 300ms, …). + +Only the connection handshake is retried. Individual requests such as +`callTool()` are always sent once — retrying them is unsafe as tool calls are not +necessarily idempotent. + ### Client Information Set the client's identity reported to servers during initialization: diff --git a/src/Client.php b/src/Client.php index dd290698..37bd3b44 100644 --- a/src/Client.php +++ b/src/Client.php @@ -55,6 +55,8 @@ */ class Client { + private const RETRY_BASE_DELAY_MS = 100; + private ?TransportInterface $transport = null; public function __construct( @@ -75,16 +77,44 @@ public static function builder(): Builder /** * Connect to an MCP server using the provided transport. * - * @throws ConnectionException If connection or initialization fails + * A failed attempt is closed and retried, see {@see Builder::setMaxRetries()}. + * + * @throws ConnectionException If connection or initialization fails on every attempt */ public function connect(TransportInterface $transport): void { $this->transport = $transport; $this->protocol->connect($transport, $this->config); - $transport->connect(); + $maxAttempts = $this->config->maxRetries + 1; + + for ($attempt = 1; $attempt <= $maxAttempts; ++$attempt) { + try { + $transport->connect(); + + $this->logger->info('Client connected and initialized', ['attempt' => $attempt]); + + return; + } catch (ConnectionException $e) { + // initialize() flags the session before sending the initialized + // notification, so a failure in between leaves the flag set. + $this->protocol->getState()->setInitialized(false); - $this->logger->info('Client connected and initialized'); + $transport->close(); + + if ($attempt === $maxAttempts) { + throw $e; + } + + $this->logger->warning('Connection attempt failed, retrying', [ + 'attempt' => $attempt, + 'max_attempts' => $maxAttempts, + 'exception' => $e, + ]); + + usleep($attempt * self::RETRY_BASE_DELAY_MS * 1000); + } + } } /** diff --git a/src/Client/Builder.php b/src/Client/Builder.php index 35f7f5de..9a69da17 100644 --- a/src/Client/Builder.php +++ b/src/Client/Builder.php @@ -96,7 +96,11 @@ public function setRequestTimeout(int $seconds): self } /** - * Set maximum retry attempts for failed connections. + * Set the number of times a failed connection attempt is retried. + * + * Counts retries, not attempts: 3 means one initial attempt plus up to three + * retries. Pass 0 to fail on the first failure. Only applies to + * {@see Client::connect()}; individual requests are never retried. */ public function setMaxRetries(int $retries): self { diff --git a/src/Client/Configuration.php b/src/Client/Configuration.php index 8c7e909b..e22c740f 100644 --- a/src/Client/Configuration.php +++ b/src/Client/Configuration.php @@ -11,6 +11,7 @@ namespace Mcp\Client; +use Mcp\Exception\InvalidArgumentException; use Mcp\Schema\ClientCapabilities; use Mcp\Schema\Enum\ProtocolVersion; use Mcp\Schema\Implementation; @@ -30,5 +31,8 @@ public function __construct( public readonly int $requestTimeout = 120, public readonly int $maxRetries = 3, ) { + if ($maxRetries < 0) { + throw new InvalidArgumentException(\sprintf('The maximum number of retries must be zero or greater, got %d.', $maxRetries)); + } } } diff --git a/src/Client/Protocol.php b/src/Client/Protocol.php index 75648456..98c18ae8 100644 --- a/src/Client/Protocol.php +++ b/src/Client/Protocol.php @@ -145,22 +145,29 @@ public function request(Request $request, int $timeout, bool $withProgress = fal } $this->state->addPendingRequest($requestId, $timeout); - $this->sendRequest($request); - $immediate = $this->state->consumeResponse($requestId); - if (null !== $immediate) { - $this->logger->debug('Received immediate response', ['id' => $requestId]); + try { + $this->sendRequest($request); - return $immediate; - } + $immediate = $this->state->consumeResponse($requestId); + if (null !== $immediate) { + $this->logger->debug('Received immediate response', ['id' => $requestId]); - $this->logger->debug('Suspending fiber for response', ['id' => $requestId]); + return $immediate; + } - return \Fiber::suspend([ - 'type' => 'await_response', - 'request_id' => $requestId, - 'timeout' => $timeout, - ]); + $this->logger->debug('Suspending fiber for response', ['id' => $requestId]); + + return \Fiber::suspend([ + 'type' => 'await_response', + 'request_id' => $requestId, + 'timeout' => $timeout, + ]); + } finally { + // Only the response path clears it, so a request that timed out or + // whose send() threw would stay pending and fail every later one. + $this->state->removePendingRequest($requestId); + } } /** diff --git a/tests/Unit/ClientTest.php b/tests/Unit/ClientTest.php new file mode 100644 index 00000000..558de954 --- /dev/null +++ b/tests/Unit/ClientTest.php @@ -0,0 +1,283 @@ +assertInstanceOf(Builder::class, Client::builder()); + } + + #[TestDox('connect() succeeds on the first attempt without retrying')] + public function testConnectSucceedsWithoutRetrying(): void + { + $transport = new FakeTransport([FakeTransport::ACCEPT]); + + $client = Client::builder()->build(); + $client->connect($transport); + + $this->assertSame(1, $transport->connectCalls); + $this->assertSame(0, $transport->closeCalls); + $this->assertTrue($client->isConnected()); + } + + #[TestDox('connect() retries a failed attempt and succeeds on a later one')] + public function testConnectRetriesUntilItSucceeds(): void + { + $transport = new FakeTransport([FakeTransport::REJECT, FakeTransport::REJECT, FakeTransport::ACCEPT]); + + $client = Client::builder()->setMaxRetries(3)->build(); + $client->connect($transport); + + $this->assertSame(3, $transport->connectCalls); + $this->assertTrue($client->isConnected()); + $this->assertSame('Test Server', $client->getServerInfo()?->name); + } + + #[TestDox('connect() closes the transport between two attempts')] + public function testConnectClosesTransportBetweenAttempts(): void + { + $transport = new FakeTransport([FakeTransport::REJECT, FakeTransport::ACCEPT]); + + $client = Client::builder()->setMaxRetries(1)->build(); + $client->connect($transport); + + $this->assertSame(2, $transport->connectCalls); + $this->assertSame(1, $transport->closeCalls, 'the failed attempt must be cleaned up before the retry'); + } + + #[TestDox('connect() rethrows once the retries are exhausted')] + public function testConnectThrowsWhenRetriesAreExhausted(): void + { + $transport = new FakeTransport([FakeTransport::REJECT, FakeTransport::REJECT, FakeTransport::REJECT]); + + $client = Client::builder()->setMaxRetries(2)->build(); + + try { + $client->connect($transport); + $this->fail(\sprintf('Expected a "%s" to be thrown.', ConnectionException::class)); + } catch (ConnectionException) { + // Expected. + } + + $this->assertSame(3, $transport->connectCalls, 'one initial attempt plus two retries'); + $this->assertSame(3, $transport->closeCalls, 'the last attempt must be cleaned up as well'); + $this->assertFalse($client->isConnected()); + } + + #[TestDox('setMaxRetries(0) disables retrying')] + public function testZeroRetriesAttemptsToConnectOnce(): void + { + $transport = new FakeTransport([FakeTransport::REJECT, FakeTransport::ACCEPT]); + + $client = Client::builder()->setMaxRetries(0)->build(); + + $this->expectException(ConnectionException::class); + + try { + $client->connect($transport); + } finally { + $this->assertSame(1, $transport->connectCalls); + } + } + + #[TestDox('a negative retry count is rejected')] + public function testNegativeRetryCountIsRejected(): void + { + $builder = Client::builder()->setMaxRetries(-1); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('The maximum number of retries must be zero or greater, got -1.'); + + $builder->build(); + } + + #[TestDox('a connection breaking after the handshake result does not leave the client connected')] + public function testFailureAfterHandshakeResultDoesNotLeaveClientConnected(): void + { + $transport = new FakeTransport([FakeTransport::BREAK_AFTER_ACCEPT]); + + $client = Client::builder()->setMaxRetries(0)->build(); + + try { + $client->connect($transport); + $this->fail(\sprintf('Expected a "%s" to be thrown.', ConnectionException::class)); + } catch (ConnectionException) { + // Expected. + } + + $this->assertFalse($client->isConnected()); + } + + #[TestDox('a timed out attempt does not leave state behind that fails the retry')] + public function testTimedOutAttemptDoesNotPoisonTheRetry(): void + { + // The shortest allowed init timeout, so this waits out about a second. + $transport = new FakeTransport([FakeTransport::IGNORE, FakeTransport::ACCEPT]); + + $client = Client::builder()->setInitTimeout(1)->setMaxRetries(1)->build(); + $client->connect($transport); + + $this->assertSame(2, $transport->connectCalls); + $this->assertTrue($client->isConnected()); + } +} + +/** + * Transport whose connection attempts succeed or fail on command. + * + * Requests are answered from the polling loop rather than from send(), the way + * a stdio transport does, with each connect() call answering according to the + * next configured outcome. + * + * @phpstan-import-type McpFiber from TransportInterface + */ +final class FakeTransport extends BaseTransport +{ + /** The initialize request is answered with a result. */ + public const ACCEPT = 'accept'; + + /** The initialize request is answered with a JSON-RPC error. */ + public const REJECT = 'reject'; + + /** The initialize request is not answered at all and has to time out. */ + public const IGNORE = 'ignore'; + + /** The initialize request is answered, but the connection breaks right after. */ + public const BREAK_AFTER_ACCEPT = 'break_after_accept'; + + public int $connectCalls = 0; + public int $closeCalls = 0; + + private string $outcome = self::ACCEPT; + + /** @var list Answers not yet delivered to the client */ + private array $outbox = []; + + /** + * @param list $attempts How each successive connect() call behaves + */ + public function __construct(private array $attempts = [self::ACCEPT]) + { + parent::__construct(); + } + + public function connect(): void + { + ++$this->connectCalls; + $this->outcome = array_shift($this->attempts) ?? self::ACCEPT; + + $result = $this->runRequest(new \Fiber(fn () => $this->handleInitialize())); + + if ($result instanceof Error) { + throw new ConnectionException('Initialization failed: '.$result->message); + } + } + + public function send(string $data): void + { + if (self::IGNORE === $this->outcome) { + return; + } + + $message = json_decode($data, true, 512, \JSON_THROW_ON_ERROR); + + if (!isset($message['id'])) { + if (self::BREAK_AFTER_ACCEPT === $this->outcome) { + throw new ConnectionException('Connection lost'); + } + + return; // A notification, nothing to answer. + } + + $answer = self::REJECT === $this->outcome + ? ['error' => ['code' => Error::INTERNAL_ERROR, 'message' => 'Server unavailable']] + : ['result' => [ + 'protocolVersion' => ProtocolVersion::V2025_11_25->value, + 'capabilities' => [], + 'serverInfo' => ['name' => 'Test Server', 'version' => '1.0.0'], + ]]; + + $this->outbox[] = json_encode(['jsonrpc' => '2.0', 'id' => $message['id']] + $answer, \JSON_THROW_ON_ERROR); + } + + public function runRequest(\Fiber $fiber, ?callable $onProgress = null): Response|Error + { + $fiber->start(); + + // Polls at the same 1ms interval as the real transports, so a request + // left unanswered runs into its timeout in real time. + for ($poll = 0; !$fiber->isTerminated(); ++$poll) { + if ($poll > 5000) { + throw new \LogicException('The fiber never terminated, no pending request became resolvable.'); + } + + foreach ($this->outbox as $answer) { + $this->handleMessage($answer); + } + $this->outbox = []; + + $this->resumeFiber($fiber); + + usleep(1000); + } + + return $fiber->getReturn(); + } + + public function close(): void + { + ++$this->closeCalls; + + $this->handleClose('Transport closed'); + } + + /** + * @param McpFiber $fiber + */ + private function resumeFiber(\Fiber $fiber): void + { + if (!$fiber->isSuspended() || null === $this->state) { + return; + } + + foreach ($this->state->getPendingRequests() as $pending) { + $response = $this->state->consumeResponse($pending['request_id']); + + if (null !== $response) { + $fiber->resume($response); + + return; + } + + if (time() - $pending['timestamp'] >= $pending['timeout']) { + $fiber->resume(Error::forInternalError('Request timed out', $pending['request_id'])); + + return; + } + } + } +}