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