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\DeferredFuture;
22: use Amp\Http\Client\DelegateHttpClient;
23: use Amp\Http\Client\HttpClientBuilder;
24: use Amp\Http\Client\Request;
25: use Amp\Http\Client\Response;
26: use Nexus\Assert\Assert;
27: use Nexus\Mcp\Core\Dispatch\PendingCoroutines;
28: use Nexus\Mcp\Core\Exception\OutboundRequestFailedException;
29: use Nexus\Mcp\Core\Exception\ResponseTooLargeException;
30: use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
31: use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
32: use Nexus\Mcp\Core\Exception\TransportNotStartedException;
33: use Nexus\Mcp\Core\Exception\UnexpectedHttpStatusException;
34: use Nexus\Mcp\Core\Http\HttpStatus;
35: use Nexus\Mcp\Core\Http\SseFrameParser;
36: use Nexus\Mcp\Core\Http\StandardHeaders;
37: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
38: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcRequest;
39: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResponse;
40: use Nexus\Mcp\Core\Schema\RequestId;
41: use Nexus\Mcp\Core\Transport\AbortableTransportInterface;
42: use Nexus\Mcp\Core\Transport\ListenerHandleInterface;
43: use Nexus\Mcp\Core\Transport\ParameterHeaderMirroringInterface;
44: use Nexus\Mcp\Core\Transport\ReceiveContext;
45: use Nexus\Mcp\Core\Transport\SendContext;
46: use Nexus\Mcp\Core\Transport\TransportEvents;
47: use Nexus\Mcp\Core\Transport\TransportState;
48: use Psr\Log\LoggerInterface;
49: use Psr\Log\NullLogger;
50:
51: use function Amp\async;
52:
53: /**
54: * Streamable HTTP MCP client transport, one POST per outbound message.
55: *
56: * @see https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http
57: */
58: final class StreamableHttpClientTransport implements AbortableTransportInterface, ParameterHeaderMirroringInterface
59: {
60: public const int DEFAULT_MAX_RESPONSE_BYTES = SseFrameParser::DEFAULT_MAX_FRAME_BYTES;
61:
62: private readonly DelegateHttpClient $client;
63: private readonly TransportEvents $events;
64: private readonly PendingCoroutines $exchanges;
65: private readonly StandardHeaders $standardHeaders;
66: private TransportState $state = TransportState::Idle;
67:
68: /**
69: * True from the first `close()` on, which `state` cannot signal as it stays `Running` across the
70: * drain so a listener may still send.
71: */
72: private bool $closing = false;
73:
74: /**
75: * @var null|\Fiber<mixed, mixed, mixed, mixed> The close owner, `null` while no close is in progress or when {main} owns it
76: */
77: private ?\Fiber $closingFiber = null;
78:
79: /**
80: * @var null|DeferredFuture<null>
81: */
82: private ?DeferredFuture $closeCompletion = null;
83:
84: /**
85: * @var array<int, \Fiber<mixed, mixed, mixed, mixed>> The in-flight exchange fibers, close participants while the flush drains them
86: */
87: private array $exchangeFibers = [];
88:
89: private ?DeferredCancellation $lifetime = null;
90:
91: /**
92: * @var array<non-empty-string, DeferredCancellation>
93: */
94: private array $inFlight = [];
95:
96: /**
97: * @param non-empty-string $endpoint Absolute URL of the server's MCP endpoint
98: * @param null|DelegateHttpClient $client Defaults to the amphp default client
99: * @param float $readTimeout Seconds a response may stall before the exchange is abandoned.
100: * It must exceed the server's SSE keep-alive interval, or a quiet
101: * long-lived stream is torn down between keep-alives.
102: * @param int $maxResponseBytes Bytes a buffered body, or one SSE frame, may occupy
103: */
104: public function __construct(
105: private readonly string $endpoint,
106: ?DelegateHttpClient $client = null,
107: private readonly LoggerInterface $logger = new NullLogger(),
108: private readonly float $readTimeout = 30.0,
109: private readonly int $maxResponseBytes = self::DEFAULT_MAX_RESPONSE_BYTES,
110: ) {
111: Assert::that($endpoint)->isNonEmptyString('Streamable HTTP client endpoint must be a non-empty string.');
112:
113: if ($readTimeout <= 0.0) {
114: throw new \InvalidArgumentException(\sprintf('Streamable HTTP client read timeout must be positive, %s given.', $readTimeout));
115: }
116:
117: Assert::that($maxResponseBytes)->isPositiveInt(
118: 'Streamable HTTP client maximum response size must be a positive integer, {value} given.',
119: );
120:
121: $this->client = $client ?? HttpClientBuilder::buildDefault();
122: $this->events = TransportEvents::create($logger, 'Streamable HTTP client');
123: $this->exchanges = new PendingCoroutines();
124: $this->standardHeaders = new StandardHeaders();
125: }
126:
127: #[\Override]
128: public function start(): void
129: {
130: match ($this->state) {
131: TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
132: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
133: TransportState::Idle => null,
134: };
135:
136: $this->state = TransportState::Running;
137: $this->lifetime = new DeferredCancellation();
138: $this->logger->info('Streamable HTTP client transport started. Endpoint: {endpoint}.', ['endpoint' => $this->endpoint]);
139: }
140:
141: #[\Override]
142: public function send(JsonRpcMessage $message, ?SendContext $context = null): void
143: {
144: match ($this->state) {
145: TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
146: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
147: TransportState::Running => null,
148: };
149:
150: if ($message instanceof JsonRpcResponse) {
151: $this->logger->warning('Streamable HTTP client transport dropped an outbound response, which a client must not send.');
152:
153: return;
154: }
155:
156: $headers = $context->headers ?? [];
157: $requestId = $message instanceof JsonRpcRequest ? $message->id : null;
158:
159: \assert($this->lifetime instanceof DeferredCancellation);
160: $cancellation = $this->lifetime->getCancellation();
161: $held = null;
162:
163: if (null !== $requestId) {
164: $abort = new DeferredCancellation();
165: $this->inFlight[$this->buildKey($requestId)] = $abort;
166: $held = $abort;
167:
168: $cancellation = new CompositeCancellation($cancellation, $abort->getCancellation());
169: }
170:
171: $this->exchanges->track(async(function () use ($message, $headers, $cancellation, $requestId, $held): void {
172: $fiber = \Fiber::getCurrent();
173: \assert($fiber instanceof \Fiber);
174: $this->exchangeFibers[spl_object_id($fiber)] = $fiber;
175:
176: try {
177: $this->exchange($message, $headers, $cancellation);
178: } catch (CancelledException) {
179: // A close or an abort is not a fault, and the protocol layer signalled both.
180: } catch (\Throwable $e) {
181: try {
182: $this->events->emitError(null === $requestId ? $e : new OutboundRequestFailedException($requestId, $e));
183: } catch (\Throwable) {
184: // A listener that throws must not cost this exchange its release, and the error channel just failed anyway.
185: }
186: }
187:
188: if (null !== $requestId) {
189: $key = $this->buildKey($requestId);
190:
191: if (($this->inFlight[$key] ?? null) === $held) {
192: unset($this->inFlight[$key]);
193: }
194: }
195:
196: unset($this->exchangeFibers[spl_object_id($fiber)]);
197: }));
198: }
199:
200: #[\Override]
201: public function close(): void
202: {
203: if ($this->closing) {
204: if ($this->participatesInClose()) {
205: return;
206: }
207:
208: $this->closeCompletion?->getFuture()->await();
209:
210: return;
211: }
212:
213: $this->closing = true;
214: $this->closingFiber = \Fiber::getCurrent();
215:
216: /** @var DeferredFuture<null> $completion */
217: $completion = new DeferredFuture();
218: $this->closeCompletion = $completion;
219:
220: try {
221: $this->events->emitDrain();
222: } finally {
223: $this->state = TransportState::Closed;
224:
225: try {
226: $this->lifetime?->cancel();
227: $this->exchanges->flushPending();
228: $this->events->emitClose();
229: $this->logger->info('Streamable HTTP client transport closed.');
230: } finally {
231: $completion->complete();
232: }
233: }
234: }
235:
236: #[\Override]
237: public function onMessage(\Closure $listener): ListenerHandleInterface
238: {
239: return $this->events->onMessage($listener);
240: }
241:
242: #[\Override]
243: public function onError(\Closure $listener): ListenerHandleInterface
244: {
245: return $this->events->onError($listener);
246: }
247:
248: #[\Override]
249: public function onDrain(\Closure $listener): ListenerHandleInterface
250: {
251: return $this->events->onDrain($listener);
252: }
253:
254: #[\Override]
255: public function onClose(\Closure $listener): ListenerHandleInterface
256: {
257: return $this->events->onClose($listener);
258: }
259:
260: #[\Override]
261: public function abort(RequestId $id): void
262: {
263: ($this->inFlight[$this->buildKey($id)] ?? null)?->cancel();
264: }
265:
266: /**
267: * Whether the current fiber is one the in-progress close is itself running or draining.
268: */
269: private function participatesInClose(): bool
270: {
271: $current = \Fiber::getCurrent();
272:
273: return $current === $this->closingFiber || \in_array($current, $this->exchangeFibers, true);
274: }
275:
276: /**
277: * @return non-empty-string
278: */
279: private function buildKey(RequestId $id): string
280: {
281: return \sprintf('"id":%s', var_export($id->id, true));
282: }
283:
284: /**
285: * @param array<non-empty-string, string> $headers Mirrored parameter headers the protocol layer computed
286: */
287: private function exchange(JsonRpcMessage $message, array $headers, Cancellation $cancellation): void
288: {
289: $response = $this->client->request($this->buildRequest($message, $headers), $cancellation);
290: $status = $response->getStatus();
291:
292: if (HttpStatus::Accepted->value === $status) {
293: if ($message instanceof JsonRpcRequest) {
294: throw new UnexpectedHttpStatusException($status);
295: }
296:
297: return;
298: }
299:
300: if (HttpStatus::Ok->value !== $status) {
301: if (! $message instanceof JsonRpcRequest || $this->isEventStream($response)) {
302: throw new UnexpectedHttpStatusException($status);
303: }
304:
305: try {
306: $payload = $this->buffer($response, $cancellation);
307: } catch (ResponseTooLargeException) {
308: throw new UnexpectedHttpStatusException($status);
309: }
310:
311: $decoded = json_decode($payload, true);
312:
313: if (\is_array($decoded)
314: && JsonRpcMessage::JSONRPC_VERSION === ($decoded['jsonrpc'] ?? null)
315: && $message->id->id === ($decoded['id'] ?? null)
316: && ! \array_key_exists('method', $decoded)
317: && (\array_key_exists('result', $decoded) || \array_key_exists('error', $decoded))
318: ) {
319: Assert::that($decoded)->isMap('Streamable HTTP client received a response envelope that is not a string-keyed object.');
320: $this->events->emitMessage($decoded, new ReceiveContext());
321:
322: return;
323: }
324:
325: throw new UnexpectedHttpStatusException($status, $payload);
326: }
327:
328: if ($this->isEventStream($response)) {
329: $this->readStream($response, $cancellation);
330:
331: return;
332: }
333:
334: $this->events->emitMessage($this->decode($this->buffer($response, $cancellation)), new ReceiveContext());
335: }
336:
337: /**
338: * @throws ResponseTooLargeException
339: */
340: private function buffer(Response $response, Cancellation $cancellation): string
341: {
342: try {
343: return $response->getBody()->buffer($cancellation, $this->maxResponseBytes);
344: } catch (BufferException $e) {
345: throw new ResponseTooLargeException($this->maxResponseBytes, $e);
346: }
347: }
348:
349: /**
350: * @param array<non-empty-string, string> $headers
351: */
352: private function buildRequest(JsonRpcMessage $message, array $headers): Request
353: {
354: $request = new Request($this->endpoint, 'POST', json_encode($message, \JSON_THROW_ON_ERROR | \JSON_UNESCAPED_SLASHES | \JSON_UNESCAPED_UNICODE));
355:
356: $request->setHeaders([
357: 'Content-Type' => 'application/json',
358: 'Accept' => 'application/json, text/event-stream',
359: ...$headers,
360: ...$this->standardHeaders->build($message->toArray()),
361: ]);
362:
363: $request->setTransferTimeout(0.0);
364: $request->setInactivityTimeout($this->readTimeout);
365:
366: return $request;
367: }
368:
369: private function readStream(Response $response, Cancellation $cancellation): void
370: {
371: $parser = new SseFrameParser($this->maxResponseBytes);
372: $body = $response->getBody();
373: $chunk = $body->read($cancellation);
374:
375: while (null !== $chunk) {
376: foreach ($parser->feed($chunk) as $frame) {
377: try {
378: $envelope = $this->decode($frame->data);
379: } catch (\InvalidArgumentException|\JsonException $e) {
380: $this->events->emitError($e);
381:
382: continue;
383: }
384:
385: $this->events->emitMessage($envelope, new ReceiveContext());
386: }
387:
388: $chunk = $body->read($cancellation);
389: }
390: }
391:
392: /**
393: * @return array<string, mixed>
394: *
395: * @throws \InvalidArgumentException
396: * @throws \JsonException
397: */
398: private function decode(string $payload): array
399: {
400: $envelope = json_decode($payload, associative: true, flags: \JSON_THROW_ON_ERROR);
401: Assert::that($envelope)->isMap('Streamable HTTP client received a payload that is not a JSON-RPC envelope.');
402:
403: return $envelope;
404: }
405:
406: private function isEventStream(Response $response): bool
407: {
408: return str_starts_with(strtolower($response->getHeader('Content-Type') ?? ''), 'text/event-stream');
409: }
410: }
411:
Nexus MCP SDK (1.x-dev) API documentation generated by ApiGen dev-master