Coverage for gco_mcp/cli_runner.py: 83.74%
87 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
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(): 51 ↛ 53line 51 didn't jump to line 53 because the condition on line 51 was always true
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 _run_cli(
82 *args: str,
83 timeout_seconds: int = 120,
84 pass_fds: tuple[int, ...] = (),
85) -> str:
86 """Run a gco CLI command and return its output.
88 All args are passed as separate list elements to subprocess (shell=False),
89 so shell metacharacters in user-provided values are treated as literals
90 and cannot cause command injection. Path arguments are validated to prevent
91 traversal outside the project root. ``timeout_seconds`` may be increased by
92 wrappers for intentionally long-running transfers while preserving the
93 two-minute default for normal tools. ``pass_fds`` is reserved for verified
94 descriptor-backed local-data paths and is omitted from ``subprocess.run``
95 when empty for compatibility with platforms that do not support it.
96 """
97 # Validate any path-like arguments to prevent directory traversal.
98 for arg in args:
99 if arg.startswith("-"):
100 continue # flag, not a path
101 if ".." in arg.split("/"):
102 return json.dumps({"error": f"Invalid argument: path traversal not allowed: {arg}"})
104 cmd = [_gco_executable(), "--output", "json", *args]
105 try:
106 if pass_fds:
107 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - shell=False; validated literal argv
108 cmd,
109 capture_output=True,
110 text=True,
111 timeout=timeout_seconds,
112 cwd=str(PROJECT_ROOT),
113 pass_fds=pass_fds,
114 )
115 else:
116 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - shell=False; validated literal argv
117 cmd,
118 capture_output=True,
119 text=True,
120 timeout=timeout_seconds,
121 cwd=str(PROJECT_ROOT),
122 )
123 output = result.stdout.strip()
124 if result.returncode != 0:
125 error = result.stderr.strip() or output
126 return json.dumps({"error": error, "exit_code": result.returncode})
127 return output if output else json.dumps({"status": "ok"})
128 except subprocess.TimeoutExpired:
129 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
130 except FileNotFoundError:
131 return json.dumps(
132 {
133 "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)."
134 }
135 )
138async def _stop_cli_process(
139 process: asyncio.subprocess.Process,
140 communication: asyncio.Task[tuple[bytes, bytes]],
141 *,
142 grace_seconds: float,
143) -> None:
144 """Terminate a CLI process and drain output before escalating to a kill."""
145 if process.returncode is None:
146 with suppress(ProcessLookupError):
147 process.terminate()
148 try:
149 await asyncio.wait_for(asyncio.shield(communication), timeout=grace_seconds)
150 except TimeoutError:
151 if process.returncode is None: 151 ↛ 154line 151 didn't jump to line 154 because the condition on line 151 was always true
152 with suppress(ProcessLookupError):
153 process.kill()
154 await communication
157async def _run_cli_async(
158 *args: str,
159 timeout_seconds: int = 120,
160 terminate_grace_seconds: float = 5,
161) -> str:
162 """Run ``gco`` asynchronously and terminate it on timeout or cancellation."""
163 for arg in args:
164 if arg.startswith("-"): 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 continue
166 if ".." in arg.split("/"):
167 return json.dumps({"error": f"Invalid argument: path traversal not allowed: {arg}"})
169 cmd = [_gco_executable(), "--output", "json", *args]
170 try:
171 process = await asyncio.create_subprocess_exec( # nosemgrep: dangerous-subprocess-use-audit - shell=False; args are validated and passed as literal argv elements
172 *cmd,
173 stdout=asyncio.subprocess.PIPE,
174 stderr=asyncio.subprocess.PIPE,
175 cwd=str(PROJECT_ROOT),
176 )
177 except FileNotFoundError:
178 return json.dumps(
179 {
180 "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)."
181 }
182 )
184 communication = asyncio.create_task(process.communicate())
185 try:
186 stdout_bytes, stderr_bytes = await asyncio.wait_for(
187 asyncio.shield(communication), timeout=timeout_seconds
188 )
189 except TimeoutError:
190 await _stop_cli_process(
191 process,
192 communication,
193 grace_seconds=terminate_grace_seconds,
194 )
195 return json.dumps({"error": f"Command timed out after {timeout_seconds} seconds"})
196 except asyncio.CancelledError:
197 await _stop_cli_process(
198 process,
199 communication,
200 grace_seconds=terminate_grace_seconds,
201 )
202 raise
204 output = stdout_bytes.decode(errors="replace").strip()
205 if process.returncode != 0:
206 error = stderr_bytes.decode(errors="replace").strip() or output
207 return json.dumps({"error": error, "exit_code": process.returncode})
208 return output if output else json.dumps({"status": "ok"})