File Server/Transport/StreamableHttpServerTransport.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\Transport;
15:
16:	use Amp\DeferredFuture;
17:	use Nexus\Assert\Assert;
18:	use Nexus\Mcp\Core\Auth\VerifiedAccessToken;
19:	use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
20:	use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
21:	use Nexus\Mcp\Core\Exception\TransportNotStartedException;
22:	use Nexus\Mcp\Core\Http\HttpStatus;
23:	use Nexus\Mcp\Core\Http\HttpStatusResolver;
24:	use Nexus\Mcp\Core\Http\StandardHeaders;
25:	use Nexus\Mcp\Core\JsonRpc\EnvelopeRequestId;
26:	use Nexus\Mcp\Core\Schema\Error;
27:	use Nexus\Mcp\Core\Schema\Error\InternalError;
28:	use Nexus\Mcp\Core\Schema\Error\InvalidRequestError;
29:	use Nexus\Mcp\Core\Schema\Error\ParseError;
30:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcErrorResponse;
31:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
32:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
33:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResultResponse;
34:	use Nexus\Mcp\Core\Schema\Notification\CancelledNotification;
35:	use Nexus\Mcp\Core\Schema\Request\SubscriptionsListenRequest;
36:	use Nexus\Mcp\Core\Schema\RequestId;
37:	use Nexus\Mcp\Core\Transport\CancellableTransportInterface;
38:	use Nexus\Mcp\Core\Transport\ReceiveContext;
39:	use Nexus\Mcp\Core\Transport\SendContext;
40:	use Nexus\Mcp\Core\Transport\Subscription;
41:	use Nexus\Mcp\Core\Transport\SubscriptionInterface;
42:	use Nexus\Mcp\Core\Transport\TransportEvents;
43:	use Nexus\Mcp\Core\Transport\TransportState;
44:	use Nexus\Mcp\Server\Transport\Http\ResponseMode;
45:	use Nexus\Mcp\Server\Transport\Http\SseResponseStream;
46:	use Psr\Http\Message\ResponseFactoryInterface;
47:	use Psr\Http\Message\ResponseInterface;
48:	use Psr\Http\Message\ServerRequestInterface;
49:	use Psr\Http\Message\StreamFactoryInterface;
50:	use Psr\Http\Server\RequestHandlerInterface;
51:	use Psr\Log\LoggerInterface;
52:	use Psr\Log\NullLogger;
53:
54:	/**
55:	* Stateless Streamable HTTP server transport, and the PSR-15 request handler for the MCP endpoint.
56:	*
57:	* @see https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http
58:	*/
59:	final class StreamableHttpServerTransport implements CancellableTransportInterface, RequestHandlerInterface
60:	{
61:	private TransportState $state = TransportState::Idle;
62:	private readonly TransportEvents $events;
63:
64:	/**
65:	* The last transport-internal request id minted. Ids ascend and are never reused for the process's
66:	* lifetime, so a retired sink's id cannot reach a later request while the handler that still holds it
67:	* runs on.
68:	*/
69:	private int $lastRequestId = 0;
70:
71:	/**
72:	* In-flight requests, keyed by the transport-internal id emitted to the dispatcher. On the buffered path
73:	* the `buffered` deferred carries the response `handle()` awaits. Once `stream` is set the sink streams
74:	* SSE frames instead and the deferred goes unused.
75:	*
76:	* @var array<int, array{
77:	* clientId: int\|non-empty-string,
78:	* buffered: DeferredFuture<ResponseInterface>,
79:	* stream: null\|SseResponseStream,
80:	* }>
81:	*/
82:	private array $sinks = [];
83:
84:	/**
85:	* @var array<int, \Closure(RequestId): void>
86:	*/
87:	private array $cancelListeners = [];
88:
89:	public function __construct(
90:	private readonly ResponseFactoryInterface $responseFactory,
91:	private readonly StreamFactoryInterface $streamFactory,
92:	private readonly LoggerInterface $logger = new NullLogger(),
93:	private readonly ResponseMode $responseMode = ResponseMode::Auto,
94:	private readonly float $keepAliveInterval = 15.0,
95:	) {
96:	if ($this->keepAliveInterval <= 0.0) {
97:	throw new \InvalidArgumentException(\sprintf('The SSE keep-alive interval must be positive, %s given.', $this->keepAliveInterval));
98:	}
99:
100:	$this->events = new TransportEvents();
101:	}
102:
103:	/**
104:	* Handles one HTTP request against the MCP endpoint, returning the response to write back.
105:	*/
106:	#[\Override]
107:	public function handle(ServerRequestInterface $request): ResponseInterface
108:	{
109:	if ($request->getMethod() !== 'POST') {
110:	return $this->responseFactory->createResponse(HttpStatus::MethodNotAllowed->value)->withHeader('Allow', 'POST');
111:	}
112:
113:	if (TransportState::Running !== $this->state) {
114:	// Nothing is listening, so no dispatch would ever resolve the request. Fail fast rather than
115:	// suspending on a response that cannot arrive.
116:	return $this->buildErrorResponse(
117:	new InternalError(message: 'The MCP endpoint is not accepting requests.'),
118:	HttpStatus::ServiceUnavailable->value,
119:	);
120:	}
121:
122:	if (! self::acceptsRequiredContentTypes($request)) {
123:	// The client must accept both media types so the server may answer with JSON or an SSE stream.
124:	return $this->responseFactory->createResponse(HttpStatus::NotAcceptable->value);
125:	}
126:
127:	try {
128:	$envelope = json_decode((string) $request->getBody(), associative: true, flags: \JSON_THROW_ON_ERROR);
129:	} catch (\JsonException) {
130:	// An empty or otherwise undecodable body.
131:	return $this->buildErrorResponse(new ParseError(message: ParseError::DEFAULT_MESSAGE));
132:	}
133:
134:	try {
135:	Assert::that($envelope)->isMap('JSON-RPC envelope must be a JSON object, {type} given.');
136:	} catch (\InvalidArgumentException) {
137:	// Valid JSON that is not an object (a scalar, or a JSON array such as a removed batch).
138:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
139:	}
140:
141:	if (\array_key_exists('result', $envelope) \|\| \array_key_exists('error', $envelope)) {
142:	// The body must be a request or notification. A response is not a valid client-to-server message,
143:	// and the dispatcher discards responses without replying, so admitting one would hang the POST.
144:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
145:	}
146:
147:	if (! \array_key_exists('id', $envelope)) {
148:	if (! self::isAcceptableNotification($envelope)) {
149:	// The server cannot accept a malformed notification, so it answers an HTTP error, not 202.
150:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
151:	}
152:
153:	if (CancelledNotification::getMethod() === ($envelope['method'] ?? null)) {
154:	// Closing the response stream is the cancellation signal here, so this message is neither
155:	// required nor expected. Its `params.requestId` names the client's own id space, which no
156:	// sink is keyed by, so honouring it would cancel whichever request holds that internal id.
157:	$this->logger->debug(
158:	'Ignoring a client cancellation notification: the response stream is the signal on this transport.',
159:	);
160:
161:	return $this->responseFactory->createResponse(HttpStatus::Accepted->value);
162:	}
163:
164:	$this->events->emitMessage($envelope, self::buildReceiveContext($request));
165:
166:	return $this->responseFactory->createResponse(HttpStatus::Accepted->value);
167:	}
168:
169:	$requestId = EnvelopeRequestId::recover($envelope);
170:
171:	if (null === $requestId) {
172:	// MCP narrows the request id to int\|non-empty-string.
173:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
174:	}
175:
176:	$clientId = $requestId->id;
177:
178:	$mismatch = StandardHeaders::validate(self::readHeaders($request), $envelope);
179:
180:	if (null !== $mismatch) {
181:	return $this->buildErrorResponse($mismatch);
182:	}
183:
184:	// A listen request is answered only when its stream ends, so the buffered path would hold the POST
185:	// open with nowhere to push the acknowledgement. It streams whatever the configured mode says.
186:	$streams = ResponseMode::Sse === $this->responseMode
187:	\|\| SubscriptionsListenRequest::getMethod() === ($envelope['method'] ?? null);
188:
189:	return $streams
190:	? $this->dispatchStreaming($envelope, $clientId, $request)
191:	: $this->dispatchBuffered($envelope, $clientId, $request);
192:	}
193:
194:	#[\Override]
195:	public function start(): void
196:	{
197:	match ($this->state) {
198:	TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
199:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
200:	TransportState::Idle => null,
201:	};
202:
203:	$this->state = TransportState::Running;
204:	}
205:
206:	#[\Override]
207:	public function send(JsonRpcMessage $message, ?SendContext $context = null): void
208:	{
209:	match ($this->state) {
210:	TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
211:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
212:	TransportState::Running => null,
213:	};
214:
215:	if ($message instanceof JsonRpcResultResponse \|\| $message instanceof JsonRpcErrorResponse) {
216:	$this->routeResponse($message, $context);
217:
218:	return;
219:	}
220:
221:	if ($message instanceof JsonRpcNotification) {
222:	$this->routeNotification($message, $context);
223:
224:	return;
225:	}
226:
227:	// The server issues no client-bound requests over Streamable HTTP.
228:	$this->logger->warning('Dropping an unexpected server-initiated request.');
229:	}
230:
231:	#[\Override]
232:	public function close(): void
233:	{
234:	if (TransportState::Closed === $this->state) {
235:	return;
236:	}
237:
238:	// Draining lets any in-flight dispatch coroutine send its response and resolve the awaiting request
239:	// before the transport is marked closed.
240:	try {
241:	$this->events->emitDrain();
242:	} finally {
243:	$this->state = TransportState::Closed;
244:	$this->events->emitClose();
245:	}
246:	}
247:
248:	#[\Override]
249:	public function onMessage(\Closure $listener): SubscriptionInterface
250:	{
251:	return $this->events->onMessage($listener);
252:	}
253:
254:	#[\Override]
255:	public function onError(\Closure $listener): SubscriptionInterface
256:	{
257:	return $this->events->onError($listener);
258:	}
259:
260:	#[\Override]
261:	public function onDrain(\Closure $listener): SubscriptionInterface
262:	{
263:	return $this->events->onDrain($listener);
264:	}
265:
266:	#[\Override]
267:	public function onCancel(\Closure $listener): SubscriptionInterface
268:	{
269:	$id = spl_object_id($listener);
270:	$this->cancelListeners[$id] = $listener;
271:
272:	return new Subscription(function () use ($id): void {
273:	unset($this->cancelListeners[$id]);
274:	});
275:	}
276:
277:	#[\Override]
278:	public function onClose(\Closure $listener): SubscriptionInterface
279:	{
280:	return $this->events->onClose($listener);
281:	}
282:
283:	/**
284:	* The client MUST accept both media types (`Accept: application/json, text/event-stream`) so the server
285:	* is free to answer with a buffered JSON object or an SSE stream.
286:	*/
287:	private static function acceptsRequiredContentTypes(ServerRequestInterface $request): bool
288:	{
289:	$accept = strtolower($request->getHeaderLine('Accept'));
290:
291:	return str_contains($accept, 'application/json') && str_contains($accept, 'text/event-stream');
292:	}
293:
294:	/**
295:	* A notification receives no dispatcher reply, so the transport gates its acceptance itself: a
296:	* well-formed JSON-RPC 2.0 notification carries a non-empty string method.
297:	*
298:	* @param array<string, mixed> $envelope
299:	*/
300:	private static function isAcceptableNotification(array $envelope): bool
301:	{
302:	$method = $envelope['method'] ?? null;
303:
304:	return JsonRpcMessage::JSONRPC_VERSION === ($envelope['jsonrpc'] ?? null)
305:	&& \is_string($method)
306:	&& '' !== $method;
307:	}
308:
309:	/**
310:	* Buffers the request, resolving a JSON response when the final response arrives. Under `Auto`, the
311:	* first progress notification upgrades the awaited response to an SSE stream instead.
312:	*
313:	* @param array<string, mixed> $envelope
314:	* @param int\|non-empty-string $clientId
315:	*/
316:	private function dispatchBuffered(array $envelope, int\|string $clientId, ServerRequestInterface $request): ResponseInterface
317:	{
318:	/** @var DeferredFuture<ResponseInterface> $deferred */
319:	$deferred = new DeferredFuture();
320:
321:	$internalId = ++$this->lastRequestId;
322:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $deferred, 'stream' => null];
323:
324:	$envelope['id'] = $internalId;
325:	$this->events->emitMessage($envelope, self::buildReceiveContext($request, $clientId));
326:
327:	return $deferred->getFuture()->await();
328:	}
329:
330:	/**
331:	* Answers immediately with an SSE stream the dispatch coroutine writes progress frames and the final
332:	* response to.
333:	*
334:	* @param array<string, mixed> $envelope
335:	* @param int\|non-empty-string $clientId
336:	*/
337:	private function dispatchStreaming(array $envelope, int\|string $clientId, ServerRequestInterface $request): ResponseInterface
338:	{
339:	// The streaming response is returned directly rather than through the sink's deferred, which only an
340:	// `Auto` upgrade from the buffered path ever completes.
341:	/** @var DeferredFuture<ResponseInterface> $unused */
342:	$unused = new DeferredFuture();
343:	$internalId = ++$this->lastRequestId;
344:	$stream = new SseResponseStream($this->keepAliveInterval, fn(): null => $this->releaseStream($internalId));
345:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $unused, 'stream' => $stream];
346:
347:	$envelope['id'] = $internalId;
348:	$this->events->emitMessage($envelope, self::buildReceiveContext($request, $clientId));
349:
350:	return $this->buildSseResponse($stream);
351:	}
352:
353:	/**
354:	* Carries the HTTP request, and the token a bearer-authentication stage validated it with, to handlers.
355:	*
356:	* @param null\|int\|non-empty-string $clientId
357:	*/
358:	private static function buildReceiveContext(ServerRequestInterface $request, null\|int\|string $clientId = null): ReceiveContext
359:	{
360:	$token = $request->getAttribute(VerifiedAccessToken::REQUEST_ATTRIBUTE);
361:
362:	// Dispatch runs under a transport-internal id, so anything the handler puts back on the message
363:	// (a subscription id, say) has to name the id the client actually sent.
364:	return new ReceiveContext(
365:	$request,
366:	$token instanceof VerifiedAccessToken ? $token : null,
367:	null === $clientId ? null : new RequestId(id: $clientId),
368:	);
369:	}
370:
371:	private function routeResponse(JsonRpcErrorResponse\|JsonRpcResultResponse $message, ?SendContext $context): void
372:	{
373:	$id = $message->id;
374:
375:	if (! $id instanceof RequestId) {
376:	$this->logger->warning('Discarding a response that carries no id to correlate.');
377:
378:	return;
379:	}
380:
381:	$internalId = $id->id;
382:
383:	if (! \is_int($internalId) \|\| ! \array_key_exists($internalId, $this->sinks)) {
384:	$this->logger->warning('Discarding an orphan response with no in-flight request.');
385:
386:	return;
387:	}
388:
389:	$sink = $this->sinks[$internalId];
390:	$envelope = $message->jsonSerialize();
391:	$envelope['id'] = $sink['clientId'];
392:	$stream = $sink['stream'];
393:
394:	if (null !== $stream) {
395:	$stream->push(self::frame($envelope));
396:	$this->endStream($internalId, $stream);
397:	} else {
398:	$fromHandler = null !== $context && $context->fromHandler;
399:	$sink['buffered']->complete($this->buildJsonResponse($envelope, self::resolveStatus($message, $fromHandler)));
400:	unset($this->sinks[$internalId]);
401:	}
402:	}
403:
404:	/**
405:	* @param JsonRpcNotification<non-empty-string> $notification
406:	*/
407:	private function routeNotification(JsonRpcNotification $notification, ?SendContext $context): void
408:	{
409:	$related = $context?->relatedRequestId;
410:
411:	if (! $related instanceof RequestId) {
412:	$this->logger->debug('Dropping a notification with no related request to stream it to.');
413:
414:	return;
415:	}
416:
417:	$internalId = $related->id;
418:
419:	if (! \is_int($internalId) \|\| ! \array_key_exists($internalId, $this->sinks)) {
420:	$this->logger->debug('Dropping a notification for a request that is no longer in flight.');
421:
422:	return;
423:	}
424:
425:	$sink = $this->sinks[$internalId];
426:	$stream = $sink['stream'];
427:
428:	if (null !== $stream) {
429:	$stream->push(self::frame($notification->jsonSerialize()));
430:	} elseif (ResponseMode::Auto === $this->responseMode) {
431:	$this->upgradeToStream($internalId, $sink['clientId'], $sink['buffered'], $notification->jsonSerialize());
432:	} else {
433:	// The JSON response mode buffers a single object and has no stream to carry a notification.
434:	$this->logger->debug('Dropping a notification: the JSON response mode cannot stream it.');
435:	}
436:	}
437:
438:	/**
439:	* Lazily promotes a buffered `Auto` request to an SSE stream, replaying the progress notification that
440:	* triggered the upgrade and resolving the awaiting `handle()` with the streaming response.
441:	*
442:	* @param int\|non-empty-string $clientId
443:	* @param DeferredFuture<ResponseInterface> $buffered
444:	* @param array<string, mixed> $envelope
445:	*/
446:	private function upgradeToStream(int $internalId, int\|string $clientId, DeferredFuture $buffered, array $envelope): void
447:	{
448:	$stream = new SseResponseStream($this->keepAliveInterval, fn(): null => $this->releaseStream($internalId));
449:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $buffered, 'stream' => $stream];
450:
451:	$stream->push(self::frame($envelope));
452:	$buffered->complete($this->buildSseResponse($stream));
453:	}
454:
455:	/**
456:	* Ends a stream after its final response frame: signals end-of-body to the reader and retires the sink.
457:	*/
458:	private function endStream(int $internalId, SseResponseStream $stream): void
459:	{
460:	$stream->end();
461:	unset($this->sinks[$internalId]);
462:	}
463:
464:	/**
465:	* Retires a stream whose body the consumer closed (a client disconnect); a no-op once it has ended.
466:	*/
467:	private function releaseStream(int $internalId): null
468:	{
469:	$sink = $this->sinks[$internalId] ?? null;
470:	unset($this->sinks[$internalId]);
471:
472:	if (null === $sink) {
473:	// `endStream()` already retired the sink, so the body closing is the graceful path finishing
474:	// rather than the peer walking away.
475:	return null;
476:	}
477:
478:	$abandoned = new RequestId(id: $internalId);
479:
480:	foreach ($this->cancelListeners as $listener) {
481:	$listener($abandoned);
482:	}
483:
484:	return null;
485:	}
486:
487:	/**
488:	* A result rides HTTP 200. An error's status turns on its origin: a handler-produced error rides 200
489:	* with the JSON-RPC error in the body, while a protocol error carries a real status.
490:	*/
491:	private static function resolveStatus(JsonRpcErrorResponse\|JsonRpcResultResponse $message, bool $fromHandler): int
492:	{
493:	if ($message instanceof JsonRpcResultResponse) {
494:	return HttpStatus::Ok->value;
495:	}
496:
497:	return HttpStatusResolver::resolve($message->error->code, $fromHandler);
498:	}
499:
500:	/**
501:	* @param array<string, mixed> $envelope
502:	*/
503:	private function buildJsonResponse(array $envelope, int $status): ResponseInterface
504:	{
505:	return $this->responseFactory->createResponse($status)
506:	->withHeader('Content-Type', 'application/json')
507:	->withBody($this->streamFactory->createStream(self::encode($envelope)))
508:	;
509:	}
510:
511:	private function buildSseResponse(SseResponseStream $body): ResponseInterface
512:	{
513:	return $this->responseFactory->createResponse(HttpStatus::Ok->value)
514:	->withHeader('Content-Type', 'text/event-stream')
515:	->withHeader('Cache-Control', 'no-cache')
516:	->withHeader('Connection', 'keep-alive')
517:	->withHeader('X-Accel-Buffering', 'no')
518:	->withBody($body)
519:	;
520:	}
521:
522:	/**
523:	* @param null\|int $status The HTTP status to pin, or `null` to derive it from the error's code
524:	*/
525:	private function buildErrorResponse(Error $error, ?int $status = null): ResponseInterface
526:	{
527:	$status ??= HttpStatusResolver::resolve($error->code, fromHandler: false);
528:	$envelope = new JsonRpcErrorResponse(id: null, error: $error)->jsonSerialize();
529:
530:	return $this->responseFactory->createResponse($status)
531:	->withHeader('Content-Type', 'application/json')
532:	->withBody($this->streamFactory->createStream(self::encode($envelope)))
533:	;
534:	}
535:
536:	/**
537:	* Flattens the PSR-7 header bag to the single-value-per-name map the header validator consumes.
538:	*
539:	* @return array<string, string>
540:	*/
541:	private static function readHeaders(ServerRequestInterface $request): array
542:	{
543:	return array_map(
544:	static fn(array $values): string => implode(', ', $values),
545:	$request->getHeaders(),
546:	);
547:	}
548:
549:	/**
550:	* @param array<string, mixed> $envelope
551:	*/
552:	private static function frame(array $envelope): string
553:	{
554:	return \sprintf("event: message\ndata: %s\n\n", self::encode($envelope));
555:	}
556:
557:	/**
558:	* @param array<string, mixed> $envelope
559:	*/
560:	private static function encode(array $envelope): string
561:	{
562:	return json_encode($envelope, \JSON_THROW_ON_ERROR \| \JSON_UNESCAPED_SLASHES \| \JSON_UNESCAPED_UNICODE);
563:	}
564:	}
565:

Namespaces

Classes