Coverage for gco_mcp / cli_runner.py: 100.00%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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 check=False,
140 )
141 else:
142 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - shell=False; validated literal argv
143 cmd,
144 capture_output=True,
145 text=True,
146 timeout=timeout_seconds,
147 cwd=str(PROJECT_ROOT),
148 check=False,
149 )
150 output = result.stdout.strip()
151 if result.returncode != 0:
152 error = result.stderr.strip() or output
153 return json.dumps({"error": error, "exit_code": result.returncode})
154 return _validated_cli_json_output(output)
155 except subprocess.TimeoutExpired:
156 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
157 except FileNotFoundError:
158 return json.dumps(
159 {
160 "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)."
161 }
162 )
165async def _stop_cli_process(
166 process: asyncio.subprocess.Process,
167 communication: asyncio.Task[tuple[bytes, bytes]],
168 *,
169 grace_seconds: float,
170) -> None:
171 """Terminate a CLI process and drain output before escalating to a kill."""
172 if process.returncode is None:
173 with suppress(ProcessLookupError):
174 process.terminate()
175 try:
176 await asyncio.wait_for(asyncio.shield(communication), timeout=grace_seconds)
177 except TimeoutError:
178 if process.returncode is None:
179 with suppress(ProcessLookupError):
180 process.kill()
181 await communication
184async def _run_cli_async(
185 *args: str,
186 timeout_seconds: int = 120,
187 terminate_grace_seconds: float = 5,
188) -> str:
189 """Run ``gco`` asynchronously and terminate it on timeout or cancellation."""
190 for arg in args:
191 if arg.startswith("-"):
192 continue
193 if ".." in arg.split("/"):
194 return json.dumps({"error": f"Invalid argument: path traversal not allowed: {arg}"})
196 cmd = [_gco_executable(), "--output", "json", *args]
197 try:
198 process = await asyncio.create_subprocess_exec( # nosemgrep: dangerous-subprocess-use-audit - shell=False; args are validated and passed as literal argv elements
199 *cmd,
200 stdout=asyncio.subprocess.PIPE,
201 stderr=asyncio.subprocess.PIPE,
202 cwd=str(PROJECT_ROOT),
203 )
204 except FileNotFoundError:
205 return json.dumps(
206 {
207 "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)."
208 }
209 )
211 communication = asyncio.create_task(process.communicate())
212 try:
213 stdout_bytes, stderr_bytes = await asyncio.wait_for(
214 asyncio.shield(communication), timeout=timeout_seconds
215 )
216 except TimeoutError:
217 await _stop_cli_process(
218 process,
219 communication,
220 grace_seconds=terminate_grace_seconds,
221 )
222 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
223 except asyncio.CancelledError:
224 await _stop_cli_process(
225 process,
226 communication,
227 grace_seconds=terminate_grace_seconds,
228 )
229 raise
231 output = stdout_bytes.decode(errors="replace").strip()
232 if process.returncode != 0:
233 error = stderr_bytes.decode(errors="replace").strip() or output
234 return json.dumps({"error": error, "exit_code": process.returncode})
235 return _validated_cli_json_output(output)