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\CancelledException;
17: use Amp\DeferredCancellation;
18: use Amp\Process\Process;
19: use Amp\Process\ProcessException;
20: use Nexus\Assert\Assert;
21: use Nexus\Mcp\Core\JsonRpc\SafeDisplay;
22: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
23: use Nexus\Mcp\Core\Transport\LineDuplex;
24: use Nexus\Mcp\Core\Transport\LineReader;
25: use Nexus\Mcp\Core\Transport\SendContext;
26: use Nexus\Mcp\Core\Transport\Subscription;
27: use Nexus\Mcp\Core\Transport\SubscriptionInterface;
28: use Nexus\Mcp\Core\Transport\SupervisableTransportInterface;
29: use Psr\Log\LoggerInterface;
30: use Psr\Log\NullLogger;
31:
32: use function Amp\async;
33:
34: /**
35: * Stdio MCP client transport. Launches an MCP server subprocess and exchanges
36: * line-framed JSON-RPC envelopes over its STDIN/STDOUT.
37: */
38: final class StdioClientTransport implements SupervisableTransportInterface
39: {
40: private const string LABEL = 'Stdio client';
41:
42: /**
43: * Environment variable names safe to inherit by default, across POSIX and Windows hosts.
44: */
45: private const array INHERITED_ENV_NAMES = [
46: 'APPDATA',
47: 'HOME',
48: 'HOMEDRIVE',
49: 'HOMEPATH',
50: 'LOCALAPPDATA',
51: 'LOGNAME',
52: 'PATH',
53: 'PROCESSOR_ARCHITECTURE',
54: 'SHELL',
55: 'SYSTEMDRIVE',
56: 'SYSTEMROOT',
57: 'TEMP',
58: 'TERM',
59: 'USER',
60: 'USERNAME',
61: 'USERPROFILE',
62: ];
63:
64: private readonly LineDuplex $duplex;
65: private readonly LoggerInterface $logger;
66: private ?Process $process = null;
67:
68: /**
69: * Bounds the exit watch. `Process::join()` references the event loop while it awaits, so an
70: * unbounded watch would hold the loop open for the lifetime of the subprocess.
71: */
72: private ?DeferredCancellation $exitWatch = null;
73:
74: /**
75: * @var array<int, \Closure(null|int): void>
76: */
77: private array $exitListeners = [];
78:
79: /**
80: * @param list<string> $command Subprocess argv (no shell interpretation).
81: * @param null|array<string, string> $env Subprocess environment (`null` prunes to a safe allowlist).
82: */
83: public function __construct(
84: private readonly array $command,
85: private readonly ?string $workingDirectory = null,
86: private readonly ?array $env = null,
87: LoggerInterface $logger = new NullLogger(),
88: int $maxLineBytes = LineReader::DEFAULT_MAX_LINE_BYTES,
89: ) {
90: Assert::that($command)->isList(\sprintf('%s command must be a list, {type} given.', self::LABEL));
91: Assert::that(\count($command))->isPositiveInt(\sprintf('%s command must not be empty.', self::LABEL));
92:
93: $this->logger = $logger;
94: $this->duplex = new LineDuplex(
95: hostTransport: self::class,
96: label: self::LABEL,
97: logger: $logger,
98: maxLineBytes: $maxLineBytes,
99: onBeforeClose: function (): void {
100: if (null === $this->process) {
101: return;
102: }
103:
104: $this->process->getStdin()->close();
105: $this->process->kill();
106: },
107: );
108: }
109:
110: /**
111: * Builds the pruned default subprocess environment: the inherited-name allowlist
112: * populated from `$source`, skipping values that look like exported shell functions.
113: *
114: * @internal
115: *
116: * @param null|array<string, string> $source Defaults to the parent process environment.
117: *
118: * @return array<string, string>
119: */
120: public static function buildDefaultEnvironment(?array $source = null): array
121: {
122: $source ??= getenv();
123: $environment = [];
124:
125: foreach (self::INHERITED_ENV_NAMES as $name) {
126: $value = $source[$name] ?? null;
127:
128: if (null === $value) {
129: continue;
130: }
131:
132: if (str_starts_with($value, '()')) {
133: // Skip exported shell-function definitions (Shellshock mitigation).
134: continue;
135: }
136:
137: $environment[$name] = $value;
138: }
139:
140: return $environment;
141: }
142:
143: #[\Override]
144: public function start(): void
145: {
146: $process = Process::start($this->command, $this->workingDirectory, $this->env ?? self::buildDefaultEnvironment());
147:
148: try {
149: $this->duplex->start($process->getStdout(), $process->getStdin());
150: } catch (\Throwable $e) {
151: $process->getStdin()->close();
152: $process->kill();
153:
154: throw $e;
155: }
156:
157: $this->process = $process;
158: $this->logger->info(
159: '{label} transport spawned subprocess. Command: {command} (PID {pid}).',
160: ['label' => self::LABEL, 'command' => implode(' ', $this->command), 'pid' => $process->getPid()],
161: );
162:
163: $this->duplex->forwardLines(
164: $process->getStderr(),
165: function (string $line): void {
166: $this->logger->info('Subprocess stderr: {line}', ['line' => SafeDisplay::sanitise($line)]);
167: },
168: );
169:
170: $this->watchForExit($process);
171: }
172:
173: #[\Override]
174: public function onUnexpectedExit(\Closure $listener): SubscriptionInterface
175: {
176: $id = spl_object_id($listener);
177: $this->exitListeners[$id] = $listener;
178:
179: return new Subscription(function () use ($id): void {
180: unset($this->exitListeners[$id]);
181: });
182: }
183:
184: #[\Override]
185: public function send(JsonRpcMessage $message, ?SendContext $context = null): void
186: {
187: $this->duplex->send($message);
188: }
189:
190: #[\Override]
191: public function close(): void
192: {
193: $this->exitWatch?->cancel();
194: $this->duplex->close();
195: }
196:
197: #[\Override]
198: public function onMessage(\Closure $listener): SubscriptionInterface
199: {
200: return $this->duplex->onMessage($listener);
201: }
202:
203: #[\Override]
204: public function onError(\Closure $listener): SubscriptionInterface
205: {
206: return $this->duplex->onError($listener);
207: }
208:
209: #[\Override]
210: public function onDrain(\Closure $listener): SubscriptionInterface
211: {
212: return $this->duplex->onDrain($listener);
213: }
214:
215: #[\Override]
216: public function onClose(\Closure $listener): SubscriptionInterface
217: {
218: return $this->duplex->onClose($listener);
219: }
220:
221: /**
222: * Reports an exit nobody asked for. `close()` cancels the watch, so a requested shutdown settles
223: * it without reaching the listeners.
224: */
225: private function watchForExit(Process $process): void
226: {
227: $this->exitWatch = new DeferredCancellation();
228: $cancellation = $this->exitWatch->getCancellation();
229:
230: async(function () use ($process, $cancellation): void {
231: try {
232: $exitCode = $process->join($cancellation);
233: } catch (CancelledException|ProcessException $e) {
234: if ($e instanceof CancelledException) {
235: return;
236: }
237:
238: // The wrapper died without reporting a status. No test drives this: POSIX resolves an
239: // empty status pipe to 0, and CI runs no Windows job.
240: $exitCode = null; // @codeCoverageIgnore
241: }
242:
243: $this->logger->warning(
244: '{label} transport subprocess exited unexpectedly (code {exitCode}).',
245: ['label' => self::LABEL, 'exitCode' => $exitCode ?? 'unknown'],
246: );
247:
248: foreach ($this->exitListeners as $listener) {
249: try {
250: $listener($exitCode);
251: } catch (\Throwable $e) {
252: $this->logger->warning(
253: '{label} transport exit listener threw.',
254: ['label' => self::LABEL, 'exception' => $e],
255: );
256: }
257: }
258: })->ignore();
259: }
260: }
261:
Nexus MCP SDK API documentation generated by ApiGen dev-master