Add list_images and delete_image tools (Gruppe 1)

- list_images: lists all local Docker images sorted by size desc, shows size (human-readable), creation date, in-use marker, and update-available marker; gracefully handles container list failure - delete_image: accepts name:tag or image hash, blocks deletion when image is in use by a container, requires confirmed=True to execute; default shows a dry-run preview - 16 unit tests covering all paths (mock DSM client) - ruff format + check clean Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-13 17:44:57 +02:00
parent b921e3a649
commit 06735bb447
4 changed files with 1416 additions and 73 deletions
@@ -1,21 +1,257 @@
-"""MCP tools for SYNO.Docker.Image: list and check for updates."""
+"""MCP tools for SYNO.Docker.Image: list, check updates, delete."""

 from __future__ import annotations

 import logging
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any

 if TYPE_CHECKING:
    from mcp.server.fastmcp import FastMCP
+
    from mcp_synology_container.config import AppConfig
    from mcp_synology_container.dsm_client import DsmClient

 logger = logging.getLogger(__name__)


+def _human_size(size_bytes: int) -> str:
+    """Convert byte count to human-readable string (GiB / MiB / KiB)."""
+    if size_bytes >= 1024**3:
+        return f"{size_bytes / 1024**3:.1f} GiB"
+    if size_bytes >= 1024**2:
+        return f"{size_bytes / 1024**2:.0f} MiB"
+    if size_bytes >= 1024:
+        return f"{size_bytes / 1024:.0f} KiB"
+    return f"{size_bytes} B"
+
+
+def _format_created(ts: int) -> str:
+    """Format a Unix timestamp as a UTC date string."""
+    if not ts:
+        return "unknown"
+    try:
+        return datetime.fromtimestamp(ts, tz=UTC).strftime("%Y-%m-%d")
+    except (OSError, OverflowError, ValueError):
+        return "unknown"
+
+
+async def _filter_images_for_project(
+    client: DsmClient,
+    project_name: str,
+    all_images: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Filter images to those used by a specific project.
+
+    Fetches project details and cross-references container image IDs.
+    Falls back to returning all images if project details are unavailable.
+
+    Args:
+        client: DsmClient instance.
+        project_name: Project name to filter for.
+        all_images: Full list of images from SYNO.Docker.Image.
+
+    Returns:
+        Subset of images used by the project.
+    """
+    try:
+        list_data = await client.request("SYNO.Docker.Project", "list")
+        projects: dict[str, Any] = list_data if isinstance(list_data, dict) else {}
+        project_entry = next((p for p in projects.values() if p.get("name") == project_name), None)
+
+        if not project_entry:
+            return []
+
+        project_id = project_entry.get("id", "")
+        detail_data = await client.request(
+            "SYNO.Docker.Project",
+            "get",
+            params={"id": project_id},
+        )
+
+        containers = detail_data.get("containers", []) or []
+        image_ids: set[str] = set()
+        image_names: set[str] = set()
+
+        for container in containers:
+            img_id = container.get("Image", "")
+            if img_id:
+                image_ids.add(img_id)
+            cfg_image = container.get("Config", {}).get("Image", "")
+            if cfg_image:
+                name = cfg_image.split(":")[0] if ":" in cfg_image else cfg_image
+                image_names.add(name)
+
+        result = []
+        for img in all_images:
+            img_id = img.get("id", "")
+            repo = img.get("repository", "")
+            if img_id in image_ids or repo in image_names:
+                result.append(img)
+
+        return result
+
+    except Exception as e:
+        logger.debug("Could not filter images for project '%s': %s", project_name, e)
+        return all_images
+
+
 def register_images(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
    """Register all image management tools with the MCP server."""

+    @mcp.tool()
+    async def list_images() -> str:
+        """List all local Docker images sorted by size (largest first).
+
+        Shows name:tag, size, creation date, and whether the image is
+        currently used by at least one container.
+        """
+        try:
+            img_data = await client.request(
+                "SYNO.Docker.Image",
+                "list",
+                params={"limit": "-1", "offset": "0", "show_dsm": "false"},
+            )
+        except Exception as e:
+            return f"Error listing images: {e}"
+
+        images: list[dict[str, Any]] = img_data.get("images", [])
+        if not images:
+            return "No local images found."
+
+        # Collect image IDs in use by containers
+        in_use_ids: set[str] = set()
+        try:
+            ctr_data = await client.request(
+                "SYNO.Docker.Container",
+                "list",
+                params={"limit": "-1", "offset": "0", "type": "all"},
+            )
+            for ctr in ctr_data.get("containers", []):
+                img_id = ctr.get("image_id") or ctr.get("ImageID") or ctr.get("Image", "")
+                if img_id:
+                    in_use_ids.add(img_id)
+        except Exception as e:
+            logger.debug("Could not fetch containers for in-use check: %s", e)
+
+        # Sort by size descending
+        images_sorted = sorted(images, key=lambda x: x.get("size", 0), reverse=True)
+
+        total_size = sum(img.get("size", 0) for img in images)
+        lines = [f"Local images ({len(images)} total, {_human_size(total_size)}):", ""]
+
+        for img in images_sorted:
+            repo = img.get("repository", "<none>")
+            tags = img.get("tags") or ["<none>"]
+            tag_str = ", ".join(tags)
+            size = _human_size(img.get("size", 0))
+            created = _format_created(img.get("created", 0))
+            img_id = img.get("id", "")
+            used_marker = "  [in use]" if img_id in in_use_ids else ""
+            upgradable = "  [update available]" if img.get("upgradable") else ""
+            lines.append(f"  {repo}:{tag_str}  {size}  {created}{used_marker}{upgradable}")
+
+        return "\n".join(lines)
+
+    @mcp.tool()
+    async def delete_image(image_id: str, confirmed: bool = False) -> str:
+        """Delete a local Docker image by name:tag or image ID hash.
+
+        Without confirmed=True, returns a preview of what would be deleted.
+        Refuses deletion if the image is used by any container.
+
+        Args:
+            image_id: Image reference as "name:tag" (e.g. "nginx:1.24")
+                      or a full/short image hash (e.g. "sha256:14300de7…").
+            confirmed: Must be True to actually delete. Default False shows
+                       a preview only.
+        """
+        # Parse name:tag
+        name, _, tag = image_id.partition(":")
+        if not tag:
+            tag = "latest"
+
+        # Fetch the local image list for size reporting and in-use detection
+        try:
+            img_data = await client.request(
+                "SYNO.Docker.Image",
+                "list",
+                params={"limit": "-1", "offset": "0", "show_dsm": "false"},
+            )
+        except Exception as e:
+            return f"Error fetching image list: {e}"
+
+        images: list[dict[str, Any]] = img_data.get("images", [])
+
+        # Locate the target image by name+tag or hash prefix
+        is_hash = image_id.startswith("sha256:") or (len(image_id) >= 12 and ":" not in image_id)
+        target: dict[str, Any] | None = None
+
+        for img in images:
+            if is_hash:
+                img_hash = img.get("id", "")
+                if img_hash == image_id or img_hash.startswith(image_id):
+                    target = img
+                    break
+            else:
+                repo = img.get("repository", "")
+                img_tags = img.get("tags") or []
+                if repo == name and tag in img_tags:
+                    target = img
+                    break
+
+        if target is None:
+            return f"Image '{image_id}' not found locally."
+
+        # Resolve display info from the found image
+        repo = target.get("repository", name)
+        img_tags = target.get("tags") or [tag]
+        display_name = f"{repo}:{img_tags[0]}"
+        size_str = _human_size(target.get("size", 0))
+        img_hash = target.get("id", "")
+
+        # Check if image is in use by any container
+        in_use_by: list[str] = []
+        try:
+            ctr_data = await client.request(
+                "SYNO.Docker.Container",
+                "list",
+                params={"limit": "-1", "offset": "0", "type": "all"},
+            )
+            for ctr in ctr_data.get("containers", []):
+                ctr_img_id = ctr.get("image_id") or ctr.get("ImageID") or ctr.get("Image", "")
+                hash_prefix = img_hash[:12] if img_hash else ""
+                if img_hash and (
+                    ctr_img_id == img_hash or (hash_prefix and ctr_img_id.startswith(hash_prefix))
+                ):
+                    ctr_name = ctr.get("name") or ctr.get("Names", ["?"])[0]
+                    in_use_by.append(ctr_name)
+        except Exception as e:
+            logger.debug("Could not fetch containers for in-use check: %s", e)
+
+        if in_use_by:
+            return f"Cannot delete '{display_name}': image is used by " + ", ".join(in_use_by)
+
+        if not confirmed:
+            return (
+                f"Preview: would delete {display_name} ({size_str}).\n"
+                f"Call delete_image(image_id={image_id!r}, confirmed=True) to confirm."
+            )
+
+        # Perform deletion using the resolved name+tag
+        delete_name = repo
+        delete_tag = img_tags[0] if img_tags else tag
+        try:
+            await client.request(
+                "SYNO.Docker.Image",
+                "delete",
+                params={"name": delete_name, "tag": delete_tag},
+            )
+        except Exception as e:
+            return f"Error deleting image '{display_name}': {e}"
+
+        return f"Deleted {display_name} — {size_str} freed."
+
    @mcp.tool()
    async def check_image_updates(project_name: str | None = None) -> str:
        """Check for available image updates for a project or all images.
@@ -40,7 +276,6 @@ def register_images(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
        if not images:
            return "No images found."

-        # If project_name given, cross-reference with project containers
        if project_name:
            images = await _filter_images_for_project(client, project_name, images)
            if not images:
@@ -75,71 +310,3 @@ def register_images(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
            lines.append("All images are up to date.")

        return "\n".join(lines)
-
-
-async def _filter_images_for_project(
-    client: DsmClient,
-    project_name: str,
-    all_images: list[dict[str, Any]],
-) -> list[dict[str, Any]]:
-    """Filter images to those used by a specific project.
-
-    Fetches project details and cross-references container image IDs.
-    Falls back to name-based matching if project details unavailable.
-
-    Args:
-        client: DsmClient instance.
-        project_name: Project name to filter for.
-        all_images: Full list of images from SYNO.Docker.Image.
-
-    Returns:
-        Subset of images used by the project.
-    """
-    # Get project details to find used images
-    try:
-        # Find project by name
-        list_data = await client.request("SYNO.Docker.Project", "list")
-        projects: dict[str, Any] = list_data if isinstance(list_data, dict) else {}
-        project_entry = next(
-            (p for p in projects.values() if p.get("name") == project_name), None
-        )
-
-        if not project_entry:
-            return []
-
-        project_id = project_entry.get("id", "")
-        # Get project detail which includes container image info
-        detail_data = await client.request(
-            "SYNO.Docker.Project",
-            "get",
-            params={"id": project_id},
-        )
-
-        containers = detail_data.get("containers", []) or []
-        image_ids: set[str] = set()
-        image_names: set[str] = set()
-
-        for container in containers:
-            img_id = container.get("Image", "")
-            if img_id:
-                image_ids.add(img_id)
-            cfg_image = container.get("Config", {}).get("Image", "")
-            if cfg_image:
-                # Strip tag for name matching
-                name = cfg_image.split(":")[0] if ":" in cfg_image else cfg_image
-                image_names.add(name)
-
-        # Match images
-        result = []
-        for img in all_images:
-            img_id = img.get("id", "")
-            repo = img.get("repository", "")
-            if img_id in image_ids or repo in image_names:
-                result.append(img)
-
-        return result
-
-    except Exception as e:
-        logger.debug("Could not filter images for project '%s': %s", project_name, e)
-        # Fallback: return all images
-        return all_images