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\Transport;
15:
16: use Amp\ByteStream\BufferException;
17: use Amp\Cancellation;
18: use Amp\CancelledException;
19: use Amp\CompositeCancellation;
20: use Amp\DeferredCancellation;
21: use Amp\Http\Client\DelegateHttpClient;
22: use Amp\Http\Client\HttpClientBuilder;
23: use Amp\Http\Client\Request;
24: use Amp\Http\Client\Response;
25: use Nexus\Assert\Assert;
26: use Nexus\Mcp\Core\Dispatch\PendingCoroutines;
27: use Nexus\Mcp\Core\Exception\OutboundRequestFailedException;
28: use Nexus\Mcp\Core\Exception\ResponseTooLargeException;
29: use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
30: use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
31: use Nexus\Mcp\Core\Exception\TransportNotStartedException;
32: use Nexus\Mcp\Core\Http\HttpStatus;
33: use Nexus\Mcp\Core\Http\SseFrameParser;
34: use Nexus\Mcp\Core\Http\StandardHeaders;
35: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
36: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcRequest;
37: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResponse;
38: use Nexus\Mcp\Core\Schema\RequestId;
39: use Nexus\Mcp\Core\Transport\AbortableTransportInterface;
40: use Nexus\Mcp\Core\Transport\ParameterHeaderMirroringInterface;
41: use Nexus\Mcp\Core\Transport\ReceiveContext;
42: use Nexus\Mcp\Core\Transport\SendContext;
43: use Nexus\Mcp\Core\Transport\SubscriptionInterface;
44: use Nexus\Mcp\Core\Transport\TransportEvents;
45: use Nexus\Mcp\Core\Transport\TransportState;
46: use Psr\Log\LoggerInterface;
47: use Psr\Log\NullLogger;
48:
49: use function Amp\async;
50:
51: /**
52: * Streamable HTTP MCP client transport. Every outbound message is its own POST to the MCP endpoint, and the
53: * response, a single JSON object or a request-scoped SSE stream, is emitted back as inbound envelopes.
54: *
55: * @see https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http
56: */
57: final class StreamableHttpClientTransport implements AbortableTransportInterface, ParameterHeaderMirroringInterface
58: {
59: /**
60: * Bytes a single response may occupy before it is abandoned.
61: */
62: public const int DEFAULT_MAX_RESPONSE_BYTES = SseFrameParser::DEFAULT_MAX_FRAME_BYTES;
63:
64: private const string LABEL = 'Streamable HTTP client';
65: private const string ACCEPT = 'application/json, text/event-stream';
66:
67: private readonly DelegateHttpClient $client;
68: private readonly TransportEvents $events;
69: private readonly PendingCoroutines $exchanges;
70: private TransportState $state = TransportState::Idle;
71: private ?DeferredCancellation $lifetime = null;
72:
73: /**
74: * One entry per request POST still in flight, so a caller giving up on a response can stop just that
75: * exchange. Notifications carry no id and nobody awaits them, so they are not tracked.
76: *
77: * @var array<non-empty-string, DeferredCancellation>
78: */
79: private array $inFlight = [];
80:
81: /**
82: * @param non-empty-string $endpoint Absolute URL of the server's MCP endpoint
83: * @param null|DelegateHttpClient $client Defaults to the amphp default client
84: * @param float $readTimeout Seconds a response may stall before the exchange is abandoned.
85: * It must exceed the server's SSE keep-alive interval, or a quiet
86: * long-lived stream is torn down between keep-alives.
87: * @param int $maxResponseBytes Bytes a buffered body, or one SSE frame, may occupy
88: */
89: public function __construct(
90: private readonly string $endpoint,
91: ?DelegateHttpClient $client = null,
92: private readonly LoggerInterface $logger = new NullLogger(),
93: private readonly float $readTimeout = 30.0,
94: private readonly int $maxResponseBytes = self::DEFAULT_MAX_RESPONSE_BYTES,
95: ) {
96: Assert::that($endpoint)->isNonEmptyString(\sprintf('%s endpoint must be a non-empty string.', self::LABEL));
97:
98: if ($readTimeout <= 0.0) {
99: throw new \InvalidArgumentException(\sprintf('%s read timeout must be positive, %s given.', self::LABEL, $readTimeout));
100: }
101:
102: Assert::that($maxResponseBytes)->isPositiveInt(
103: \sprintf('%s maximum response size must be a positive integer, {value} given.', self::LABEL),
104: );
105:
106: $this->client = $client ?? HttpClientBuilder::buildDefault();
107: $this->events = new TransportEvents();
108: $this->exchanges = new PendingCoroutines();
109: }
110:
111: #[\Override]
112: public function start(): void
113: {
114: match ($this->state) {
115: TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
116: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
117: TransportState::Idle => null,
118: };
119:
120: $this->state = TransportState::Running;
121: $this->lifetime = new DeferredCancellation();
122: $this->logger->info('{label} transport started. Endpoint: {endpoint}.', ['label' => self::LABEL, 'endpoint' => $this->endpoint]);
123: }
124:
125: #[\Override]
126: public function send(JsonRpcMessage $message, ?SendContext $context = null): void
127: {
128: $lifetime = match ($this->state) {
129: TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
130: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
131: TransportState::Running => $this->lifetime,
132: };
133: \assert($lifetime instanceof DeferredCancellation);
134:
135: if ($message instanceof JsonRpcResponse) {
136: // A POST body must be a request or a notification: the spec forbids a client from sending
137: // JSON-RPC responses at all, so there is no legal envelope to POST here.
138: $this->logger->warning('{label} transport dropped an outbound response, which a client must not send.', ['label' => self::LABEL]);
139:
140: return;
141: }
142:
143: $headers = $context->headers ?? [];
144:
145: // Only a request has a caller awaiting a response, so only a request names one to fail.
146: $requestId = $message instanceof JsonRpcRequest ? $message->id : null;
147: $cancellation = $lifetime->getCancellation();
148: $held = null;
149:
150: if (null !== $requestId) {
151: $abort = new DeferredCancellation();
152: $this->inFlight[self::buildKey($requestId)] = $abort;
153: $held = $abort;
154:
155: // Composed rather than replaced: a close still stops every exchange, and `abort()` reaches
156: // only this one.
157: $cancellation = new CompositeCancellation($cancellation, $abort->getCancellation());
158: }
159:
160: // The POST runs detached so a caller awaiting the correlated response is not the thing driving it.
161: $this->exchanges->track(async(function () use ($message, $headers, $cancellation, $requestId, $held): void {
162: try {
163: $this->exchange($message, $headers, $cancellation);
164: } catch (CancelledException) {
165: // Either the transport closed or the caller abandoned this request. Neither is a fault:
166: // the protocol layer learns of a close from the close signal, and it is what asked for
167: // the abort.
168: } catch (\Throwable $e) {
169: try {
170: $this->events->emitError(null === $requestId ? $e : new OutboundRequestFailedException($requestId, $e));
171: } catch (\Throwable) {
172: // A listener that throws must not cost this exchange its release, and an error channel
173: // that just failed is no place to report its own failure.
174: }
175: }
176:
177: // Only its own entry: an id already re-sent belongs to a later exchange, and evicting that
178: // one would cancel it through the destructor. Leaving an entry behind instead would grow the
179: // map for the transport's life.
180: if (null !== $requestId) {
181: $key = self::buildKey($requestId);
182:
183: if (($this->inFlight[$key] ?? null) === $held) {
184: unset($this->inFlight[$key]);
185: }
186: }
187: }));
188: }
189:
190: #[\Override]
191: public function close(): void
192: {
193: if (TransportState::Closed === $this->state) {
194: return;
195: }
196:
197: // Draining runs before the state flips so a listener still settling an exchange can send its last
198: // message. Marking the transport closed first would answer that send with a closed-transport throw.
199: try {
200: $this->events->emitDrain();
201: } finally {
202: $this->state = TransportState::Closed;
203:
204: // Closing the response stream is itself the cancellation signal, so shutdown aborts the in-flight
205: // POSTs. Awaiting them first would hang on a `subscriptions/listen` stream, which never ends.
206: // The reference is kept: releasing it would cancel through the destructor instead, leaving the
207: // shutdown to depend on refcounting rather than on this call.
208: $this->lifetime?->cancel();
209: $this->exchanges->flushPending();
210: $this->events->emitClose();
211: }
212: }
213:
214: #[\Override]
215: public function onMessage(\Closure $listener): SubscriptionInterface
216: {
217: return $this->events->onMessage($listener);
218: }
219:
220: #[\Override]
221: public function onError(\Closure $listener): SubscriptionInterface
222: {
223: return $this->events->onError($listener);
224: }
225:
226: #[\Override]
227: public function onDrain(\Closure $listener): SubscriptionInterface
228: {
229: return $this->events->onDrain($listener);
230: }
231:
232: #[\Override]
233: public function onClose(\Closure $listener): SubscriptionInterface
234: {
235: return $this->events->onClose($listener);
236: }
237:
238: #[\Override]
239: public function abort(RequestId $id): void
240: {
241: // Cancelling settles the exchange's own cancellation, which unwinds the POST and, on a streaming
242: // answer, the read loop with it. The entry itself is released where every exchange releases it,
243: // once the unwinding reaches the end of the coroutine.
244: ($this->inFlight[self::buildKey($id)] ?? null)?->cancel();
245: }
246:
247: /**
248: * @return non-empty-string
249: */
250: private static function buildKey(RequestId $id): string
251: {
252: return \sprintf('"id":%s', var_export($id->id, true));
253: }
254:
255: /**
256: * POSTs one message and emits whatever the server answers with.
257: *
258: * @param array<non-empty-string, string> $headers Mirrored parameter headers the protocol layer computed
259: */
260: private function exchange(JsonRpcMessage $message, array $headers, Cancellation $cancellation): void
261: {
262: $response = $this->client->request($this->buildRequest($message, $headers), $cancellation);
263:
264: if ($response->getStatus() === HttpStatus::Accepted->value) {
265: // The server accepted a notification. There is no body to correlate.
266: return;
267: }
268:
269: if (self::isEventStream($response)) {
270: $this->readStream($response, $cancellation);
271:
272: return;
273: }
274:
275: try {
276: $payload = $response->getBody()->buffer($cancellation, $this->maxResponseBytes);
277: } catch (BufferException $e) {
278: throw new ResponseTooLargeException($this->maxResponseBytes, $e);
279: }
280:
281: $this->events->emitMessage(self::decode($payload), new ReceiveContext());
282: }
283:
284: /**
285: * @param array<non-empty-string, string> $headers
286: */
287: private function buildRequest(JsonRpcMessage $message, array $headers): Request
288: {
289: $request = new Request($this->endpoint, 'POST', json_encode($message, \JSON_THROW_ON_ERROR | \JSON_UNESCAPED_SLASHES | \JSON_UNESCAPED_UNICODE));
290:
291: // `setHeaders()` replaces the whole bag rather than merging, so every header goes in one call.
292: $request->setHeaders([
293: 'Content-Type' => 'application/json',
294: 'Accept' => self::ACCEPT,
295: ...$headers,
296: ...StandardHeaders::build($message->toArray()),
297: ]);
298:
299: // A request-scoped stream lives as long as the server keeps it open, so only a stall may end it.
300: $request->setTransferTimeout(0.0);
301: $request->setInactivityTimeout($this->readTimeout);
302:
303: return $request;
304: }
305:
306: /**
307: * Emits each frame of an SSE response as it arrives, until the server ends the stream.
308: */
309: private function readStream(Response $response, Cancellation $cancellation): void
310: {
311: $parser = new SseFrameParser($this->maxResponseBytes);
312: $body = $response->getBody();
313: $chunk = $body->read($cancellation);
314:
315: // A null chunk is the server closing the stream, which ends the exchange.
316: while (null !== $chunk) {
317: foreach ($parser->feed($chunk) as $frame) {
318: try {
319: $envelope = self::decode($frame->data);
320: } catch (\InvalidArgumentException|\JsonException $e) {
321: // One unreadable frame does not end the stream: a later frame may still carry the
322: // response, so the exchange reads on rather than failing its caller here. Only the
323: // decode is guarded, so a listener fault stays a fault rather than an unreadable frame.
324: $this->events->emitError($e);
325:
326: continue;
327: }
328:
329: $this->events->emitMessage($envelope, new ReceiveContext());
330: }
331:
332: $chunk = $body->read($cancellation);
333: }
334: }
335:
336: /**
337: * Decodes one JSON-RPC envelope.
338: *
339: * @return array<string, mixed>
340: *
341: * @throws \InvalidArgumentException
342: * @throws \JsonException
343: */
344: private static function decode(string $payload): array
345: {
346: $envelope = json_decode($payload, associative: true, flags: \JSON_THROW_ON_ERROR);
347: Assert::that($envelope)->isMap(\sprintf('%s received a payload that is not a JSON-RPC envelope.', self::LABEL));
348:
349: return $envelope;
350: }
351:
352: private static function isEventStream(Response $response): bool
353: {
354: // `Content-Type` carries one media type, so the type is the head of the value. A match anywhere
355: // else in it belongs to a parameter, not to the type.
356: return str_starts_with(strtolower($response->getHeader('Content-Type') ?? ''), 'text/event-stream');
357: }
358: }
359:
Nexus MCP SDK API documentation generated by ApiGen dev-master