Coverage for gco_mcp / cli_runner.py: 100.00%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2CLI runner for the GCO MCP server.
4Provides synchronous and cancellation-aware asynchronous wrappers which shell
5out to the ``gco`` CLI with ``--output json`` and return the result. All
6arguments are passed as separate list elements (shell=False) to prevent command
7injection.
8"""
10import asyncio
11import json
12import os
13import shutil
14import subprocess
15import sys
16from contextlib import suppress
17from pathlib import Path
20def _resolve_project_root() -> Path:
21 """Resolve the directory every ``gco`` subprocess runs from.
23 The CLI discovers ``cdk.json`` (and the rest of the checkout) by walking
24 up from its working directory, so this choice decides whether config- and
25 stack-aware tools see the user's project. Resolution order:
27 1. ``GCO_PROJECT_ROOT`` environment variable — explicit override for MCP
28 clients that cannot set a server working directory. Ignored (with a
29 stderr warning) when it does not point at an existing directory.
30 2. The package's parent directory, when it is a checkout (has
31 ``cdk.json``) — the clone / editable-install / dev-container layout,
32 where the historical ``Path(__file__).parent.parent`` was correct.
33 3. The nearest ancestor of the process working directory containing
34 ``cdk.json`` — a ``uvx`` / ``uv tool install`` launched with the MCP
35 client's ``cwd`` pointing at (or inside) a checkout. Previously this
36 layout silently pinned subprocesses to uv's site-packages and the
37 client-provided ``cwd`` never reached the CLI.
38 4. The process working directory itself — matches the CLI's own
39 fallback when no ``cdk.json`` is found (AWS-facing tools need none).
40 """
41 env_root = os.environ.get("GCO_PROJECT_ROOT", "").strip()
42 if env_root:
43 candidate = Path(env_root).expanduser()
44 if candidate.is_dir():
45 return candidate.resolve()
46 print(
47 f"gco-mcp: GCO_PROJECT_ROOT={env_root!r} is not a directory; ignoring it.",
48 file=sys.stderr,
49 )
50 package_parent = Path(__file__).resolve().parent.parent
51 if (package_parent / "cdk.json").exists():
52 return package_parent
53 cwd = Path.cwd()
54 for parent in (cwd, *cwd.parents):
55 if (parent / "cdk.json").exists():
56 return parent
57 return cwd
60PROJECT_ROOT = _resolve_project_root()
63def _gco_executable() -> str:
64 """Resolve the ``gco`` CLI to invoke.
66 Prefer the ``gco`` console script installed next to the current
67 interpreter -- the copy shipped in the SAME environment as this MCP
68 server -- so a ``uv tool install`` / ``uvx`` install is self-contained
69 and version-matched, never picking up an unrelated ``gco`` earlier on
70 PATH. Fall back to a PATH lookup (the dev / pipx layout), then the bare
71 name so the FileNotFoundError handler below can report it.
72 """
73 bindir = Path(sys.executable).parent
74 for name in ("gco", "gco.exe"):
75 candidate = bindir / name
76 if candidate.exists():
77 return str(candidate)
78 return shutil.which("gco") or "gco"
81def _validated_cli_json_output(output: str) -> str:
82 """Return one strict JSON value or a fail-closed MCP error envelope."""
83 if not output:
84 return json.dumps(
85 {
86 "error": "gco CLI returned empty stdout despite --output json",
87 "exit_code": 1,
88 }
89 )
91 def reject_nonstandard_constant(token: str) -> None:
92 raise ValueError(f"non-standard JSON constant: {token}")
94 try:
95 json.loads(output, parse_constant=reject_nonstandard_constant)
96 except json.JSONDecodeError, ValueError:
97 return json.dumps(
98 {
99 "error": "gco CLI returned malformed or multiple JSON documents",
100 "exit_code": 1,
101 }
102 )
103 return output
106def _run_cli(
107 *args: str,
108 timeout_seconds: int = 120,
109 pass_fds: tuple[int, ...] = (),
110) -> str:
111 """Run a gco CLI command and return its output.
113 All args are passed as separate list elements to subprocess (shell=False),
114 so shell metacharacters in user-provided values are treated as literals
115 and cannot cause command injection. Path arguments are validated to prevent
116 traversal outside the project root. ``timeout_seconds`` may be increased by
117 wrappers for intentionally long-running transfers while preserving the
118 two-minute default for normal tools. ``pass_fds`` is reserved for verified
119 descriptor-backed local-data paths and is omitted from ``subprocess.run``
120 when empty for compatibility with platforms that do not support it.
121 """
122 # Validate any path-like arguments to prevent directory traversal.
123 for arg in args:
124 if arg.startswith("-"):
125 continue # flag, not a path
126 if ".." in arg.split("/"):
127 return json.dumps({"error": f"Invalid argument: path traversal not allowed: {arg}"})
129 cmd = [_gco_executable(), "--output", "json", *args]
130 try:
131 if pass_fds:
132 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - shell=False; validated literal argv
133 cmd,
134 capture_output=True,
135 text=True,
136 timeout=timeout_seconds,
137 cwd=str(PROJECT_ROOT),
138 pass_fds=pass_fds,
139 )
140 else:
141 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - shell=False; validated literal argv
142 cmd,
143 capture_output=True,
144 text=True,
145 timeout=timeout_seconds,
146 cwd=str(PROJECT_ROOT),
147 )
148 output = result.stdout.strip()
149 if result.returncode != 0:
150 error = result.stderr.strip() or output
151 return json.dumps({"error": error, "exit_code": result.returncode})
152 return _validated_cli_json_output(output)
153 except subprocess.TimeoutExpired:
154 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
155 except FileNotFoundError:
156 return json.dumps(
157 {
158 "error": "gco CLI not found. Install GCO so the gco console script is on PATH (e.g. uv tool install the GCO git URL, or pip install -e . from a clone)."
159 }
160 )
163async def _stop_cli_process(
164 process: asyncio.subprocess.Process,
165 communication: asyncio.Task[tuple[bytes, bytes]],
166 *,
167 grace_seconds: float,
168) -> None:
169 """Terminate a CLI process and drain output before escalating to a kill."""
170 if process.returncode is None:
171 with suppress(ProcessLookupError):
172 process.terminate()
173 try:
174 await asyncio.wait_for(asyncio.shield(communication), timeout=grace_seconds)
175 except TimeoutError:
176 if process.returncode is None:
177 with suppress(ProcessLookupError):
178 process.kill()
179 await communication
182async def _run_cli_async(
183 *args: str,
184 timeout_seconds: int = 120,
185 terminate_grace_seconds: float = 5,
186) -> str:
187 """Run ``gco`` asynchronously and terminate it on timeout or cancellation."""
188 for arg in args:
189 if arg.startswith("-"):
190 continue
191 if ".." in arg.split("/"):
192 return json.dumps({"error": f"Invalid argument: path traversal not allowed: {arg}"})
194 cmd = [_gco_executable(), "--output", "json", *args]
195 try:
196 process = await asyncio.create_subprocess_exec( # nosemgrep: dangerous-subprocess-use-audit - shell=False; args are validated and passed as literal argv elements
197 *cmd,
198 stdout=asyncio.subprocess.PIPE,
199 stderr=asyncio.subprocess.PIPE,
200 cwd=str(PROJECT_ROOT),
201 )
202 except FileNotFoundError:
203 return json.dumps(
204 {
205 "error": "gco CLI not found. Install GCO so the gco console script is on PATH (e.g. uv tool install the GCO git URL, or pip install -e . from a clone)."
206 }
207 )
209 communication = asyncio.create_task(process.communicate())
210 try:
211 stdout_bytes, stderr_bytes = await asyncio.wait_for(
212 asyncio.shield(communication), timeout=timeout_seconds
213 )
214 except TimeoutError:
215 await _stop_cli_process(
216 process,
217 communication,
218 grace_seconds=terminate_grace_seconds,
219 )
220 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
221 except asyncio.CancelledError:
222 await _stop_cli_process(
223 process,
224 communication,
225 grace_seconds=terminate_grace_seconds,
226 )
227 raise
229 output = stdout_bytes.decode(errors="replace").strip()
230 if process.returncode != 0:
231 error = stderr_bytes.decode(errors="replace").strip() or output
232 return json.dumps({"error": error, "exit_code": process.returncode})
233 return _validated_cli_json_output(output)