File Server/Server.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\Server;
15:
16:	use Amp\DeferredFuture;
17:	use Nexus\Mcp\Core\Dispatch\MessageDispatcherInterface;
18:	use Nexus\Mcp\Core\Schema\RequestId;
19:	use Nexus\Mcp\Core\Transport\CancellableTransportInterface;
20:	use Nexus\Mcp\Core\Transport\ReceiveContext;
21:	use Nexus\Mcp\Core\Transport\TransportInterface;
22:	use Nexus\Mcp\Server\Subscription\SubscriptionStoreInterface;
23:	use Psr\Log\LoggerInterface;
24:	use Psr\Log\NullLogger;
25:
26:	/**
27:	* Thin shell that drives a single transport's lifecycle, exposing a blocking
28:	* `run()` and a non-blocking `listen()`.
29:	*/
30:	final readonly class Server
31:	{
32:	public function __construct(
33:	private MessageDispatcherInterface $dispatcher,
34:	private LoggerInterface $logger = new NullLogger(),
35:	private ?SubscriptionStoreInterface $subscriptions = null,
36:	) {
37:	}
38:
39:	/**
40:	* Runs the server on the transport, blocking until it closes. Use for a
41:	* long-lived transport that owns its read loop (stdio). For a request-scoped
42:	* transport the host drives per request, use `listen()` instead.
43:	*/
44:	public function run(TransportInterface $transport): void
45:	{
46:	$this->logger->info('Starting MCP server.');
47:
48:	$deferred = new DeferredFuture();
49:
50:	$this->attachDispatchListeners($transport);
51:
52:	$transport->onClose(static function () use ($deferred): void {
53:	// Transports may emit `close` more than once if an error raises during shutdown
54:	// (e.g. stdin EOF followed by a write failure). Ignore the duplicate.
55:	if ($deferred->isComplete()) {
56:	return;
57:	}
58:
59:	$deferred->complete();
60:	});
61:
62:	$transport->start();
63:	$deferred->getFuture()->await();
64:
65:	$this->logger->info('MCP server stopped.');
66:	}
67:
68:	/**
69:	* Attaches the dispatcher and starts the transport without blocking. Use for
70:	* a request-scoped transport (streamable HTTP mounted in a PSR-15 stack) the
71:	* host drives per request. For a long-lived transport, use `run()` instead.
72:	*/
73:	public function listen(TransportInterface $transport): void
74:	{
75:	$this->attachDispatchListeners($transport);
76:
77:	$transport->start();
78:	}
79:
80:	private function attachDispatchListeners(TransportInterface $transport): void
81:	{
82:	$transport->onMessage(function (array $envelope, ReceiveContext $context) use ($transport): void {
83:	$this->dispatcher->dispatch($envelope, $transport, $context);
84:	});
85:	$transport->onError(function (\Throwable $e): void {
86:	$this->logger->error('Transport error.', ['exception' => $e]);
87:	});
88:	$transport->onDrain(function (): void {
89:	// A held-open `subscriptions/listen` handler never settles on its own, so the streams close
90:	// before the drain waits on the coroutines running them.
91:	$this->subscriptions?->closeAll();
92:	$this->dispatcher->flushPending();
93:	});
94:
95:	if ($transport instanceof CancellableTransportInterface) {
96:	$transport->onCancel(function (RequestId $id): void {
97:	$this->dispatcher->cancelRequest($id);
98:	});
99:	}
100:	}
101:	}
102:

Namespaces

Classes

Interfaces

Enums