File Client/Client.php | Nexus MCP SDK API

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\ProgressListenerRegistry;
19:	use Nexus\Mcp\Client\Dispatch\RequestDeadline;
20:	use Nexus\Mcp\Client\Exception\ClientAlreadyConnectedException;
21:	use Nexus\Mcp\Client\Exception\ClientNotConnectedException;
22:	use Nexus\Mcp\Client\Exception\ServerCapabilityNotSupportedException;
23:	use Nexus\Mcp\Client\Subscription\OpenSubscription;
24:	use Nexus\Mcp\Client\Subscription\SubscriptionRegistry;
25:	use Nexus\Mcp\Client\Subscription\SubscriptionStream;
26:	use Nexus\Mcp\Core\Dispatch\MessageDispatcherInterface;
27:	use Nexus\Mcp\Core\Dispatch\PendingOutboundRequests;
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\Schema\ClientCapabilities;
37:	use Nexus\Mcp\Core\Schema\Cursor;
38:	use Nexus\Mcp\Core\Schema\Enum\ProtocolErrorCode;
39:	use Nexus\Mcp\Core\Schema\Error\UnsupportedProtocolVersionError;
40:	use Nexus\Mcp\Core\Schema\Implementation;
41:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
42:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcRequest;
43:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResultResponse;
44:	use Nexus\Mcp\Core\Schema\MetaObject\RequestMetaObject;
45:	use Nexus\Mcp\Core\Schema\Notification\CancelledNotification;
46:	use Nexus\Mcp\Core\Schema\NotificationParams\CancelledNotificationParams;
47:	use Nexus\Mcp\Core\Schema\ProgressToken;
48:	use Nexus\Mcp\Core\Schema\Prompt\PromptReference;
49:	use Nexus\Mcp\Core\Schema\ProtocolVersion;
50:	use Nexus\Mcp\Core\Schema\Request\CallToolRequest;
51:	use Nexus\Mcp\Core\Schema\Request\CompleteRequest;
52:	use Nexus\Mcp\Core\Schema\Request\DiscoverRequest;
53:	use Nexus\Mcp\Core\Schema\Request\GetPromptRequest;
54:	use Nexus\Mcp\Core\Schema\Request\ListPromptsRequest;
55:	use Nexus\Mcp\Core\Schema\Request\ListResourcesRequest;
56:	use Nexus\Mcp\Core\Schema\Request\ListResourceTemplatesRequest;
57:	use Nexus\Mcp\Core\Schema\Request\ListToolsRequest;
58:	use Nexus\Mcp\Core\Schema\Request\ReadResourceRequest;
59:	use Nexus\Mcp\Core\Schema\Request\SubscriptionsListenRequest;
60:	use Nexus\Mcp\Core\Schema\RequestId;
61:	use Nexus\Mcp\Core\Schema\RequestParams;
62:	use Nexus\Mcp\Core\Schema\RequestParams\CallToolRequestParams;
63:	use Nexus\Mcp\Core\Schema\RequestParams\CompleteRequestParams;
64:	use Nexus\Mcp\Core\Schema\RequestParams\EmptyRequestParams;
65:	use Nexus\Mcp\Core\Schema\RequestParams\GetPromptRequestParams;
66:	use Nexus\Mcp\Core\Schema\RequestParams\InputResponseRequestParams;
67:	use Nexus\Mcp\Core\Schema\RequestParams\PaginatedRequestParams;
68:	use Nexus\Mcp\Core\Schema\RequestParams\ReadResourceRequestParams;
69:	use Nexus\Mcp\Core\Schema\RequestParams\SubscriptionsListenRequestParams;
70:	use Nexus\Mcp\Core\Schema\RequestParamsInterface;
71:	use Nexus\Mcp\Core\Schema\Resource\ResourceTemplateReference;
72:	use Nexus\Mcp\Core\Schema\Result\CallToolResult;
73:	use Nexus\Mcp\Core\Schema\Result\CompleteResult;
74:	use Nexus\Mcp\Core\Schema\Result\DiscoverResult;
75:	use Nexus\Mcp\Core\Schema\Result\GetPromptResult;
76:	use Nexus\Mcp\Core\Schema\Result\InputRequiredResult;
77:	use Nexus\Mcp\Core\Schema\Result\InputResponse;
78:	use Nexus\Mcp\Core\Schema\Result\ListPromptsResult;
79:	use Nexus\Mcp\Core\Schema\Result\ListResourcesResult;
80:	use Nexus\Mcp\Core\Schema\Result\ListResourceTemplatesResult;
81:	use Nexus\Mcp\Core\Schema\Result\ListToolsResult;
82:	use Nexus\Mcp\Core\Schema\Result\ReadResourceResult;
83:	use Nexus\Mcp\Core\Schema\Result\SubscriptionsListenResult;
84:	use Nexus\Mcp\Core\Schema\ResultResponse\CallToolResultResponse;
85:	use Nexus\Mcp\Core\Schema\ResultResponse\CompleteResultResponse;
86:	use Nexus\Mcp\Core\Schema\ResultResponse\DiscoverResultResponse;
87:	use Nexus\Mcp\Core\Schema\ResultResponse\GetPromptResultResponse;
88:	use Nexus\Mcp\Core\Schema\ResultResponse\ListPromptsResultResponse;
89:	use Nexus\Mcp\Core\Schema\ResultResponse\ListResourcesResultResponse;
90:	use Nexus\Mcp\Core\Schema\ResultResponse\ListResourceTemplatesResultResponse;
91:	use Nexus\Mcp\Core\Schema\ResultResponse\ListToolsResultResponse;
92:	use Nexus\Mcp\Core\Schema\ResultResponse\ReadResourceResultResponse;
93:	use Nexus\Mcp\Core\Schema\ResultResponse\SubscriptionsListenResultResponse;
94:	use Nexus\Mcp\Core\Schema\ServerCapabilities;
95:	use Nexus\Mcp\Core\Schema\SubscriptionFilter;
96:	use Nexus\Mcp\Core\Transport\AbortableTransportInterface;
97:	use Nexus\Mcp\Core\Transport\ParameterHeaderMirroringInterface;
98:	use Nexus\Mcp\Core\Transport\ReceiveContext;
99:	use Nexus\Mcp\Core\Transport\ReconnectingTransportInterface;
100:	use Nexus\Mcp\Core\Transport\SendContext;
101:	use Nexus\Mcp\Core\Transport\TransportInterface;
102:	use Psr\Log\LoggerInterface;
103:	use Psr\Log\NullLogger;
104:	use Revolt\EventLoop;
105:
106:	/**
107:	* Client-side entry point: drives the transport lifecycle and exposes the typed
108:	* JSON-RPC operations a client issues against an MCP server.
109:	*/
110:	final class Client
111:	{
112:	/**
113:	* Seconds a request may go unanswered before it is abandoned. Each progress notification for the
114:	* request restarts it.
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:	/**
124:	* The requests a lost call may be sent again as. A retry is at-least-once, because the peer may have
125:	* carried the work out before it died, so only requests that read state are eligible. The spec marks
126:	* no tool as idempotent, which keeps `tools/call` off the list however harmless a given tool is, and
127:	* a vendor method sent through `sendRequest()` has no semantics this SDK can judge.
128:	*/
129:	private const array RETRYABLE_REQUESTS = [
130:	CompleteRequest::class,
131:	DiscoverRequest::class,
132:	GetPromptRequest::class,
133:	ListPromptsRequest::class,
134:	ListResourcesRequest::class,
135:	ListResourceTemplatesRequest::class,
136:	ListToolsRequest::class,
137:	ReadResourceRequest::class,
138:	];
139:
140:	private ?TransportInterface $transport = null;
141:	private ?Implementation $serverInfo = null;
142:	private ?ServerCapabilities $serverCapabilities = null;
143:
144:	/**
145:	* `x-mcp-header` bindings of every tool a `tools/list` admitted, keyed by tool name. Only a transport
146:	* that mirrors parameter headers populates it.
147:	*
148:	* @var array<string, list<ParameterHeaderBinding>>
149:	*/
150:	private array $toolHeaderBindings = [];
151:
152:	/**
153:	* @param \Closure(): (int\|non-empty-string) $requestIdFactory
154:	* @param \Closure(): (int\|non-empty-string) $progressTokenFactory
155:	* @param null\|float $requestTimeout Seconds a request may go unanswered, or `null` to wait indefinitely
156:	* @param null\|float $maxRequestTimeout Seconds a request may run however much progress arrives, or `null` to leave it unbounded
157:	*/
158:	public function __construct(
159:	private readonly Implementation $clientInfo,
160:	private readonly ClientCapabilities $clientCapabilities,
161:	private readonly MessageDispatcherInterface $dispatcher,
162:	private readonly PendingOutboundRequests $outboundRequests,
163:	private readonly \Closure $requestIdFactory,
164:	private readonly \Closure $progressTokenFactory,
165:	private readonly ProtocolVersion $protocolVersion = new ProtocolVersion(version: ProtocolVersion::LATEST_VERSION),
166:	private readonly ProgressListenerRegistry $progressListeners = new ProgressListenerRegistry(),
167:	private readonly SubscriptionRegistry $subscriptions = new SubscriptionRegistry(),
168:	private readonly LoggerInterface $logger = new NullLogger(),
169:	private readonly ?float $requestTimeout = self::DEFAULT_REQUEST_TIMEOUT,
170:	private readonly ?float $maxRequestTimeout = self::DEFAULT_MAX_REQUEST_TIMEOUT,
171:	private readonly bool $retryLostRequests = false,
172:	) {
173:	}
174:
175:	/**
176:	* Non-blocking connect to the transport.
177:	*
178:	* @throws ClientAlreadyConnectedException
179:	*/
180:	public function connect(TransportInterface $transport): void
181:	{
182:	if (null !== $this->transport) {
183:	// Reject reentry to avoid orphaning the previous transport.
184:	throw new ClientAlreadyConnectedException();
185:	}
186:
187:	$this->logger->info('Starting MCP client.');
188:
189:	$this->transport = $transport;
190:
191:	$transport->onMessage(function (array $envelope, ReceiveContext $context) use ($transport): void {
192:	$this->dispatcher->dispatch($envelope, $transport, $context);
193:	});
194:	$transport->onError(function (\Throwable $e): void {
195:	if ($e instanceof OutboundRequestFailedException) {
196:	// The exchange that carried this request is over, so its response can no longer arrive.
197:	// A caller still awaiting one would otherwise block for the life of the process.
198:	$this->outboundRequests->reject($e->requestId, $e);
199:	}
200:
201:	if ($e instanceof SupervisionExhaustedException) {
202:	// No further peer is coming, so a stream held open across the restarts has nothing left
203:	// to be re-opened against, and a retained request has nothing left to be sent to. Both
204:	// would otherwise wait for the life of the process.
205:	$this->settleSubscriptions($e);
206:	$this->outboundRequests->cancelAll($e);
207:	}
208:
209:	$this->logger->error('Transport error.', ['exception' => $e]);
210:	});
211:	$transport->onDrain(function (): void {
212:	$this->dispatcher->flushPending();
213:	});
214:	$transport->onClose(function () use ($transport): void {
215:	$error = new TransportAlreadyClosedException(operation: 'await-response');
216:
217:	// A retained request outlives the peer that was carrying it, so only the rest fail here.
218:	$this->outboundRequests->cancelUnretained($error);
219:
220:	// The supervisor decides on a replacement after emitting this close, so whether one is coming
221:	// is only readable on the next tick.
222:	EventLoop::queue(function () use ($transport, $error): void {
223:	if ($transport !== $this->transport) {
224:	// A transport this client has already let go of speaks for nothing that is pending
225:	// now. `disconnect()` failed what it was owed before detaching it.
226:	return;
227:	}
228:
229:	if ($transport instanceof ReconnectingTransportInterface && $transport->isReconnecting()) {
230:	return;
231:	}
232:
233:	// Nothing replaces this peer, so a retained request has run out of peers to be sent to.
234:	// Streams need no help here: the close that got us this far freed their correlation
235:	// slots, and each stream settles from its own failed exchange.
236:	$this->outboundRequests->cancelAll($error);
237:	});
238:	});
239:
240:	if ($transport instanceof ReconnectingTransportInterface) {
241:	$transport->onReconnect(function () use ($transport): void {
242:	foreach ($this->outboundRequests->collectRetained() as $retained) {
243:	$request = $retained['request'];
244:
245:	try {
246:	$transport->send($request, $retained['context']);
247:	} catch (\Throwable $e) {
248:	if ($transport->isReconnecting()) {
249:	// This replacement died too. Left retained so the peer after it tries again,
250:	// matching what the subscription walk below does.
251:	$this->logger->warning(
252:	'Could not send request {id} again to the replacement peer.',
253:	['id' => $request->id->id, 'exception' => $e],
254:	);
255:
256:	continue;
257:	}
258:
259:	// Nothing else will carry it, so the caller hears now rather than at the deadline.
260:	$this->outboundRequests->reject($request->id, $e);
261:	}
262:	}
263:
264:	foreach ($this->subscriptions->all() as $subscription) {
265:	try {
266:	$this->openStream($subscription, $transport);
267:	} catch (\Throwable $e) {
268:	// Left registered, so the next replacement peer gets another go at it.
269:	$this->logger->error(
270:	'Could not re-open subscription {id} against the replacement peer.',
271:	['id' => $subscription->subscriptionId->id, 'exception' => $e],
272:	);
273:	}
274:	}
275:	});
276:	}
277:
278:	$transport->start();
279:	}
280:
281:	/**
282:	* Closes the transport and detaches it so a fresh `connect()` can run.
283:	* A no-op when the client is not connected.
284:	*/
285:	public function disconnect(): void
286:	{
287:	$transport = $this->transport;
288:	$this->transport = null;
289:
290:	// The cached bindings describe the server that just went away, so a later connection must not
291:	// mirror headers from them.
292:	$this->toolHeaderBindings = [];
293:
294:	// Settled before the close, so a supervised transport cannot answer the peer loss it is about to
295:	// see by re-opening streams, or re-sending requests, the caller has just given up.
296:	$error = new TransportAlreadyClosedException(operation: 'await-response');
297:	$this->settleSubscriptions($error);
298:	$this->outboundRequests->cancelAll($error);
299:
300:	$transport?->close();
301:	}
302:
303:	/**
304:	* The server's `Implementation` block from the last `server/discover`
305:	* response `_meta`, or `null` if discovery has not run or the server did
306:	* not identify itself.
307:	*/
308:	public function getServerInfo(): ?Implementation
309:	{
310:	return $this->serverInfo;
311:	}
312:
313:	/**
314:	* The server's capabilities from the last `server/discover` response, or
315:	* `null` if discovery has not run.
316:	*/
317:	public function getServerCapabilities(): ?ServerCapabilities
318:	{
319:	return $this->serverCapabilities;
320:	}
321:
322:	/**
323:	* Sends `server/discover` and records the advertised server info and capabilities.
324:	*
325:	* @throws ClientNotConnectedException
326:	* @throws RequestTimeoutException
327:	* @throws ServerCapabilityNotSupportedException
328:	* @throws TransportAlreadyClosedException
329:	*/
330:	public function discover(): DiscoverResult
331:	{
332:	$result = $this->sendRequest(
333:	new DiscoverRequest(id: $this->mintRequestId(), params: new EmptyRequestParams(meta: $this->stampMeta())),
334:	DiscoverResultResponse::class,
335:	)->result;
336:
337:	$this->serverInfo = $result->meta->serverInfo;
338:	$this->serverCapabilities = $result->capabilities;
339:
340:	return $result;
341:	}
342:
343:	/**
344:	* Opens a `subscriptions/listen` stream and routes every notification the server tags with its id to
345:	* `$onNotification`. Returns as soon as the request is away: the stream runs until either side ends it.
346:	*
347:	* @param \Closure(JsonRpcNotification<non-empty-string>): void $onNotification
348:	*
349:	* @throws ClientNotConnectedException
350:	* @throws TransportAlreadyClosedException
351:	*/
352:	public function listen(SubscriptionFilter $notifications, \Closure $onNotification): SubscriptionStream
353:	{
354:	$transport = $this->transport ?? throw new ClientNotConnectedException();
355:
356:	$id = $this->mintRequestId();
357:
358:	/** @var DeferredFuture<SubscriptionsListenResult> $outcome */
359:	$outcome = new DeferredFuture();
360:
361:	// Only an explicit await() observes the outcome. Left unignored, a refused subscription would
362:	// surface as an unhandled future when the stream is collected.
363:	$future = $outcome->getFuture();
364:	$future->ignore();
365:
366:	$subscription = new OpenSubscription($id, $notifications, $onNotification, $outcome);
367:
368:	// Routed only once the correlation slot is claimed, so an id already in flight is refused before
369:	// this stream can displace the routing entry of the live one that owns it.
370:	$this->openStream($subscription, $transport);
371:	$this->subscriptions->register($subscription);
372:
373:	return new SubscriptionStream($id, $future, function () use ($id, $transport): void {
374:	$this->subscriptions->forget($id);
375:
376:	if (! $this->outboundRequests->forget($id)) {
377:	// The server already answered, so no in-flight request remains for a cancellation to name.
378:	return;
379:	}
380:
381:	self::abortExchange($transport, $id);
382:
383:	try {
384:	$transport->send(new CancelledNotification(
385:	params: new CancelledNotificationParams(requestId: $id, reason: 'The subscription was closed.'),
386:	));
387:	} catch (\Throwable $e) {
388:	// Closing a stream whose transport already went away is teardown, not a failure to report.
389:	$this->logger->debug(
390:	'Could not tell the server that subscription {id} was closed.',
391:	['id' => $id->id, 'exception' => $e],
392:	);
393:	}
394:	});
395:	}
396:
397:	/**
398:	* @throws ClientNotConnectedException
399:	* @throws RequestTimeoutException
400:	* @throws ServerCapabilityNotSupportedException
401:	* @throws TransportAlreadyClosedException
402:	*/
403:	public function listTools(?Cursor $cursor = null): ListToolsResult
404:	{
405:	$result = $this->sendRequest(
406:	new ListToolsRequest(
407:	id: $this->mintRequestId(),
408:	params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
409:	),
410:	ListToolsResultResponse::class,
411:	)->result;
412:
413:	return $this->transport instanceof ParameterHeaderMirroringInterface
414:	? $this->admitMirrorableTools($result)
415:	: $result;
416:	}
417:
418:	/**
419:	* @throws ClientNotConnectedException
420:	* @throws RequestTimeoutException
421:	* @throws ServerCapabilityNotSupportedException
422:	* @throws TransportAlreadyClosedException
423:	*/
424:	public function listResources(?Cursor $cursor = null): ListResourcesResult
425:	{
426:	return $this->sendRequest(
427:	new ListResourcesRequest(
428:	id: $this->mintRequestId(),
429:	params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
430:	),
431:	ListResourcesResultResponse::class,
432:	)->result;
433:	}
434:
435:	/**
436:	* @throws ClientNotConnectedException
437:	* @throws RequestTimeoutException
438:	* @throws ServerCapabilityNotSupportedException
439:	* @throws TransportAlreadyClosedException
440:	*/
441:	public function listResourceTemplates(?Cursor $cursor = null): ListResourceTemplatesResult
442:	{
443:	return $this->sendRequest(
444:	new ListResourceTemplatesRequest(
445:	id: $this->mintRequestId(),
446:	params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
447:	),
448:	ListResourceTemplatesResultResponse::class,
449:	)->result;
450:	}
451:
452:	/**
453:	* @throws ClientNotConnectedException
454:	* @throws RequestTimeoutException
455:	* @throws ServerCapabilityNotSupportedException
456:	* @throws TransportAlreadyClosedException
457:	*/
458:	public function listPrompts(?Cursor $cursor = null): ListPromptsResult
459:	{
460:	return $this->sendRequest(
461:	new ListPromptsRequest(
462:	id: $this->mintRequestId(),
463:	params: new PaginatedRequestParams(meta: $this->stampMeta(), cursor: $cursor),
464:	),
465:	ListPromptsResultResponse::class,
466:	)->result;
467:	}
468:
469:	/**
470:	* @throws ClientNotConnectedException
471:	* @throws RequestTimeoutException
472:	* @throws ServerCapabilityNotSupportedException
473:	* @throws TransportAlreadyClosedException
474:	*/
475:	public function readResource(string $uri): InputRequiredResult\|ReadResourceResult
476:	{
477:	return $this->sendRequest(
478:	new ReadResourceRequest(
479:	id: $this->mintRequestId(),
480:	params: new ReadResourceRequestParams(uri: $uri, meta: $this->stampMeta()),
481:	),
482:	ReadResourceResultResponse::class,
483:	)->result;
484:	}
485:
486:	/**
487:	* @param null\|array<string, string> $arguments
488:	*
489:	* @throws ClientNotConnectedException
490:	* @throws RequestTimeoutException
491:	* @throws ServerCapabilityNotSupportedException
492:	* @throws TransportAlreadyClosedException
493:	*/
494:	public function getPrompt(string $name, ?array $arguments = null): GetPromptResult\|InputRequiredResult
495:	{
496:	return $this->sendRequest(
497:	new GetPromptRequest(
498:	id: $this->mintRequestId(),
499:	params: new GetPromptRequestParams(name: $name, meta: $this->stampMeta(), arguments: $arguments),
500:	),
501:	GetPromptResultResponse::class,
502:	)->result;
503:	}
504:
505:	/**
506:	* @param array{name: string, value: string} $argument
507:	* @param null\|array{arguments?: array<string, string>} $context
508:	*
509:	* @throws ClientNotConnectedException
510:	* @throws RequestTimeoutException
511:	* @throws ServerCapabilityNotSupportedException
512:	* @throws TransportAlreadyClosedException
513:	*/
514:	public function complete(
515:	PromptReference\|ResourceTemplateReference $ref,
516:	array $argument,
517:	?array $context = null,
518:	): CompleteResult {
519:	return $this->sendRequest(
520:	new CompleteRequest(
521:	id: $this->mintRequestId(),
522:	params: new CompleteRequestParams(
523:	ref: $ref,
524:	argument: $argument,
525:	meta: $this->stampMeta(),
526:	context: $context,
527:	),
528:	),
529:	CompleteResultResponse::class,
530:	)->result;
531:	}
532:
533:	/**
534:	* Invokes a tool. When `$onProgress` is given, a fresh `progressToken` is
535:	* minted into the request's `_meta` and the callback receives every
536:	* matching `notifications/progress` for the duration of the call.
537:	*
538:	* A server that needs more input answers with an `InputRequiredResult` rather
539:	* than a result. Satisfy each of its `inputRequests` and call again with the
540:	* answers plus the `requestState` it carried, which is opaque and must be
541:	* echoed back unchanged.
542:	*
543:	* @param null\|array<string, mixed> $arguments
544:	* @param null\|\Closure(float $progress, ?float $total, ?string $message): void $onProgress
545:	* @param null\|array<string, InputResponse> $inputResponses Answers to a prior `InputRequiredResult`, keyed as its `inputRequests` were
546:	* @param null\|string $requestState Echoed verbatim from the `InputRequiredResult` being answered
547:	*
548:	* @throws ClientNotConnectedException
549:	* @throws RequestTimeoutException
550:	* @throws ServerCapabilityNotSupportedException
551:	* @throws TransportAlreadyClosedException
552:	*/
553:	public function callTool(
554:	string $name,
555:	?array $arguments = null,
556:	?\Closure $onProgress = null,
557:	?array $inputResponses = null,
558:	?string $requestState = null,
559:	): CallToolResult\|InputRequiredResult {
560:	try {
561:	return $this->attemptToolCall($name, $arguments, $onProgress, $inputResponses, $requestState);
562:	} catch (RemoteCallFailedException $e) {
563:	if ($e->getCode() !== ProtocolErrorCode::HeaderMismatch->value) {
564:	throw $e;
565:	}
566:	}
567:
568:	// A header mismatch means the cached `inputSchema` is behind the server's, so refresh it and
569:	// retry exactly once. A second mismatch is the server's answer and propagates.
570:	$this->refreshToolHeaderBindings($name);
571:
572:	return $this->attemptToolCall($name, $arguments, $onProgress, $inputResponses, $requestState);
573:	}
574:
575:	/**
576:	* Sends an outbound JSON-RPC request and awaits the correlated response.
577:	*
578:	* @template TResponse of JsonRpcResultResponse = JsonRpcResultResponse
579:	*
580:	* @param JsonRpcRequest<non-empty-string> $request
581:	* @param class-string<TResponse> $response
582:	* @param null\|float $timeout Seconds this one request may go unanswered, overriding the client's default
583:	*
584:	* @return TResponse
585:	*
586:	* @throws ClientNotConnectedException
587:	* @throws RequestTimeoutException
588:	* @throws ServerCapabilityNotSupportedException
589:	* @throws TransportAlreadyClosedException
590:	*/
591:	public function sendRequest(
592:	JsonRpcRequest $request,
593:	string $response,
594:	?SendContext $context = null,
595:	?float $timeout = null,
596:	): JsonRpcResultResponse {
597:	return $this->dispatch($request, $response, $context, $this->openDeadline($timeout));
598:	}
599:
600:	/**
601:	* Sends `$subscription`'s listen request on `$transport` and routes that one connection's answer to
602:	* the caller-facing outcome. Runs once per connection the subscription is carried on.
603:	*
604:	* @throws TransportAlreadyClosedException
605:	*/
606:	private function openStream(OpenSubscription $subscription, TransportInterface $transport): void
607:	{
608:	$id = $subscription->subscriptionId;
609:
610:	// Claim the correlation slot first: a duplicate id must not leave a routing entry behind.
611:	$response = $this->outboundRequests->register($id, SubscriptionsListenResultResponse::class);
612:
613:	$response
614:	->map(function (SubscriptionsListenResultResponse $response) use ($id): void {
615:	// Absent when the caller closed the stream first, which owes them no outcome.
616:	$this->subscriptions->forget($id)?->outcome->complete($response->result);
617:	})
618:	->catch(function (\Throwable $e) use ($id, $transport): void {
619:	// Read here rather than when the request went out, because only a failure that a
620:	// replacement peer will be given another go at leaves the stream owing an outcome. A peer
621:	// that answers "no" is answering, so it ends the stream however replaceable it was.
622:	if ($transport instanceof ReconnectingTransportInterface && $transport->isReconnecting()) {
623:	return;
624:	}
625:
626:	$this->subscriptions->forget($id)?->outcome->error($e);
627:	})
628:	->ignore()
629:	;
630:
631:	try {
632:	$transport->send(new SubscriptionsListenRequest(
633:	id: $id,
634:	params: new SubscriptionsListenRequestParams(
635:	notifications: $subscription->notifications,
636:	meta: $this->stampMeta(),
637:	),
638:	));
639:	} catch (\Throwable $e) {
640:	$this->outboundRequests->forget($id);
641:
642:	throw $e;
643:	}
644:	}
645:
646:	/**
647:	* Whether a peer that dies mid-`$request` should be replaced and the request sent again, rather than
648:	* the caller hearing that the connection went away.
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:	// A multi-round-trip continuation only names a state-reading method. It carries the user's
663:	// answers and an opaque resume token, so sending it again hands a one-time answer over
664:	// twice and resumes work the peer that issued the token no longer has.
665:	return ! self::resumesAnEarlierRound($request->params);
666:	}
667:	}
668:
669:	return false;
670:	}
671:
672:	/**
673:	* Whether these params continue an exchange the server suspended, rather than opening a fresh one.
674:	*/
675:	private static function resumesAnEarlierRound(?RequestParamsInterface $params): bool
676:	{
677:	[$inputResponses, $requestState] = match (true) {
678:	$params instanceof InputResponseRequestParams => [$params->inputResponses, $params->requestState],
679:	$params instanceof ReadResourceRequestParams => [$params->inputResponses, $params->requestState],
680:	default => [null, null],
681:	};
682:
683:	return null !== $inputResponses \|\| null !== $requestState;
684:	}
685:
686:	/**
687:	* Stops the transport-level work still running for a request nobody is waiting on any more. A stream
688:	* left reading would otherwise run until the peer ends it, which a `subscriptions/listen` never does.
689:	*/
690:	private static function abortExchange(TransportInterface $transport, RequestId $id): void
691:	{
692:	if ($transport instanceof AbortableTransportInterface) {
693:	$transport->abort($id);
694:	}
695:	}
696:
697:	/**
698:	* Fails every stream still open, for a client that will not be re-opening them.
699:	*/
700:	private function settleSubscriptions(\Throwable $reason): void
701:	{
702:	foreach ($this->subscriptions->drain() as $subscription) {
703:	$subscription->outcome->error($reason);
704:	}
705:	}
706:
707:	/**
708:	* Re-lists the named tool so its cached `x-mcp-header` bindings match the server's current
709:	* `inputSchema`, walking pages until the listing yields it or runs out of them.
710:	*/
711:	private function refreshToolHeaderBindings(string $name): void
712:	{
713:	$cursor = null;
714:
715:	do {
716:	$page = $this->listTools($cursor);
717:
718:	foreach ($page->tools as $tool) {
719:	if ($tool->name === $name) {
720:	return;
721:	}
722:	}
723:
724:	$cursor = $page->nextCursor;
725:	} while (null !== $cursor);
726:	}
727:
728:	/**
729:	* One `tools/call` attempt, mirroring whatever parameter headers are cached for the tool.
730:	*
731:	* @param null\|array<string, mixed> $arguments
732:	* @param null\|\Closure(float $progress, ?float $total, ?string $message): void $onProgress
733:	* @param null\|array<string, InputResponse> $inputResponses
734:	*/
735:	private function attemptToolCall(
736:	string $name,
737:	?array $arguments,
738:	?\Closure $onProgress,
739:	?array $inputResponses = null,
740:	?string $requestState = null,
741:	): CallToolResult\|InputRequiredResult {
742:	$context = new SendContext(headers: $this->mirrorParameterHeaders($name, $arguments));
743:
744:	if (null === $onProgress) {
745:	return $this->sendRequest(
746:	new CallToolRequest(
747:	id: $this->mintRequestId(),
748:	params: new CallToolRequestParams(
749:	name: $name,
750:	meta: $this->stampMeta(),
751:	arguments: $arguments,
752:	inputResponses: $inputResponses,
753:	requestState: $requestState,
754:	),
755:	),
756:	CallToolResultResponse::class,
757:	$context,
758:	)->result;
759:	}
760:
761:	$progressToken = $this->mintProgressToken();
762:
763:	// The deadline arms its timers on construction, so everything from here on runs under the `finally`
764:	// that disarms them. A throw in between would otherwise hold the event loop open for the ceiling.
765:	$deadline = $this->openDeadline();
766:
767:	try {
768:	// Progress means the call is alive, so each report buys it another idle window, up to the ceiling.
769:	$this->progressListeners->register(
770:	$progressToken,
771:	static function (float $progress, ?float $total, ?string $message) use ($onProgress, $deadline): void {
772:	$deadline?->extend();
773:	$onProgress($progress, $total, $message);
774:	},
775:	);
776:
777:	return $this->dispatch(
778:	new CallToolRequest(
779:	id: $this->mintRequestId(),
780:	params: new CallToolRequestParams(
781:	name: $name,
782:	meta: $this->stampMeta($progressToken),
783:	arguments: $arguments,
784:	inputResponses: $inputResponses,
785:	requestState: $requestState,
786:	),
787:	),
788:	CallToolResultResponse::class,
789:	$context,
790:	$deadline,
791:	)->result;
792:	} finally {
793:	$deadline?->release();
794:	$this->progressListeners->unregister($progressToken);
795:	}
796:	}
797:
798:	/**
799:	* @template TResponse of JsonRpcResultResponse
800:	*
801:	* @param JsonRpcRequest<non-empty-string> $request
802:	* @param class-string<TResponse> $response
803:	*
804:	* @return TResponse
805:	*
806:	* @throws ClientNotConnectedException
807:	* @throws RequestTimeoutException
808:	* @throws ServerCapabilityNotSupportedException
809:	*/
810:	private function dispatch(
811:	JsonRpcRequest $request,
812:	string $response,
813:	?SendContext $context,
814:	?RequestDeadline $deadline,
815:	): JsonRpcResultResponse {
816:	try {
817:	try {
818:	return $this->exchange($request, $response, $context, $deadline);
819:	} catch (RemoteCallFailedException $e) {
820:	// SEP-2575: a server that rejects the requested version names the ones it accepts, and the
821:	// client SHOULD retry with one of them. The retry is not itself retried.
822:	$retry = $this->renegotiateProtocolVersion($request, $e);
823:
824:	if (null === $retry) {
825:	throw $e;
826:	}
827:
828:	$this->logger->info(
829:	'Retrying request {id} as {retry}: the server does not support {requested}.',
830:	['id' => $request->id->id, 'retry' => $retry->id->id, 'requested' => $e->error->message],
831:	);
832:
833:	return $this->exchange($retry, $response, $context, $deadline);
834:	}
835:	} finally {
836:	$deadline?->release();
837:	}
838:	}
839:
840:	/**
841:	* Sends one request and awaits its correlated response.
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 ClientNotConnectedException
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->transport ?? throw new ClientNotConnectedException();
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:	// A failed send leaves the registration with no awaiter and no
877:	// response to correlate, so free the slot before propagating.
878:	$this->outboundRequests->forget($request->id);
879:
880:	throw $e;
881:	}
882:
883:	if (null === $deadline) {
884:	return $future->await();
885:	}
886:
887:	try {
888:	return $future->await($deadline->getCancellation());
889:	} catch (CancelledException $e) {
890:	throw $this->abandon($request, $transport, $deadline, $e);
891:	}
892:	}
893:
894:	/**
895:	* Rebuilds the request under a version the rejection named as supported, or null when the failure is
896:	* not a version rejection or names no version this SDK speaks.
897:	*
898:	* @template TMethod of non-empty-string
899:	*
900:	* @param JsonRpcRequest<TMethod> $request
901:	*
902:	* @return null\|JsonRpcRequest<TMethod>
903:	*/
904:	private function renegotiateProtocolVersion(JsonRpcRequest $request, RemoteCallFailedException $failure): ?JsonRpcRequest
905:	{
906:	$error = $failure->error;
907:
908:	if (! $error instanceof UnsupportedProtocolVersionError) {
909:	return null;
910:	}
911:
912:	$version = self::pickSupportedVersion($error->supported);
913:
914:	if (null === $version) {
915:	return null;
916:	}
917:
918:	$params = $request->params;
919:
920:	if (! $params instanceof RequestParams) {
921:	// A request carrying no typed params carries no `_meta` to restamp either.
922:	return null;
923:	}
924:
925:	$meta = $params->meta->toArray();
926:	$meta[RequestMetaObject::PROTOCOL_VERSION_KEY] = $version;
927:
928:	$fields = $params->toArray();
929:	$fields['_meta'] = $meta;
930:
931:	$envelope = $request->toArray();
932:	$envelope['params'] = $fields;
933:
934:	// A fresh id: the rejected one has already been answered and its slot retired.
935:	$envelope['id'] = $this->mintRequestId()->id;
936:
937:	return $request::fromArray($envelope);
938:	}
939:
940:	/**
941:	* The first version the peer named that this SDK also speaks.
942:	*
943:	* @param list<string> $supported
944:	*/
945:	private static function pickSupportedVersion(array $supported): ?string
946:	{
947:	foreach ($supported as $version) {
948:	if (\in_array($version, ProtocolVersion::SUPPORTED_VERSIONS, true)) {
949:	return $version;
950:	}
951:	}
952:
953:	return null;
954:	}
955:
956:	/**
957:	* Frees the request's slot and tells the peer to stop working on it, then reports the timeout. A
958:	* response arriving after this has no awaiter left and is discarded as an orphan.
959:	*
960:	* @param JsonRpcRequest<non-empty-string> $request
961:	*/
962:	private function abandon(
963:	JsonRpcRequest $request,
964:	TransportInterface $transport,
965:	RequestDeadline $deadline,
966:	CancelledException $cause,
967:	): RequestTimeoutException {
968:	$this->outboundRequests->forget($request->id);
969:	self::abortExchange($transport, $request->id);
970:
971:	try {
972:	$transport->send(new CancelledNotification(
973:	params: new CancelledNotificationParams(requestId: $request->id, reason: 'The request timed out.'),
974:	));
975:	} catch (\Throwable $e) {
976:	// The peer goes on working on a result nobody will read, which the timeout itself survives.
977:	$this->logger->warning(
978:	'Could not tell the server that request {id} was abandoned.',
979:	['id' => $request->id->id, 'exception' => $e],
980:	);
981:	}
982:
983:	return new RequestTimeoutException($request->id, $deadline->elapsed, $cause);
984:	}
985:
986:	/**
987:	* @param null\|float $timeout Overrides the configured idle deadline for one request
988:	*/
989:	private function openDeadline(?float $timeout = null): ?RequestDeadline
990:	{
991:	$timeout ??= $this->requestTimeout;
992:
993:	return null === $timeout ? null : new RequestDeadline($timeout, $this->maxRequestTimeout);
994:	}
995:
996:	/**
997:	* Caches the `x-mcp-header` bindings of every tool whose declarations hold, and drops the rest from the
998:	* listing: the spec has a client exclude a tool it cannot mirror rather than call it unmirrored.
999:	*/
1000:	private function admitMirrorableTools(ListToolsResult $result): ListToolsResult
1001:	{
1002:	$admitted = [];
1003:
1004:	foreach ($result->tools as $tool) {
1005:	$scan = ParameterHeaderScanner::scan($tool->inputSchema);
1006:
1007:	if (! $scan->valid) {
1008:	// A re-listed tool whose declarations no longer hold must not keep mirroring the bindings
1009:	// an earlier listing cached for it.
1010:	unset($this->toolHeaderBindings[$tool->name]);
1011:
1012:	$this->logger->warning(
1013:	'Excluding tool {tool} from the listing: its "x-mcp-header" declarations are invalid.',
1014:	['tool' => $tool->name, 'reason' => $scan->reason],
1015:	);
1016:
1017:	continue;
1018:	}
1019:
1020:	$this->toolHeaderBindings[$tool->name] = $scan->bindings;
1021:	$admitted[] = $tool;
1022:	}
1023:
1024:	return $admitted === $result->tools ? $result : new ListToolsResult(
1025:	tools: $admitted,
1026:	ttlMs: $result->ttlMs,
1027:	cacheScope: $result->cacheScope,
1028:	nextCursor: $result->nextCursor,
1029:	meta: $result->meta,
1030:	);
1031:	}
1032:
1033:	/**
1034:	* The mirrored `Mcp-Param-{Name}` headers for one tool call, empty when the transport does not mirror
1035:	* them or the tool declared none.
1036:	*
1037:	* @param null\|array<string, mixed> $arguments
1038:	*
1039:	* @return array<non-empty-string, string>
1040:	*/
1041:	private function mirrorParameterHeaders(string $name, ?array $arguments): array
1042:	{
1043:	// Only a mirroring transport ever fills the cache, so an empty one already means nothing to mirror.
1044:	return ParameterHeaders::build($this->toolHeaderBindings[$name] ?? [], $arguments ?? []);
1045:	}
1046:
1047:	/**
1048:	* Builds the self-describing `_meta` envelope stamped onto every request.
1049:	*/
1050:	private function stampMeta(?ProgressToken $progressToken = null): RequestMetaObject
1051:	{
1052:	return new RequestMetaObject(
1053:	protocolVersion: $this->protocolVersion,
1054:	clientInfo: $this->clientInfo,
1055:	clientCapabilities: $this->clientCapabilities,
1056:	progressToken: $progressToken,
1057:	);
1058:	}
1059:
1060:	/**
1061:	* @throws ServerCapabilityNotSupportedException
1062:	*/
1063:	private function assertServerSupports(string $method): void
1064:	{
1065:	$capabilities = $this->serverCapabilities;
1066:
1067:	if (null === $capabilities) {
1068:	// Discovery has not run, so there are no advertised capabilities to enforce.
1069:	return;
1070:	}
1071:
1072:	$supported = match ($method) {
1073:	ListToolsRequest::getMethod(), CallToolRequest::getMethod() => null !== $capabilities->tools,
1074:	ListResourcesRequest::getMethod(),
1075:	ListResourceTemplatesRequest::getMethod(),
1076:	ReadResourceRequest::getMethod() => null !== $capabilities->resources,
1077:	ListPromptsRequest::getMethod(), GetPromptRequest::getMethod() => null !== $capabilities->prompts,
1078:	CompleteRequest::getMethod() => null !== $capabilities->completions,
1079:	default => true,
1080:	};
1081:
1082:	if (! $supported) {
1083:	throw new ServerCapabilityNotSupportedException($method);
1084:	}
1085:	}
1086:
1087:	private function mintRequestId(): RequestId
1088:	{
1089:	return new RequestId(id: ($this->requestIdFactory)());
1090:	}
1091:
1092:	private function mintProgressToken(): ProgressToken
1093:	{
1094:	return new ProgressToken(token: ($this->progressTokenFactory)());
1095:	}
1096:	}
1097:

Namespaces

Classes