Coverage for gco_mcp/tools/images.py: 97.44%
140 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Container image registry MCP tools.
3All tools wrap ``cli/images.py::ImageManager`` so the MCP layer never
4re-implements the underlying ECR/runtime logic. Read-only and
5administrative tools are unconditional. Build/push tools register only
6when ``GCO_ENABLE_IMAGE_PUBLISH`` is set; destructive tools register
7only when ``GCO_ENABLE_DESTRUCTIVE_OPERATIONS`` is set.
8"""
10from __future__ import annotations
12import asyncio
13import json
14from typing import Any
16from audit import audit_logged
17from feature_flags import (
18 FLAG_DESTRUCTIVE_OPERATIONS,
19 FLAG_IMAGE_PUBLISH,
20 is_enabled,
21)
22from server import mcp
24from tools._long_task import _run_long_task
26# FastMCP's Progress / Context dependencies are optional from this
27# module's perspective — when ``fastmcp[tasks]`` is reachable they
28# inject real instances per call; otherwise the gated build/push tools
29# still register but rely on caller-provided fakes (the test path).
30try:
31 from fastmcp.server.dependencies import CurrentContext, Progress
32except ImportError: # pragma: no cover - degraded fastmcp install
33 CurrentContext = None # type: ignore[assignment]
34 Progress = None # type: ignore[misc,assignment]
36# TaskConfig is best-effort wired so MCP clients that opt into the task
37# protocol can run build/push as background tasks. If the import path
38# moves between fastmcp versions, the tools register without it.
39try:
40 from fastmcp.server.tasks.config import TaskConfig
42 _TASK_CONFIG_OPTIONAL: Any = TaskConfig(mode="optional")
43except ImportError: # pragma: no cover - degraded fastmcp install
44 _TASK_CONFIG_OPTIONAL = None
47def _get_manager() -> Any:
48 """Lazy-import ``cli.images.get_image_manager`` so MCP server
49 import doesn't pull boto3 prematurely.
50 """
51 from cli.images import get_image_manager
53 return get_image_manager()
56def _get_image_mirror() -> Any:
57 """Lazy-import the ``cli._image_mirror`` core so MCP server import stays light.
59 The mirror tools wrap this general "copy third-party images into gco/* ECR"
60 module (the same core the deploy auto-mirror and ``gco images mirror``
61 use) rather than ``ImageManager`` — see ``docs/IMAGE_MIRROR.md``.
62 """
63 from cli import _image_mirror
65 return _image_mirror
68# =============================================================================
69# Read-only tools — Risk_Tier "safe"
70# =============================================================================
73@mcp.tool(tags={"safe", "images"})
74@audit_logged
75async def images_list() -> str:
76 """`gco images list` — list every gco/* repository in ECR."""
77 return await asyncio.to_thread(lambda: json.dumps(_get_manager().list_repos()))
80@mcp.tool(tags={"safe", "images"})
81@audit_logged
82async def images_tags(name: str) -> str:
83 """`gco images tags` — list tags within a repository.
85 Args:
86 name: Repository name (without the ``gco/`` prefix).
87 """
88 return await asyncio.to_thread(lambda: json.dumps(_get_manager().list_tags(name)))
91@mcp.tool(tags={"safe", "images"})
92@audit_logged
93async def images_describe(name: str, tag: str) -> str:
94 """`gco images describe` — full ECR details for a single image tag.
96 Args:
97 name: Repository name (without the ``gco/`` prefix).
98 tag: Image tag.
99 """
100 return await asyncio.to_thread(lambda: json.dumps(_get_manager().describe(name, tag)))
103@mcp.tool(tags={"safe", "images"})
104@audit_logged
105async def images_uri(name: str, tag: str = "latest") -> str:
106 """`gco images uri` — return the registry URI for an image. No AWS calls.
108 Args:
109 name: Repository name (without the ``gco/`` prefix).
110 tag: Image tag. Defaults to ``latest``.
111 """
112 return await asyncio.to_thread(
113 lambda: json.dumps({"uri": _get_manager().get_uri(name, tag=tag)})
114 )
117@mcp.tool(tags={"safe", "images"})
118@audit_logged
119async def images_replication_get() -> str:
120 """`gco images replication get` — current ECR replication configuration."""
121 return await asyncio.to_thread(lambda: json.dumps(_get_manager().replication_get()))
124@mcp.tool(tags={"safe", "images"})
125@audit_logged
126async def images_replication_status() -> str:
127 """`gco images replication status` — per-image replication status across project repos."""
128 return await asyncio.to_thread(lambda: json.dumps(_get_manager().replication_status()))
131@mcp.tool(tags={"safe", "images"})
132@audit_logged
133async def images_orphans(threshold_days: int = 30) -> str:
134 """`gco images orphans` — list gco/* tags older than ``threshold_days`` with no references.
136 Args:
137 threshold_days: Age threshold in days. Defaults to 30.
138 """
139 return await asyncio.to_thread(
140 lambda: json.dumps(_get_manager().orphans(threshold_days=threshold_days))
141 )
144@mcp.tool(tags={"safe", "images"})
145@audit_logged
146async def images_mirror_plan(region: str, ecr_namespace: str | None = None) -> str:
147 """Image mirror — show which third-party images would be mirrored into ECR.
149 Read-only planning view: resolves the destination registry and repository
150 for every image the mirror manages (Volcano's ``docker.io/volcanosh/vc-*``
151 images today, derived from ``charts.yaml``) without creating any repository
152 or copying anything. Use it to preview the mirror before a deploy, or to see
153 where a consumer's ``image_registry`` override should point. The ``enabled``
154 field reflects the cdk.json ``volcano_image_mirror`` toggle that drives the
155 auto-mirror on ``gco stacks deploy``. See ``docs/IMAGE_MIRROR.md``.
157 Args:
158 region: AWS region whose ECR registry is the mirror destination.
159 ecr_namespace: Destination namespace under the registry. Defaults to the
160 cdk.json ``volcano_image_mirror.ecr_namespace`` (e.g. ``gco/dockerhub``).
161 """
163 def _plan() -> str:
164 mirror = _get_image_mirror()
165 plan = mirror.plan_mirror(region, ecr_namespace)
166 plan["enabled"] = mirror.read_mirror_config()["enabled"]
167 return json.dumps(plan)
169 return await asyncio.to_thread(_plan)
172@mcp.tool(tags={"safe", "images"})
173@audit_logged
174async def images_mirror_status(region: str, ecr_namespace: str | None = None) -> str:
175 """Image mirror — report which managed images are already present in ECR.
177 Read-only: for every image the mirror manages, checks whether its tag
178 already exists in the ``gco/*`` ECR namespace (ECR ``DescribeImages``; no
179 writes). The result carries a ``mirrored`` flag per image plus top-level
180 ``all_mirrored`` / ``missing``, so you can confirm a deploy's auto-mirror
181 has populated everything the consuming Helm install (Volcano) needs before
182 it runs, or diagnose a stuck install. See ``docs/IMAGE_MIRROR.md``.
184 Args:
185 region: AWS region whose ECR registry is the mirror destination.
186 ecr_namespace: Destination namespace under the registry. Defaults to the
187 cdk.json ``volcano_image_mirror.ecr_namespace`` (e.g. ``gco/dockerhub``).
188 """
189 return await asyncio.to_thread(
190 lambda: json.dumps(_get_image_mirror().mirror_status(region, ecr_namespace))
191 )
194# =============================================================================
195# Administrative tools — Risk_Tier "low-risk"
196# =============================================================================
199@mcp.tool(tags={"low-risk", "images"})
200@audit_logged
201async def images_init(name: str, retain: bool = False) -> str:
202 """`gco images init` — create the project ECR repo idempotently with default lifecycle.
204 Args:
205 name: Repository name (without the ``gco/`` prefix).
206 retain: When True, mark the repository with ``gco:retain=true`` so it
207 survives stack destroys.
208 """
209 return await asyncio.to_thread(lambda: json.dumps(_get_manager().init(name, retain=retain)))
212@mcp.tool(tags={"low-risk", "images"})
213@audit_logged
214async def images_lifecycle_get(name: str) -> str:
215 """`gco images lifecycle get` — print the lifecycle policy on a repository.
217 Args:
218 name: Repository name (without the ``gco/`` prefix).
219 """
220 return await asyncio.to_thread(lambda: json.dumps(_get_manager().lifecycle_get(name)))
223@mcp.tool(tags={"low-risk", "images"})
224@audit_logged
225async def images_lifecycle_set(name: str, policy: dict[str, Any]) -> str:
226 """`gco images lifecycle set` — replace the lifecycle policy on a repository.
228 Args:
229 name: Repository name (without the ``gco/`` prefix).
230 policy: ECR lifecycle policy document as a dict.
231 """
232 return await asyncio.to_thread(lambda: json.dumps(_get_manager().lifecycle_set(name, policy)))
235@mcp.tool(tags={"low-risk", "images"})
236@audit_logged
237async def images_replication_sync() -> str:
238 """`gco images replication sync` — apply the standard gco/* replication rule."""
239 return await asyncio.to_thread(lambda: json.dumps(_get_manager().replication_sync()))
242# =============================================================================
243# Image publish — gated by GCO_ENABLE_IMAGE_PUBLISH
244# =============================================================================
245#
246# build/push are long-running data-upload operations. They run via
247# ``_run_long_task`` so progress messages stream back through the
248# FastMCP Progress dependency.
250if is_enabled(FLAG_IMAGE_PUBLISH):
251 # images_mirror copies third-party images (Volcano's docker.io images) into
252 # the project's gco/* ECR. Like build/push it uploads image data, so it
253 # shares the GCO_ENABLE_IMAGE_PUBLISH gate. Unlike build/push it wraps the
254 # mirror core directly via asyncio.to_thread (no Progress/Context injection
255 # and no _run_long_task), so it registers here at the top of the flag block
256 # rather than inside the Progress-dependent build/push block below.
257 @mcp.tool(tags={"image", "images"})
258 @audit_logged
259 async def images_mirror(
260 region: str,
261 ecr_namespace: str | None = None,
262 skip_existing: bool = True,
263 ) -> str:
264 """[gated by GCO_ENABLE_IMAGE_PUBLISH] image-upload.
266 Mirror third-party images into the project's ``gco/*`` ECR so the
267 cluster pulls them from same-account ECR instead of a rate-limited
268 upstream (chiefly ``docker.io``, which has no credential-free ECR
269 pull-through cache). Today that's Volcano's ``volcanosh/vc-*`` images,
270 derived from ``charts.yaml``; the set is general — see
271 ``docs/IMAGE_MIRROR.md`` for how to add another image.
273 Creates each destination repository if needed and copies the image
274 preserving its full multi-arch manifest list (so both amd64 and arm64
275 nodes find a match). Idempotent — already-mirrored tags are skipped when
276 ``skip_existing`` is True, so repeat runs are a fast no-op. This is the
277 same operation ``gco stacks deploy`` runs automatically when the cdk.json
278 ``volcano_image_mirror`` toggle is enabled; invoke it directly to
279 pre-seed or repair the mirror out of band.
281 Args:
282 region: AWS region whose ECR registry is the mirror destination.
283 ecr_namespace: Destination namespace under the registry. Defaults to
284 the cdk.json ``volcano_image_mirror.ecr_namespace`` (e.g.
285 ``gco/dockerhub``).
286 skip_existing: When True (default), skip any image whose tag already
287 exists in ECR so repeat runs don't re-copy.
288 """
290 def _run() -> str:
291 mirror = _get_image_mirror()
292 lines: list[str] = []
293 result = mirror.mirror_images(
294 region,
295 ecr_namespace=ecr_namespace,
296 skip_existing=skip_existing,
297 log=lines.append,
298 )
299 result["log"] = lines
300 return json.dumps(result)
302 return await asyncio.to_thread(_run)
304 # Build the decorator kwargs dict so we only pass ``task=...`` when
305 # TaskConfig was importable on this fastmcp version.
306 _publish_decorator_kwargs: dict[str, Any] = {"tags": {"image", "images"}}
307 if _TASK_CONFIG_OPTIONAL is not None: 307 ↛ 310line 307 didn't jump to line 310 because the condition on line 307 was always true
308 _publish_decorator_kwargs["task"] = _TASK_CONFIG_OPTIONAL
310 if Progress is not None and CurrentContext is not None: 310 ↛ 388line 310 didn't jump to line 388 because the condition on line 310 was always true
312 @mcp.tool(**_publish_decorator_kwargs) # type: ignore[untyped-decorator]
313 @audit_logged
314 async def images_build(
315 context: str,
316 name: str,
317 tag: str | None = None,
318 dockerfile: str = "Dockerfile",
319 platform: str = "linux/amd64",
320 retain: bool = False,
321 *,
322 ctx: Any = CurrentContext(),
323 progress: Any = Progress(),
324 ) -> str:
325 """[gated by GCO_ENABLE_IMAGE_PUBLISH] long-running, data-upload.
327 `gco images build` — build a container image and push to ECR.
329 Args:
330 context: Build context directory.
331 name: Image name (lowercase letters, digits, dashes; max 63 chars).
332 tag: Image tag (defaults to git short SHA, else ``latest``).
333 dockerfile: Path to the Dockerfile, relative to ``context``.
334 platform: ``--platform`` argument for the build.
335 retain: When True, mark the repository with ``gco:retain=true``
336 so it survives stack destroys.
337 """
338 argv = ["gco", "images", "build", context, "--name", name]
339 if tag:
340 argv += ["--tag", tag]
341 argv += ["--dockerfile", dockerfile, "--platform", platform]
342 if retain:
343 argv.append("--retain")
344 return await _run_long_task(argv, ctx=ctx, progress=progress, is_stack_op=False)
346 @mcp.tool(**_publish_decorator_kwargs) # type: ignore[untyped-decorator]
347 @audit_logged
348 async def images_push(
349 name: str,
350 tag: str,
351 local_image: str,
352 retain: bool = False,
353 *,
354 ctx: Any = CurrentContext(),
355 progress: Any = Progress(),
356 ) -> str:
357 """[gated by GCO_ENABLE_IMAGE_PUBLISH] long-running, data-upload.
359 `gco images push` — push an already-built local image to the project ECR repo.
361 Args:
362 name: Image name (lowercase letters, digits, dashes; max 63 chars).
363 tag: Image tag.
364 local_image: Source image reference on the local container runtime.
365 retain: When True, mark the repository with ``gco:retain=true``
366 so it survives stack destroys.
367 """
368 argv = [
369 "gco",
370 "images",
371 "push",
372 name,
373 "--tag",
374 tag,
375 "--local-image",
376 local_image,
377 ]
378 if retain:
379 argv.append("--retain")
380 return await _run_long_task(argv, ctx=ctx, progress=progress, is_stack_op=False)
383# =============================================================================
384# Destructive image tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS
385# =============================================================================
388async def _ctx_warning(message: str) -> None:
389 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context.
391 Tools wrapped here are short-lived enough that we don't need the full
392 ``_run_long_task`` stack — we just want operators (and the audit log)
393 to see a warning when destructive work runs.
394 """
395 import contextlib as _contextlib
397 try:
398 from fastmcp.server.dependencies import get_context
400 ctx = get_context()
401 except Exception:
402 return
403 with _contextlib.suppress(Exception):
404 await ctx.warning(message)
407if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS):
409 @mcp.tool(tags={"destructive", "images"})
410 @audit_logged
411 async def images_delete_tag(name: str, tag: str) -> str:
412 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
414 `gco images delete-tag` — delete a single tag from a repository.
415 Cannot be undone — the image manifest is removed from ECR.
417 Args:
418 name: Repository name (without the ``gco/`` prefix).
419 tag: Image tag to delete.
420 """
421 await _ctx_warning(f"Deleting tag {tag!r} from gco/{name} — this cannot be undone.")
422 return await asyncio.to_thread(lambda: json.dumps(_get_manager().delete_tag(name, tag)))
424 @mcp.tool(tags={"destructive", "images"})
425 @audit_logged
426 async def images_delete_repo(name: str, force: bool = False) -> str:
427 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
429 `gco images delete-repo` — delete a whole repository.
430 Cannot be undone — the repo and (when ``force=True``) every image
431 inside it are permanently removed from ECR.
433 Args:
434 name: Repository name (without the ``gco/`` prefix).
435 force: When True, also delete every image inside the repo.
436 """
437 await _ctx_warning(
438 f"Deleting repository gco/{name} (force={force}) — this cannot be undone."
439 )
440 return await asyncio.to_thread(
441 lambda: json.dumps(_get_manager().delete_repo(name, force=force))
442 )
444 @mcp.tool(tags={"destructive", "images"})
445 @audit_logged
446 async def images_cleanup(name: str | None = None, all: bool = False) -> str:
447 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
449 `gco images cleanup` — remove every untagged image across one or all project repos.
450 Cannot be undone — untagged image manifests are permanently deleted.
452 Args:
453 name: Repository name to clean (without the ``gco/`` prefix). Required
454 unless ``all=True``.
455 all: When True, clean every project repository.
456 """
457 scope = "all repos" if all else f"gco/{name}"
458 await _ctx_warning(f"Cleaning untagged images from {scope} — this cannot be undone.")
459 return await asyncio.to_thread(
460 lambda: json.dumps(_get_manager().cleanup(name=name, all=all))
461 )
463 @mcp.tool(tags={"destructive", "images"})
464 @audit_logged
465 async def images_prune(dry_run: bool = True) -> str:
466 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
468 `gco images prune` — remove untagged images older than 30 days.
469 Cannot be undone when ``dry_run=False``; the matching image manifests
470 are permanently deleted.
472 Args:
473 dry_run: When True (default), report what would be deleted without
474 deleting anything.
475 """
476 if not dry_run:
477 await _ctx_warning(
478 "Pruning untagged images older than 30 days — this cannot be undone."
479 )
480 return await asyncio.to_thread(lambda: json.dumps(_get_manager().prune(dry_run=dry_run)))