Namespaces

Classes

1: <?php
2:
3: declare(strict_types=1);
4:
5: /**
6: * This file is part of the Nexus MCP SDK package.
7: *
8: * (c) 2026 John Paul E. Balandan, CPA <paulbalandan@gmail.com>
9: *
10: * For the full copyright and license information, please view
11: * the LICENSE file that was distributed with this source code.
12: */
13:
14: namespace Nexus\Mcp\Client;
15:
16: use Amp\CancelledException;
17: use Amp\DeferredFuture;
18: use Nexus\Mcp\Client\Dispatch\DiscoveredServerCapabilities;
19: use Nexus\Mcp\Client\Dispatch\ProgressListenerRegistry;
20: use Nexus\Mcp\Client\Dispatch\RequestDeadline;
21: use Nexus\Mcp\Client\Exception\ServerCapabilityNotSupportedException;
22: use Nexus\Mcp\Client\Subscription\OpenSubscription;
23: use Nexus\Mcp\Client\Subscription\SubscriptionRegistry;
24: use Nexus\Mcp\Client\Subscription\SubscriptionStream;
25: use Nexus\Mcp\Core\Dispatch\MessageDispatcherInterface;
26: use Nexus\Mcp\Core\Dispatch\PendingOutboundRequests;
27: use Nexus\Mcp\Core\Exception\LogicException;
28: use Nexus\Mcp\Core\Exception\OutboundRequestFailedException;
29: use Nexus\Mcp\Core\Exception\RemoteCallFailedException;
30: use Nexus\Mcp\Core\Exception\RequestTimeoutException;
31: use Nexus\Mcp\Core\Exception\SupervisionExhaustedException;
32: use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
33: use Nexus\Mcp\Core\Http\ParameterHeaderBinding;
34: use Nexus\Mcp\Core\Http\ParameterHeaders;
35: use Nexus\Mcp\Core\Http\ParameterHeaderScanner;
36: use Nexus\Mcp\Core\SafeDisplay;
37: use Nexus\Mcp\Core\Schema\ClientCapabilities;
38: use Nexus\Mcp\Core\Schema\Cursor;
39: use Nexus\Mcp\Core\Schema\Enum\ProtocolErrorCode;
40: use Nexus\Mcp\Core\Schema\Error\UnsupportedProtocolVersionError;
41: use Nexus\Mcp\Core\Schema\Implementation;
42: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
43: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcRequest;
44: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResultResponse;
45: use Nexus\Mcp\Core\Schema\MetaObject\RequestMetaObject;
46: use Nexus\Mcp\Core\Schema\Notification\CancelledNotification;
47: use Nexus\Mcp\Core\Schema\NotificationParams\CancelledNotificationParams;
48: use Nexus\Mcp\Core\Schema\ProgressToken;
49: use Nexus\Mcp\Core\Schema\Prompt\PromptReference;
50: use Nexus\Mcp\Core\Schema\ProtocolVersion;
51: use Nexus\Mcp\Core\Schema\Request\CallToolRequest;
52: use Nexus\Mcp\Core\Schema\Request\CompleteRequest;
53: use Nexus\Mcp\Core\Schema\Request\DiscoverRequest;
54: use Nexus\Mcp\Core\Schema\Request\GetPromptRequest;
55: use Nexus\Mcp\Core\Schema\Request\ListPromptsRequest;
56: use Nexus\Mcp\Core\Schema\Request\ListResourcesRequest;
57: use Nexus\Mcp\Core\Schema\Request\ListResourceTemplatesRequest;
58: use Nexus\Mcp\Core\Schema\Request\ListToolsRequest;
59: use Nexus\Mcp\Core\Schema\Request\ReadResourceRequest;
60: use Nexus\Mcp\Core\Schema\Request\SubscriptionsListenRequest;
61: use Nexus\Mcp\Core\Schema\RequestId;
62: use Nexus\Mcp\Core\Schema\RequestParams;
63: use Nexus\Mcp\Core\Schema\RequestParams\CallToolRequestParams;
64: use Nexus\Mcp\Core\Schema\RequestParams\CompleteRequestParams;
65: use Nexus\Mcp\Core\Schema\RequestParams\EmptyRequestParams;
66: use Nexus\Mcp\Core\Schema\RequestParams\GetPromptRequestParams;
67: use Nexus\Mcp\Core\Schema\RequestParams\InputResponseCarrierInterface;
68: use Nexus\Mcp\Core\Schema\RequestParams\PaginatedRequestParams;
69: use Nexus\Mcp\Core\Schema\RequestParams\ReadResourceRequestParams;
70: use Nexus\Mcp\Core\Schema\RequestParams\SubscriptionsListenRequestParams;
71: use Nexus\Mcp\Core\Schema\RequestParamsInterface;
72: use Nexus\Mcp\Core\Schema\Resource\ResourceTemplateReference;
73: use Nexus\Mcp\Core\Schema\Result\CallToolResult;
74: use Nexus\Mcp\Core\Schema\Result\CompleteResult;
75: use Nexus\Mcp\Core\Schema\Result\DiscoverResult;
76: use Nexus\Mcp\Core\Schema\Result\GetPromptResult;
77: use Nexus\Mcp\Core\Schema\Result\InputRequiredResult;
78: use Nexus\Mcp\Core\Schema\Result\InputResponse;
79: use Nexus\Mcp\Core\Schema\Result\ListPromptsResult;
80: use Nexus\Mcp\Core\Schema\Result\ListResourcesResult;
81: use Nexus\Mcp\Core\Schema\Result\ListResourceTemplatesResult;
82: use Nexus\Mcp\Core\Schema\Result\ListToolsResult;
83: use Nexus\Mcp\Core\Schema\Result\ReadResourceResult;
84: use Nexus\Mcp\Core\Schema\Result\SubscriptionsListenResult;
85: use Nexus\Mcp\Core\Schema\ResultResponse\CallToolResultResponse;
86: use Nexus\Mcp\Core\Schema\ResultResponse\CompleteResultResponse;
87: use Nexus\Mcp\Core\Schema\ResultResponse\DiscoverResultResponse;
88: use Nexus\Mcp\Core\Schema\ResultResponse\GetPromptResultResponse;
89: use Nexus\Mcp\Core\Schema\ResultResponse\ListPromptsResultResponse;
90: use Nexus\Mcp\Core\Schema\ResultResponse\ListResourcesResultResponse;
91: use Nexus\Mcp\Core\Schema\ResultResponse\ListResourceTemplatesResultResponse;
92: use Nexus\Mcp\Core\Schema\ResultResponse\ListToolsResultResponse;
93: use Nexus\Mcp\Core\Schema\ResultResponse\ReadResourceResultResponse;
94: use Nexus\Mcp\Core\Schema\ResultResponse\SubscriptionsListenResultResponse;
95: use Nexus\Mcp\Core\Schema\ServerCapabilities;
96: use Nexus\Mcp\Core\Schema\SubscriptionFilter;
97: use Nexus\Mcp\Core\Transport\AbortableTransportInterface;
98: use Nexus\Mcp\Core\Transport\ListenerHandleInterface;
99: use Nexus\Mcp\Core\Transport\ParameterHeaderMirroringInterface;
100: use Nexus\Mcp\Core\Transport\ReceiveContext;
101: use Nexus\Mcp\Core\Transport\ReconnectingTransportInterface;
102: use Nexus\Mcp\Core\Transport\SendContext;
103: use Nexus\Mcp\Core\Transport\TransportInterface;
104: use Psr\Log\LoggerInterface;
105: use Psr\Log\NullLogger;
106: use Revolt\EventLoop;
107:
108: /**
109: * Client-side entry point for the typed JSON-RPC operations issued against an MCP server.
110: */
111: final class Client
112: {
113: /**
114: * Seconds a request may go unanswered before it is abandoned.
115: */
116: public const float DEFAULT_REQUEST_TIMEOUT = 60.0;
117:
118: /**
119: * Seconds a request may run in total, however much progress arrives.
120: */
121: public const float DEFAULT_MAX_REQUEST_TIMEOUT = 600.0;
122:
123: private const int MAX_HEADER_REFRESH_PAGES = 100;
124: private const array LIFECYCLE_META_KEYS = [
125: RequestMetaObject::PROTOCOL_VERSION_KEY => true,
126: RequestMetaObject::CLIENT_INFO_KEY => true,
127: RequestMetaObject::CLIENT_CAPABILITIES_KEY => true,
128: RequestMetaObject::LOG_LEVEL_KEY => true,
129: 'progressToken' => true,
130: ];
131: private const array RETRYABLE_REQUESTS = [
132: CompleteRequest::class,
133: DiscoverRequest::class,
134: GetPromptRequest::class,
135: ListPromptsRequest::class,
136: ListResourcesRequest::class,
137: ListResourceTemplatesRequest::class,
138: ListToolsRequest::class,
139: ReadResourceRequest::class,
140: ];
141:
142: private ?TransportInterface $transport = null;
143: private ?Implementation $serverInfo = null;
144:
145: /**
146: * @var array<string, list<ParameterHeaderBinding>>
147: */
148: private array $toolHeaderBindings = [];
149:
150: /**
151: * @var list<ListenerHandleInterface>
152: */
153: private array $listeners = [];
154:
155: private readonly ParameterHeaders $parameterHeaders;
156:
157: /**
158: * @param \Closure(): (int|non-empty-string) $requestIdFactory
159: * @param \Closure(): (int|non-empty-string) $progressTokenFactory
160: * @param array<non-empty-string, non-empty-string> $extensionMethods
161: * @param null|\Closure(): array<non-empty-string, mixed> $metaExtrasFactory
162: */
163: public function __construct(
164: private readonly Implementation $clientInfo,
165: private readonly ClientCapabilities $clientCapabilities,
166: private readonly MessageDispatcherInterface $dispatcher,
167: private readonly PendingOutboundRequests $outboundRequests,
168: private readonly \Closure $requestIdFactory,
169: private readonly \Closure $progressTokenFactory,
170: private readonly ProtocolVersion $protocolVersion = new ProtocolVersion(version: ProtocolVersion::LATEST_VERSION),
171: private readonly ProgressListenerRegistry $progressListeners = new ProgressListenerRegistry(),
172: private readonly SubscriptionRegistry $subscriptions = new SubscriptionRegistry(),
173: private readonly LoggerInterface $logger = new NullLogger(),
174: private readonly ?float $requestTimeout = self::DEFAULT_REQUEST_TIMEOUT,
175: private readonly ?float $maxRequestTimeout = self::DEFAULT_MAX_REQUEST_TIMEOUT,
176: private readonly bool $retryLostRequests = false,
177: private readonly array $extensionMethods = [],
178: private readonly DiscoveredServerCapabilities $serverCapabilities = new DiscoveredServerCapabilities(),
179: private readonly ?\Closure $metaExtrasFactory = null,
180: ) {
181: $this->parameterHeaders = new ParameterHeaders();
182: }
183:
184: /**
185: * Non-blocking connect to the transport.
186: *
187: * @throws LogicException
188: */
189: public function connect(TransportInterface $transport): void
190: {
191: if (null !== $this->transport) {
192: throw new LogicException('Client is already connected to a transport.');
193: }
194:
195: $this->logger->info('Starting MCP client.');
196: $this->transport = $transport;
197:
198: $this->listeners[] = $transport->onMessage(function (array $envelope, ReceiveContext $context) use ($transport): void {
199: $this->dispatcher->dispatch($envelope, $transport, $context);
200: });
201: $this->listeners[] = $transport->onError(function (\Throwable $e): void {
202: if ($e instanceof OutboundRequestFailedException) {
203: $this->outboundRequests->reject($e->requestId, $e);
204: }
205:
206: if ($e instanceof SupervisionExhaustedException) {
207: $this->failSubscriptions($e);
208: $this->outboundRequests->cancelAll($e);
209: }
210:
211: $this->logger->error('Transport error.', ['exception' => $e]);
212: });
213: $this->listeners[] = $transport->onDrain(function (): void {
214: $this->dispatcher->flushPending();
215: });
216: $this->listeners[] = $transport->onClose(function () use ($transport): void {
217: $error = new TransportAlreadyClosedException(operation: 'await-response');
218: $this->outboundRequests->cancelUnretained($error);
219:
220: EventLoop::queue(function () use ($transport, $error): void {
221: if ($transport !== $this->transport) {
222: return;
223: }
224:
225: if ($transport instanceof ReconnectingTransportInterface && $transport->isReconnecting()) {
226: return;
227: }
228:
229: $this->outboundRequests->cancelAll($error);
230: });
231: });
232:
233: if ($transport instanceof ReconnectingTransportInterface) {
234: $this->listeners[] = $transport->onReconnect(function () use ($transport): void {
235: foreach ($this->outboundRequests->collectRetained() as $retained) {
236: $request = $retained['request'];
237:
238: try {
239: $transport->send($request, $retained['context']);
240: } catch (\Throwable $e) {
241: if ($transport->isReconnecting()) {
242: $this->logger->warning(
243: 'Could not send request {id} again to the replacement peer.',
244: ['id' => $request->id->id, 'exception' => $e],
245: );
246:
247: continue;
248: }
249:
250: $this->outboundRequests->reject($request->id, $e);
251: }
252: }
253:
254: foreach ($this->subscriptions->all() as $subscription) {
255: try {
256: $this->openStream($subscription, $transport);
257: } catch (\Throwable $e) {
258: $this->logger->error(
259: 'Could not re-open subscription {id} against the replacement peer.',
260: ['id' => $subscription->subscriptionId->id, 'exception' => $e],
261: );
262: }
263: }
264: });
265: }
266:
267: $transport->start();
268: }
269:
270: /**
271: * Closes the transport and detaches it, doing nothing when not connected.
272: */
273: public function disconnect(): void
274: {
275: $transport = $this->transport;
276: $this->transport = null;
277:
278: $this->toolHeaderBindings = [];
279: $this->serverInfo = null;
280: $this->serverCapabilities->record(null);
281:
282: $error = new TransportAlreadyClosedException(operation: 'await-response');
283: $this->failSubscriptions($error);
284: $this->outboundRequests->cancelAll($error);
285:
286: $listeners = $this->listeners;
287: $this->listeners = [];
288:
289: try {
290: $transport?->close();
291: } finally {
292: foreach ($listeners as $listener) {
293: $listener->dispose();
294: }
295: }
296: }
297:
298: public function getServerInfo(): ?Implementation
299: {
300: return $this->serverInfo;
301: }
302:
303: public function getServerCapabilities(): ?ServerCapabilities
304: {
305: return $this->serverCapabilities->current();
306: }
307:
308: /**
309: * Sends `server/discover` and records the advertised server info and capabilities.
310: *
311: * @throws LogicException
312: * @throws RequestTimeoutException
313: * @throws ServerCapabilityNotSupportedException
314: * @throws TransportAlreadyClosedException
315: */
316: public function discover(): DiscoverResult
317: {
318: $result = $this->sendRequest(
319: new DiscoverRequest(id: $this->mintRequestId(), params: new EmptyRequestParams(meta: $this->stampMeta())),
320: DiscoverResultResponse::class,
321: )->result;
322:
323: $this->serverInfo = $result->meta->serverInfo;
324: $this->serverCapabilities->record($result->capabilities);
325:
326: return $result;
327: }
328:
329: /**
330: * Opens a `subscriptions/listen` stream, returning as soon as the request is away. The send skips
331: * `dispatch()`'s protocol-version renegotiation, which cannot trigger while one revision is supported.
332: *
333: * @param \Closure(JsonRpcNotification<non-empty-string>): void $onNotification
334: *
335: * @throws LogicException
336: * @throws TransportAlreadyClosedException
337: */
338: public function listen(SubscriptionFilter $notifications, \Closure $onNotification): SubscriptionStream
339: {
340: $transport = $this->requireConnectedTransport();
341:
342: $id = $this->mintRequestId();
343:
344: /** @var DeferredFuture<SubscriptionsListenResult> $outcome */
345: $outcome = new DeferredFuture();
346:
347: $future = $outcome->getFuture();
348: $future->ignore();
349:
350: $subscription = new OpenSubscription($id, $notifications, $onNotification, $outcome);
351:
352: $this->openStream($subscription, $transport);
353: $this->subscriptions->register($subscription);
354:
355: return new SubscriptionStream($id, $future, function () use ($id, $transport): void {
356: $this->subscriptions->forget($id);
357:
358: if (! $this->outboundRequests->forget($id)) {
359: return;
360: }
361:
362: $this->abortExchange($transport, $id);
363:
364: try {
365: $transport->send(new CancelledNotification(
366: params: new CancelledNotificationParams(requestId: $id, reason: 'The subscription was closed.'),
367: ));
368: } catch (\Throwable $e) {
369: $this->logger->debug(
370: 'Could not tell the server that subscription {id} was closed.',
371: ['id' => $id->id, 'exception' => $e],
372: );
373: }
374: });
375: }
376:
377: /**
378: * @throws LogicException
379: * @throws RequestTimeoutException
380: * @throws ServerCapabilityNotSupportedException
381: * @throws TransportAlreadyClosedException
382: */
383: public function listTools(?Cursor $cursor = null): ListToolsResult
384: {
385: $result = $this->sendRequest(
386: new ListToolsRequest(
387: id: $this->mintRequestId(),
388: params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
389: ),
390: ListToolsResultResponse::class,
391: )->result;
392:
393: return $this->transport instanceof ParameterHeaderMirroringInterface
394: ? $this->admitMirrorableTools($result)
395: : $result;
396: }
397:
398: /**
399: * @throws LogicException
400: * @throws RequestTimeoutException
401: * @throws ServerCapabilityNotSupportedException
402: * @throws TransportAlreadyClosedException
403: */
404: public function listResources(?Cursor $cursor = null): ListResourcesResult
405: {
406: return $this->sendRequest(
407: new ListResourcesRequest(
408: id: $this->mintRequestId(),
409: params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
410: ),
411: ListResourcesResultResponse::class,
412: )->result;
413: }
414:
415: /**
416: * @throws LogicException
417: * @throws RequestTimeoutException
418: * @throws ServerCapabilityNotSupportedException
419: * @throws TransportAlreadyClosedException
420: */
421: public function listResourceTemplates(?Cursor $cursor = null): ListResourceTemplatesResult
422: {
423: return $this->sendRequest(
424: new ListResourceTemplatesRequest(
425: id: $this->mintRequestId(),
426: params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
427: ),
428: ListResourceTemplatesResultResponse::class,
429: )->result;
430: }
431:
432: /**
433: * @throws LogicException
434: * @throws RequestTimeoutException
435: * @throws ServerCapabilityNotSupportedException
436: * @throws TransportAlreadyClosedException
437: */
438: public function listPrompts(?Cursor $cursor = null): ListPromptsResult
439: {
440: return $this->sendRequest(
441: new ListPromptsRequest(
442: id: $this->mintRequestId(),
443: params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
444: ),
445: ListPromptsResultResponse::class,
446: )->result;
447: }
448:
449: /**
450: * @param non-empty-string $uri
451: * @param null|array<int|non-empty-string, InputResponse> $inputResponses
452: *
453: * @throws LogicException
454: * @throws RequestTimeoutException
455: * @throws ServerCapabilityNotSupportedException
456: * @throws TransportAlreadyClosedException
457: */
458: public function readResource(
459: string $uri,
460: ?array $inputResponses = null,
461: ?string $requestState = null,
462: ): InputRequiredResult|ReadResourceResult {
463: return $this->sendRequest(
464: new ReadResourceRequest(
465: id: $this->mintRequestId(),
466: params: new ReadResourceRequestParams(
467: uri: $uri,
468: meta: $this->stampMeta(),
469: inputResponses: $inputResponses,
470: requestState: $requestState,
471: ),
472: ),
473: ReadResourceResultResponse::class,
474: )->result;
475: }
476:
477: /**
478: * @param non-empty-string $name
479: * @param null|array<array-key, string> $arguments
480: * @param null|array<int|non-empty-string, InputResponse> $inputResponses
481: *
482: * @throws LogicException
483: * @throws RequestTimeoutException
484: * @throws ServerCapabilityNotSupportedException
485: * @throws TransportAlreadyClosedException
486: */
487: public function getPrompt(
488: string $name,
489: ?array $arguments = null,
490: ?array $inputResponses = null,
491: ?string $requestState = null,
492: ): GetPromptResult|InputRequiredResult {
493: return $this->sendRequest(
494: new GetPromptRequest(
495: id: $this->mintRequestId(),
496: params: new GetPromptRequestParams(
497: name: $name,
498: meta: $this->stampMeta(),
499: arguments: $arguments,
500: inputResponses: $inputResponses,
501: requestState: $requestState,
502: ),
503: ),
504: GetPromptResultResponse::class,
505: )->result;
506: }
507:
508: /**
509: * @param array{name: string, value: string} $argument
510: * @param null|array{arguments?: array<array-key, string>} $context
511: *
512: * @throws LogicException
513: * @throws RequestTimeoutException
514: * @throws ServerCapabilityNotSupportedException
515: * @throws TransportAlreadyClosedException
516: */
517: public function complete(
518: PromptReference|ResourceTemplateReference $ref,
519: array $argument,
520: ?array $context = null,
521: ): CompleteResult {
522: return $this->sendRequest(
523: new CompleteRequest(
524: id: $this->mintRequestId(),
525: params: new CompleteRequestParams(
526: ref: $ref,
527: argument: $argument,
528: meta: $this->stampMeta(),
529: context: $context,
530: ),
531: ),
532: CompleteResultResponse::class,
533: )->result;
534: }
535:
536: /**
537: * Invokes a tool, answering with an `InputRequiredResult` whose `requestState` must be echoed back
538: * unchanged when the server needs more input.
539: *
540: * @param non-empty-string $name
541: * @param null|array<array-key, mixed> $arguments
542: * @param null|\Closure(float $progress, ?float $total, ?string $message): void $onProgress
543: * @param null|array<int|non-empty-string, InputResponse> $inputResponses
544: *
545: * @throws LogicException
546: * @throws RequestTimeoutException
547: * @throws ServerCapabilityNotSupportedException
548: * @throws TransportAlreadyClosedException
549: */
550: public function callTool(
551: string $name,
552: ?array $arguments = null,
553: ?\Closure $onProgress = null,
554: ?array $inputResponses = null,
555: ?string $requestState = null,
556: ): CallToolResult|InputRequiredResult {
557: try {
558: return $this->attemptToolCall($name, $arguments, $onProgress, $inputResponses, $requestState);
559: } catch (RemoteCallFailedException $e) {
560: if ($e->getCode() !== ProtocolErrorCode::HeaderMismatch->value) {
561: throw $e;
562: }
563: }
564:
565: try {
566: $this->refreshToolHeaderBindings($name);
567: } catch (\Throwable $refreshFailure) {
568: throw new RemoteCallFailedException($e->error, $refreshFailure);
569: }
570:
571: return $this->attemptToolCall($name, $arguments, $onProgress, $inputResponses, $requestState);
572: }
573:
574: /**
575: * @template TResponse of JsonRpcResultResponse = JsonRpcResultResponse
576: *
577: * @param JsonRpcRequest<non-empty-string> $request
578: * @param class-string<TResponse> $response
579: *
580: * @return TResponse
581: *
582: * @throws LogicException
583: * @throws RequestTimeoutException
584: * @throws ServerCapabilityNotSupportedException
585: * @throws TransportAlreadyClosedException
586: */
587: public function sendRequest(
588: JsonRpcRequest $request,
589: string $response,
590: ?SendContext $context = null,
591: ?float $timeout = null,
592: ): JsonRpcResultResponse {
593: return $this->dispatch($request, $response, $context, $this->openDeadline($timeout));
594: }
595:
596: public function stampMeta(?ProgressToken $progressToken = null): RequestMetaObject
597: {
598: return new RequestMetaObject(
599: protocolVersion: $this->protocolVersion,
600: clientInfo: $this->clientInfo,
601: clientCapabilities: $this->clientCapabilities,
602: progressToken: $progressToken,
603: extras: null === $this->metaExtrasFactory ? [] : array_diff_key(($this->metaExtrasFactory)(), self::LIFECYCLE_META_KEYS),
604: );
605: }
606:
607: public function mintRequestId(): RequestId
608: {
609: return new RequestId(id: ($this->requestIdFactory)());
610: }
611:
612: /**
613: * @throws TransportAlreadyClosedException
614: */
615: private function openStream(OpenSubscription $subscription, TransportInterface $transport): void
616: {
617: $id = $subscription->subscriptionId;
618: $response = $this->outboundRequests->register($id, SubscriptionsListenResultResponse::class);
619:
620: $response
621: ->map(function (SubscriptionsListenResultResponse $response) use ($id): void {
622: $this->subscriptions->forget($id)?->outcome->complete($response->result);
623: })
624: ->catch(function (\Throwable $e) use ($id, $transport): void {
625: if ($transport instanceof ReconnectingTransportInterface && $transport->isReconnecting()) {
626: return;
627: }
628:
629: $this->subscriptions->forget($id)?->outcome->error($e);
630: })
631: ->ignore()
632: ;
633:
634: try {
635: $transport->send(new SubscriptionsListenRequest(
636: id: $id,
637: params: new SubscriptionsListenRequestParams(
638: notifications: $subscription->notifications,
639: meta: $this->stampMeta(),
640: ),
641: ));
642: } catch (\Throwable $e) {
643: $this->outboundRequests->forget($id);
644:
645: throw $e;
646: }
647: }
648:
649: /**
650: * @param JsonRpcRequest<non-empty-string> $request
651: */
652: private function retainsAcrossRestart(JsonRpcRequest $request): bool
653: {
654: if (! $this->retryLostRequests) {
655: return false;
656: }
657:
658: $method = $request::getMethod();
659:
660: foreach (self::RETRYABLE_REQUESTS as $retryable) {
661: if ($retryable::getMethod() === $method) {
662: return ! $this->resumesAnEarlierRound($request->params);
663: }
664: }
665:
666: return false;
667: }
668:
669: private function resumesAnEarlierRound(?RequestParamsInterface $params): bool
670: {
671: if (! $params instanceof InputResponseCarrierInterface) {
672: return false;
673: }
674:
675: return $params->getInputResponses() !== null || $params->getRequestState() !== null;
676: }
677:
678: private function abortExchange(TransportInterface $transport, RequestId $id): void
679: {
680: if ($transport instanceof AbortableTransportInterface) {
681: $transport->abort($id);
682: }
683: }
684:
685: private function failSubscriptions(\Throwable $reason): void
686: {
687: foreach ($this->subscriptions->drain() as $subscription) {
688: $subscription->outcome->error($reason);
689: }
690: }
691:
692: private function refreshToolHeaderBindings(string $name): void
693: {
694: if (! $this->transport instanceof ParameterHeaderMirroringInterface) {
695: return;
696: }
697:
698: $cursor = null;
699: $seen = [];
700:
701: for ($page = 1; $page <= self::MAX_HEADER_REFRESH_PAGES; ++$page) {
702: $result = $this->listTools($cursor);
703:
704: foreach ($result->tools as $tool) {
705: if ($tool->name === $name) {
706: return;
707: }
708: }
709:
710: $cursor = $result->nextCursor;
711:
712: if (null === $cursor) {
713: return;
714: }
715:
716: if (isset($seen[$cursor->cursor])) {
717: unset($this->toolHeaderBindings[$name]);
718: $this->logger->warning(
719: 'Server sent cursor {cursor} again while re-listing tool {tool}, first seen on page {page}, so the refresh stopped.',
720: ['cursor' => SafeDisplay::sanitise($cursor->cursor), 'tool' => $name, 'page' => $seen[$cursor->cursor]],
721: );
722:
723: return;
724: }
725:
726: $seen[$cursor->cursor] = $page;
727: }
728:
729: unset($this->toolHeaderBindings[$name]);
730: $this->logger->warning(
731: 'Re-listing tool {tool} passed {pages} pages without reaching it, so the refresh stopped.',
732: ['tool' => $name, 'pages' => self::MAX_HEADER_REFRESH_PAGES],
733: );
734: }
735:
736: /**
737: * @param non-empty-string $name
738: * @param null|array<array-key, mixed> $arguments
739: * @param null|\Closure(float $progress, ?float $total, ?string $message): void $onProgress
740: * @param null|array<int|non-empty-string, InputResponse> $inputResponses
741: */
742: private function attemptToolCall(
743: string $name,
744: ?array $arguments,
745: ?\Closure $onProgress,
746: ?array $inputResponses = null,
747: ?string $requestState = null,
748: ): CallToolResult|InputRequiredResult {
749: $context = new SendContext(headers: $this->mirrorParameterHeaders($name, $arguments));
750:
751: if (null === $onProgress) {
752: return $this->sendRequest(
753: new CallToolRequest(
754: id: $this->mintRequestId(),
755: params: new CallToolRequestParams(
756: name: $name,
757: meta: $this->stampMeta(),
758: arguments: $arguments,
759: inputResponses: $inputResponses,
760: requestState: $requestState,
761: ),
762: ),
763: CallToolResultResponse::class,
764: $context,
765: )->result;
766: }
767:
768: $progressToken = $this->mintProgressToken();
769: $deadline = $this->openDeadline();
770:
771: try {
772: $this->progressListeners->register(
773: $progressToken,
774: static function (float $progress, ?float $total, ?string $message) use ($onProgress, $deadline): void {
775: $deadline?->extend();
776: $onProgress($progress, $total, $message);
777: },
778: );
779:
780: return $this->dispatch(
781: new CallToolRequest(
782: id: $this->mintRequestId(),
783: params: new CallToolRequestParams(
784: name: $name,
785: meta: $this->stampMeta($progressToken),
786: arguments: $arguments,
787: inputResponses: $inputResponses,
788: requestState: $requestState,
789: ),
790: ),
791: CallToolResultResponse::class,
792: $context,
793: $deadline,
794: )->result;
795: } finally {
796: $deadline?->release();
797: $this->progressListeners->unregister($progressToken);
798: }
799: }
800:
801: /**
802: * @template TResponse of JsonRpcResultResponse
803: *
804: * @param JsonRpcRequest<non-empty-string> $request
805: * @param class-string<TResponse> $response
806: *
807: * @return TResponse
808: *
809: * @throws LogicException
810: * @throws RequestTimeoutException
811: * @throws ServerCapabilityNotSupportedException
812: */
813: private function dispatch(
814: JsonRpcRequest $request,
815: string $response,
816: ?SendContext $context,
817: ?RequestDeadline $deadline,
818: ): JsonRpcResultResponse {
819: try {
820: try {
821: return $this->exchange($request, $response, $context, $deadline);
822: } catch (RemoteCallFailedException $e) {
823: // SEP-2575: the client SHOULD retry once with a version the rejection named, and never retry the retry.
824: $retry = $this->renegotiateProtocolVersion($request, $e);
825:
826: if (null === $retry) {
827: throw $e;
828: }
829:
830: $this->logger->info(
831: 'Retrying request {id} as {retry}: the server does not support {requested}.',
832: ['id' => $request->id->id, 'retry' => $retry->id->id, 'requested' => SafeDisplay::sanitiseCause($e->error->message)],
833: );
834:
835: return $this->exchange($retry, $response, $context, $deadline);
836: }
837: } finally {
838: $deadline?->release();
839: }
840: }
841:
842: /**
843: * @template TResponse of JsonRpcResultResponse
844: *
845: * @param JsonRpcRequest<non-empty-string> $request
846: * @param class-string<TResponse> $response
847: *
848: * @return TResponse
849: *
850: * @throws LogicException
851: * @throws RemoteCallFailedException
852: * @throws RequestTimeoutException
853: * @throws ServerCapabilityNotSupportedException
854: */
855: private function exchange(
856: JsonRpcRequest $request,
857: string $response,
858: ?SendContext $context,
859: ?RequestDeadline $deadline,
860: ): JsonRpcResultResponse {
861: $transport = $this->requireConnectedTransport();
862:
863: $this->assertServerSupports($request::getMethod());
864:
865: $retained = $this->retainsAcrossRestart($request);
866: $future = $this->outboundRequests->register(
867: $request->id,
868: $response,
869: $retained ? $request : null,
870: $retained ? $context : null,
871: );
872:
873: try {
874: $transport->send($request, $context);
875: } catch (\Throwable $e) {
876: $this->outboundRequests->forget($request->id);
877:
878: throw $e;
879: }
880:
881: if (null === $deadline) {
882: return $future->await();
883: }
884:
885: try {
886: return $future->await($deadline->getCancellation());
887: } catch (CancelledException $e) {
888: throw $this->abandon($request, $transport, $deadline, $e);
889: }
890: }
891:
892: /**
893: * @template TMethod of non-empty-string
894: *
895: * @param JsonRpcRequest<TMethod> $request
896: *
897: * @return null|JsonRpcRequest<TMethod>
898: */
899: private function renegotiateProtocolVersion(JsonRpcRequest $request, RemoteCallFailedException $failure): ?JsonRpcRequest
900: {
901: $error = $failure->error;
902:
903: if (! $error instanceof UnsupportedProtocolVersionError) {
904: return null;
905: }
906:
907: $version = $this->pickSupportedVersion($error->supported);
908:
909: if (null === $version) {
910: return null;
911: }
912:
913: $params = $request->params;
914:
915: if (! $params instanceof RequestParams) {
916: return null;
917: }
918:
919: $meta = $params->meta->toArray();
920: $meta[RequestMetaObject::PROTOCOL_VERSION_KEY] = $version;
921:
922: $fields = $params->toArray();
923: $fields['_meta'] = $meta;
924:
925: $envelope = $request->toArray();
926: $envelope['params'] = $fields;
927: $envelope['id'] = $this->mintRequestId()->id;
928:
929: return $request::fromArray($envelope);
930: }
931:
932: /**
933: * The first version the peer named that this SDK also speaks.
934: *
935: * @param list<string> $supported
936: */
937: private function pickSupportedVersion(array $supported): ?string
938: {
939: foreach ($supported as $version) {
940: if (\in_array($version, ProtocolVersion::SUPPORTED_VERSIONS, true)) {
941: return $version;
942: }
943: }
944:
945: return null;
946: }
947:
948: /**
949: * @param JsonRpcRequest<non-empty-string> $request
950: */
951: private function abandon(
952: JsonRpcRequest $request,
953: TransportInterface $transport,
954: RequestDeadline $deadline,
955: CancelledException $cause,
956: ): RequestTimeoutException {
957: $this->outboundRequests->forget($request->id);
958: $this->abortExchange($transport, $request->id);
959:
960: try {
961: $transport->send(new CancelledNotification(
962: params: new CancelledNotificationParams(requestId: $request->id, reason: 'The request timed out.'),
963: ));
964: } catch (\Throwable $e) {
965: $this->logger->warning(
966: 'Could not tell the server that request {id} was abandoned.',
967: ['id' => $request->id->id, 'exception' => $e],
968: );
969: }
970:
971: return new RequestTimeoutException($request->id, $deadline->readElapsed(), $cause);
972: }
973:
974: /**
975: * @param null|float $timeout Overrides the configured idle deadline for one request
976: */
977: private function openDeadline(?float $timeout = null): ?RequestDeadline
978: {
979: $timeout ??= $this->requestTimeout;
980:
981: return null === $timeout ? null : new RequestDeadline($timeout, $this->maxRequestTimeout);
982: }
983:
984: /**
985: * Caches the `x-mcp-header` bindings of every tool whose declarations hold, and drops the rest from the
986: * listing.
987: */
988: private function admitMirrorableTools(ListToolsResult $result): ListToolsResult
989: {
990: $admitted = [];
991:
992: foreach ($result->tools as $tool) {
993: $scan = ParameterHeaderScanner::scan($tool->inputSchema);
994:
995: if (! $scan->valid) {
996: unset($this->toolHeaderBindings[$tool->name]);
997:
998: $reason = $scan->reason;
999: \assert(\is_string($reason));
1000:
1001: $this->logger->warning(
1002: 'Excluding tool {tool} from the listing: its "x-mcp-header" declarations are invalid.',
1003: ['tool' => SafeDisplay::sanitise($tool->name), 'reason' => SafeDisplay::sanitiseCause($reason)],
1004: );
1005:
1006: continue;
1007: }
1008:
1009: $this->toolHeaderBindings[$tool->name] = $scan->bindings;
1010: $admitted[] = $tool;
1011: }
1012:
1013: return $admitted === $result->tools ? $result : new ListToolsResult(
1014: tools: $admitted,
1015: ttlMs: $result->ttlMs,
1016: cacheScope: $result->cacheScope,
1017: nextCursor: $result->nextCursor,
1018: meta: $result->meta,
1019: );
1020: }
1021:
1022: /**
1023: * @param null|array<array-key, mixed> $arguments
1024: *
1025: * @return array<non-empty-string, string>
1026: */
1027: private function mirrorParameterHeaders(string $name, ?array $arguments): array
1028: {
1029: return $this->parameterHeaders->build($this->toolHeaderBindings[$name] ?? [], $arguments ?? []);
1030: }
1031:
1032: /**
1033: * @throws ServerCapabilityNotSupportedException
1034: */
1035: private function assertServerSupports(string $method): void
1036: {
1037: $capabilities = $this->serverCapabilities->current();
1038:
1039: if (null === $capabilities) {
1040: return;
1041: }
1042:
1043: $owner = $this->extensionMethods[$method] ?? null;
1044:
1045: if (null !== $owner && ! \array_key_exists($owner, $capabilities->extensions ?? [])) {
1046: throw new ServerCapabilityNotSupportedException($method);
1047: }
1048:
1049: $supported = match ($method) {
1050: ListToolsRequest::getMethod(), CallToolRequest::getMethod() => null !== $capabilities->tools,
1051: ListResourcesRequest::getMethod(),
1052: ListResourceTemplatesRequest::getMethod(),
1053: ReadResourceRequest::getMethod() => null !== $capabilities->resources,
1054: ListPromptsRequest::getMethod(), GetPromptRequest::getMethod() => null !== $capabilities->prompts,
1055: CompleteRequest::getMethod() => null !== $capabilities->completions,
1056: default => true,
1057: };
1058:
1059: if (! $supported) {
1060: throw new ServerCapabilityNotSupportedException($method);
1061: }
1062: }
1063:
1064: private function mintProgressToken(): ProgressToken
1065: {
1066: return new ProgressToken(token: ($this->progressTokenFactory)());
1067: }
1068:
1069: private function requireConnectedTransport(): TransportInterface
1070: {
1071: return $this->transport ?? throw new LogicException('Client is not connected. Call connect() first.');
1072: }
1073: }
1074:
Nexus MCP SDK (1.x-dev) API documentation generated by ApiGen dev-master