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