File Server/Dispatch/ServerMessageDispatcher.php

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\Dispatch;
15:
16:	use Amp\Cancellation;
17:	use Amp\NullCancellation;
18:	use Nexus\Mcp\Core\Dispatch\MessageDispatcherInterface;
19:	use Nexus\Mcp\Core\Dispatch\PendingCoroutines;
20:	use Nexus\Mcp\Core\Dispatch\PendingInboundRequests;
21:	use Nexus\Mcp\Core\Dispatch\RequestBoundSender;
22:	use Nexus\Mcp\Core\Dispatch\ResponseSender;
23:	use Nexus\Mcp\Core\Exception\AbstractJsonRpcProtocolException;
24:	use Nexus\Mcp\Core\Exception\DuplicateInboundRequestIdException;
25:	use Nexus\Mcp\Core\Exception\MethodMisroutedException;
26:	use Nexus\Mcp\Core\Exception\MethodNotFoundException;
27:	use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
28:	use Nexus\Mcp\Core\Handler\HandlerRegistry;
29:	use Nexus\Mcp\Core\Handler\NotificationHandlerInterface;
30:	use Nexus\Mcp\Core\Handler\RequestHandlerInterface;
31:	use Nexus\Mcp\Core\JsonRpc\JsonRpcMessageParser;
32:	use Nexus\Mcp\Core\JsonRpc\ResultResponseFactory;
33:	use Nexus\Mcp\Core\Schema\Error\InternalError;
34:	use Nexus\Mcp\Core\Schema\Error\UnsupportedProtocolVersionError;
35:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcErrorResponse;
36:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
37:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcRequest;
38:	use Nexus\Mcp\Core\Schema\ProtocolVersion;
39:	use Nexus\Mcp\Core\Schema\Request\ClientRequest;
40:	use Nexus\Mcp\Core\Schema\RequestParams;
41:	use Nexus\Mcp\Core\Schema\Result;
42:	use Nexus\Mcp\Core\Transport\TransportInterface;
43:	use Nexus\Mcp\Server\ServerContext;
44:	use Psr\Log\LoggerInterface;
45:	use Psr\Log\NullLogger;
46:
47:	use function Amp\async;
48:
49:	/**
50:	* Server-side per-envelope inbound dispatch. Parses, classifies, resolves a handler,
51:	* spawns a coroutine to run it, and sends the response (or error) on the transport.
52:	*/
53:	final readonly class ServerMessageDispatcher implements MessageDispatcherInterface
54:	{
55:	private PendingCoroutines $coroutines;
56:	private PendingInboundRequests $inboundRequests;
57:	private ResponseSender $responseSender;
58:
59:	/**
60:	* @param HandlerRegistry<RequestHandlerInterface<non-empty-string, Result, ServerContext>> $requestHandlers
61:	* @param HandlerRegistry<NotificationHandlerInterface<non-empty-string>> $notificationHandlers
62:	*/
63:	public function __construct(
64:	private HandlerRegistry $requestHandlers,
65:	private HandlerRegistry $notificationHandlers,
66:	private LoggerInterface $logger = new NullLogger(),
67:	private JsonRpcMessageParser $parser = new JsonRpcMessageParser(),
68:	private Cancellation $cancellation = new NullCancellation(),
69:	) {
70:	$this->coroutines = new PendingCoroutines();
71:	$this->inboundRequests = new PendingInboundRequests();
72:	$this->responseSender = new ResponseSender($this->logger);
73:	}
74:
75:	#[\Override]
76:	public function flushPending(): void
77:	{
78:	$this->coroutines->flushPending();
79:	}
80:
81:	/**
82:	* @param array<string, mixed> $envelope
83:	*/
84:	#[\Override]
85:	public function dispatch(array $envelope, TransportInterface $transport): void
86:	{
87:	if (\array_key_exists('result', $envelope) \|\| \array_key_exists('error', $envelope)) {
88:	$this->discardResponseEnvelope($envelope);
89:
90:	return;
91:	}
92:
93:	$isNotification = ! \array_key_exists('id', $envelope);
94:
95:	try {
96:	$message = $this->parser->parse($envelope);
97:	} catch (MethodMisroutedException $e) {
98:	$this->logger->warning(
99:	'Rejecting envelope whose method was sent under the wrong JSON-RPC shape.',
100:	['envelope' => $envelope, 'exception' => $e],
101:	);
102:
103:	if (! $isNotification) {
104:	// Envelope carried an id but the method is a notification method.
105:	// JSON-RPC 2.0 §4.1 forbids responses to notifications. Drop silently.
106:	return;
107:	}
108:
109:	// Envelope omitted the id but the method is a request method.
110:	// §5 null-id fallback. Respond so the peer can fix the malformed request.
111:	$this->responseSender->send($transport, ResponseSender::buildErrorResponse($e, null), 'misrouted');
112:
113:	return;
114:	} catch (AbstractJsonRpcProtocolException $e) {
115:	if ($isNotification) {
116:	$this->logger->info(
117:	'Dropping malformed notification (JSON-RPC 2.0 §4.1 forbids responses to notifications).',
118:	['envelope' => $envelope, 'exception' => $e],
119:	);
120:
121:	return;
122:	}
123:
124:	$this->responseSender->send($transport, ResponseSender::buildErrorResponse($e, null), 'parse-error');
125:
126:	return;
127:	}
128:
129:	if ($message instanceof JsonRpcRequest) {
130:	$this->dispatchRequest($message, $transport);
131:	} elseif ($message instanceof JsonRpcNotification) {
132:	$this->dispatchNotification($message);
133:	}
134:	}
135:
136:	/**
137:	* @param array<string, mixed> $envelope
138:	*/
139:	private function discardResponseEnvelope(array $envelope): void
140:	{
141:	$this->logger->warning(
142:	'Discarding response envelope (server has no outbound-request correlation).',
143:	['envelope' => $envelope],
144:	);
145:	}
146:
147:	/**
148:	* @param JsonRpcRequest<non-empty-string> $request
149:	*/
150:	private function dispatchRequest(JsonRpcRequest $request, TransportInterface $transport): void
151:	{
152:	$method = $request::getMethod();
153:
154:	if (! $this->inboundRequests->claim($request->id)) {
155:	$exception = new DuplicateInboundRequestIdException($request->id);
156:	$this->responseSender->send($transport, ResponseSender::buildErrorResponse($exception, $request->id), $method);
157:
158:	return;
159:	}
160:
161:	$this->coroutines->track(async(function () use ($request, $transport, $method): void {
162:	try {
163:	try {
164:	if (! $request instanceof ClientRequest) {
165:	// The server services only ClientRequest methods. A server-to-client method is
166:	// one the server does not implement, so it answers MethodNotFound.
167:	throw new MethodNotFoundException($method, $request->id);
168:	}
169:
170:	// A ClientRequest always carries the heavy RequestParams (the required _meta).
171:	\assert($request->params instanceof RequestParams);
172:
173:	$requestedVersion = $request->params->meta->protocolVersion->version;
174:
175:	if (! \in_array($requestedVersion, ProtocolVersion::SUPPORTED_VERSIONS, true)) {
176:	$this->responseSender->send($transport, new JsonRpcErrorResponse(
177:	id: $request->id,
178:	error: new UnsupportedProtocolVersionError(
179:	requested: $requestedVersion,
180:	supported: ProtocolVersion::SUPPORTED_VERSIONS,
181:	),
182:	), $method);
183:
184:	return;
185:	}
186:
187:	$handler = $this->requestHandlers->get($method)
188:	?? throw new MethodNotFoundException($method, $request->id);
189:
190:	$sender = new RequestBoundSender($transport, $request->id);
191:	$context = new ServerContext(
192:	$request->id,
193:	$this->cancellation,
194:	$request->params->meta,
195:	$sender,
196:	);
197:	$result = $handler->handle($request, $context);
198:	$response = ResultResponseFactory::wrap($request, $result);
199:	} catch (TransportAlreadyClosedException $e) {
200:	$this->responseSender->logSkippedDelivery($method, $e);
201:
202:	return;
203:	} catch (AbstractJsonRpcProtocolException $e) {
204:	$this->responseSender->send($transport, ResponseSender::buildErrorResponse($e, $request->id), $method);
205:
206:	return;
207:	} catch (\Throwable $e) {
208:	$this->logger->error(
209:	'Uncaught request handler exception.',
210:	['method' => $method, 'exception' => $e],
211:	);
212:	$this->responseSender->send($transport, new JsonRpcErrorResponse(
213:	id: $request->id,
214:	error: new InternalError(message: InternalError::DEFAULT_MESSAGE),
215:	), $method);
216:
217:	return;
218:	}
219:
220:	$this->responseSender->send($transport, $response, $method);
221:	} finally {
222:	$this->inboundRequests->release($request->id);
223:	}
224:	}));
225:	}
226:
227:	/**
228:	* @param JsonRpcNotification<non-empty-string> $notification
229:	*/
230:	private function dispatchNotification(JsonRpcNotification $notification): void
231:	{
232:	$method = $notification::getMethod();
233:
234:	$handler = $this->notificationHandlers->get($method);
235:
236:	if (null === $handler) {
237:	return;
238:	}
239:
240:	$this->coroutines->track(async(function () use ($handler, $notification, $method): void {
241:	try {
242:	$handler->handle($notification);
243:	} catch (\Throwable $e) {
244:	// Notifications carry no response per JSON-RPC 2.0 §4.1. Failure is logged only.
245:	$this->logger->error(
246:	'Uncaught notification handler exception.',
247:	['method' => $method, 'exception' => $e],
248:	);
249:	}
250:	}));
251:	}
252:	}
253:

Namespaces

Classes