Add system_df and system_prune tools (Gruppe 3)

system_df:
  Assembles disk-usage report from SYNO.Docker.Image/list and
  SYNO.Docker.Container/list. Reports image count/size/reclaimable
  (images not referenced by any container), container running/stopped.
  Gracefully degrades when one API is unavailable.

system_prune:
  Without confirmed=True: lists dangling/unused images and stopped
  containers with sizes (dry-run preview).
  With confirmed=True: calls SYNO.Docker.Utils/prune and reports
  reclaimed space from the response (SpaceReclaimed field).

10 unit tests: stats counts, reclaimable detection, preview content,
confirmed execution, missing-response-field graceful handling, API error.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/mcp_synology_container/modules/system.py

@@ -0,0 +1,199 @@
+"""MCP tools for Docker system-level operations: disk usage and prune."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from mcp_synology_container.modules.images import _human_size
+
+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 register_system(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
+    """Register all system-level tools with the MCP server."""
+
+    @mcp.tool()
+    async def system_df() -> str:
+        """Show Docker disk usage: images, containers, and volumes.
+
+        Assembles disk-usage data from the image and container lists.
+        Images that are not referenced by any container are marked as
+        reclaimable.
+        """
+        errors: list[str] = []
+
+        # ── Images ───────────────────────────────────────────────────────────
+        images: list[dict[str, Any]] = []
+        try:
+            img_data = await client.request(
+                "SYNO.Docker.Image",
+                "list",
+                params={"limit": "-1", "offset": "0", "show_dsm": "false"},
+            )
+            images = img_data.get("images", [])
+        except Exception as e:
+            errors.append(f"images: {e}")
+
+        # ── Containers ───────────────────────────────────────────────────────
+        containers: list[dict[str, Any]] = []
+        try:
+            ctr_data = await client.request(
+                "SYNO.Docker.Container",
+                "list",
+                params={"limit": "-1", "offset": "0", "type": "all"},
+            )
+            containers = ctr_data.get("containers", [])
+        except Exception as e:
+            errors.append(f"containers: {e}")
+
+        # ── Compute image stats ───────────────────────────────────────────────
+        # An image is "in use" if any container references its ID
+        in_use_ids: set[str] = set()
+        for ctr in containers:
+            img_id = ctr.get("image_id") or ctr.get("ImageID") or ctr.get("Image", "")
+            if img_id:
+                in_use_ids.add(img_id)
+
+        total_image_size = sum(img.get("size", 0) for img in images)
+        reclaimable_size = sum(
+            img.get("size", 0) for img in images if img.get("id", "") not in in_use_ids
+        )
+        reclaimable_count = sum(1 for img in images if img.get("id", "") not in in_use_ids)
+
+        # ── Compute container stats ───────────────────────────────────────────
+        running = sum(1 for c in containers if c.get("status") in ("running", "up"))
+        stopped = len(containers) - running
+
+        # ── Format output ─────────────────────────────────────────────────────
+        lines = ["Docker Disk Usage", ""]
+
+        # Images table
+        lines.append(f"  Images:     {len(images):>4}  total   {_human_size(total_image_size):>10}")
+        rec_str = _human_size(reclaimable_size)
+        lines.append(f"              {reclaimable_count:>4}  unused  {rec_str:>10}  (reclaimable)")
+
+        # Containers table
+        lines.append("")
+        lines.append(f"  Containers: {len(containers):>4}  total")
+        lines.append(f"              {running:>4}  running")
+        lines.append(f"              {stopped:>4}  stopped  (reclaimable via system_prune)")
+
+        if errors:
+            lines.append("")
+            lines.append("Warnings:")
+            for err in errors:
+                lines.append(f"  {err}")
+
+        return "\n".join(lines)
+
+    @mcp.tool()
+    async def system_prune(confirmed: bool = False) -> str:
+        """Remove unused Docker resources: dangling images, stopped containers, unused networks.
+
+        Without confirmed=True, shows a preview of what would be removed.
+        With confirmed=True, runs the prune and reports reclaimed space.
+
+        Args:
+            confirmed: Must be True to actually prune. Default False shows
+                       a preview only.
+        """
+        # ── Gather preview data ───────────────────────────────────────────────
+        dangling_images: list[dict[str, Any]] = []
+        stopped_containers: list[dict[str, Any]] = []
+
+        try:
+            img_data = await client.request(
+                "SYNO.Docker.Image",
+                "list",
+                params={"limit": "-1", "offset": "0", "show_dsm": "false"},
+            )
+            ctr_data = await client.request(
+                "SYNO.Docker.Container",
+                "list",
+                params={"limit": "-1", "offset": "0", "type": "all"},
+            )
+        except Exception as e:
+            return f"Error fetching resource list: {e}"
+
+        images: list[dict[str, Any]] = img_data.get("images", [])
+        containers: list[dict[str, Any]] = ctr_data.get("containers", [])
+
+        # Images in use by any container
+        in_use_ids: set[str] = {
+            ctr.get("image_id") or ctr.get("ImageID") or ctr.get("Image", "") for ctr in containers
+        } - {""}
+
+        # Dangling = untagged (<none>) or unused
+        for img in images:
+            tags = img.get("tags") or []
+            is_untagged = not tags or tags == ["<none>"]
+            is_unused = img.get("id", "") not in in_use_ids
+            if is_untagged or is_unused:
+                dangling_images.append(img)
+
+        for ctr in containers:
+            if ctr.get("status") not in ("running", "up"):
+                stopped_containers.append(ctr)
+
+        dangling_size = sum(img.get("size", 0) for img in dangling_images)
+
+        if not confirmed:
+            lines = ["system_prune — preview (nothing deleted yet):", ""]
+            lines.append(
+                f"  Dangling/unused images:  {len(dangling_images)}  ({_human_size(dangling_size)})"
+            )
+            for img in dangling_images[:10]:
+                repo = img.get("repository", "<none>")
+                tags = img.get("tags") or ["<none>"]
+                lines.append(f"    - {repo}:{tags[0]}")
+            if len(dangling_images) > 10:
+                lines.append(f"    … and {len(dangling_images) - 10} more")
+
+            lines.append(f"  Stopped containers:      {len(stopped_containers)}")
+            for ctr in stopped_containers[:10]:
+                lines.append(f"    - {ctr.get('name', '?')}")
+            if len(stopped_containers) > 10:
+                lines.append(f"    … and {len(stopped_containers) - 10} more")
+
+            lines.append("  Unused networks:         (not counted — run prune to remove)")
+            lines.append("")
+            lines.append(
+                f"Call system_prune(confirmed=True) to free ~{_human_size(dangling_size)}."
+            )
+            return "\n".join(lines)
+
+        # ── Execute prune ─────────────────────────────────────────────────────
+        try:
+            result = await client.request("SYNO.Docker.Utils", "prune")
+        except Exception as e:
+            return f"Error running system prune: {e}"
+
+        # Parse reclaimed space from response (field names vary by DSM version)
+        reclaimed = (
+            result.get("SpaceReclaimed")
+            or result.get("space_reclaimed")
+            or result.get("reclaimed")
+            or 0
+        )
+
+        lines = ["system_prune — completed.", ""]
+        if reclaimed:
+            lines.append(f"  Space reclaimed: {_human_size(int(reclaimed))}")
+        else:
+            lines.append("  Space reclaimed: (not reported by DSM)")
+
+        # Surface any containers/images counts from the response
+        for key in ("ContainersDeleted", "ImagesDeleted", "VolumesDeleted", "NetworksDeleted"):
+            val = result.get(key)
+            if val is not None:
+                label = key.replace("Deleted", " deleted")
+                lines.append(f"  {label}: {len(val) if isinstance(val, list) else val}")
+
+        return "\n".join(lines)

src/mcp_synology_container/server.py

@@ -26,15 +26,17 @@ def create_server(config: AppConfig, client: DsmClient) -> FastMCP:
     """
     mcp = FastMCP("mcp-synology-container")
 
-    from mcp_synology_container.modules.projects import register_projects
-    from mcp_synology_container.modules.containers import register_containers
     from mcp_synology_container.modules.compose import register_compose
+    from mcp_synology_container.modules.containers import register_containers
     from mcp_synology_container.modules.images import register_images
+    from mcp_synology_container.modules.projects import register_projects
+    from mcp_synology_container.modules.system import register_system
 
     register_projects(mcp, config, client)
     register_containers(mcp, config, client)
     register_compose(mcp, config, client)
     register_images(mcp, config, client)
+    register_system(mcp, config, client)
 
     logger.info("MCP server configured with all tool modules")
     return mcp

tests/test_modules/test_system.py

@@ -0,0 +1,307 @@
+"""Tests for modules/system.py."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+
+def make_mock_mcp():
+    tools: dict = {}
+
+    class MockMCP:
+        def tool(self):
+            def decorator(fn):
+                tools[fn.__name__] = fn
+                return fn
+
+            return decorator
+
+    return MockMCP(), tools
+
+
+def make_config():
+    from mcp_synology_container.config import AppConfig, ConnectionConfig
+
+    return AppConfig(
+        schema_version=1,
+        connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
+    )
+
+
+SAMPLE_IMAGES = {
+    "images": [
+        {
+            "id": "sha256:aaaa",
+            "repository": "nginx",
+            "tags": ["1.24"],
+            "size": 50 * 1024 * 1024,
+        },
+        {
+            "id": "sha256:bbbb",
+            "repository": "postgres",
+            "tags": ["15"],
+            "size": 80 * 1024 * 1024,
+        },
+        {
+            "id": "sha256:cccc",
+            "repository": "<none>",
+            "tags": [],
+            "size": 10 * 1024 * 1024,
+        },
+    ]
+}
+
+SAMPLE_CONTAINERS = {
+    "containers": [
+        {"name": "web", "status": "running", "image_id": "sha256:aaaa"},
+        {"name": "db", "status": "running", "image_id": "sha256:bbbb"},
+        {"name": "old", "status": "stopped", "image_id": "sha256:aaaa"},
+    ]
+}
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+#  system_df
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_system_df_shows_image_stats():
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_df"]()
+    assert "Images" in result
+    assert "3" in result  # total images
+    assert "Containers" in result
+
+
+@pytest.mark.asyncio
+async def test_system_df_reclaimable():
+    """Unused images (not referenced by any container) are counted as reclaimable."""
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_df"]()
+    # sha256:cccc (<none>) is not referenced by any container → reclaimable
+    assert "reclaimable" in result
+    assert "unused" in result
+
+
+@pytest.mark.asyncio
+async def test_system_df_running_vs_stopped():
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_df"]()
+    assert "running" in result
+    assert "stopped" in result
+
+
+@pytest.mark.asyncio
+async def test_system_df_image_api_error_graceful():
+    """Container data is still shown even when image API fails."""
+    from mcp_synology_container.dsm_client import SynologyError
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            raise SynologyError("image API down", code=102)
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_df"]()
+    # Should still show container section and a warning
+    assert "Containers" in result or "Warnings" in result
+
+
+@pytest.mark.asyncio
+async def test_system_df_no_images():
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return {"images": []}
+        if api == "SYNO.Docker.Container":
+            return {"containers": []}
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_df"]()
+    assert "0" in result
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+#  system_prune
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_system_prune_preview():
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_prune"]()
+    assert "preview" in result.lower()
+    assert "confirmed=True" in result
+    # prune API must NOT be called
+    calls = [c.args[:2] for c in client.request.call_args_list]
+    assert ("SYNO.Docker.Utils", "prune") not in calls
+
+
+@pytest.mark.asyncio
+async def test_system_prune_preview_lists_dangling():
+    """Preview names dangling/unused images and stopped containers."""
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_prune"]()
+    # <none>:<none> is dangling
+    assert "<none>" in result
+    # "old" container is stopped
+    assert "old" in result
+
+
+@pytest.mark.asyncio
+async def test_system_prune_confirmed():
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        if api == "SYNO.Docker.Utils" and method == "prune":
+            return {"SpaceReclaimed": 10 * 1024 * 1024}
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_prune"](confirmed=True)
+    assert "completed" in result.lower()
+    assert "MiB" in result or "reclaimed" in result.lower()
+
+
+@pytest.mark.asyncio
+async def test_system_prune_confirmed_no_space_reported():
+    """Prune works even when DSM doesn't report reclaimed bytes."""
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        if api == "SYNO.Docker.Utils" and method == "prune":
+            return {}
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_prune"](confirmed=True)
+    assert "completed" in result.lower()
+    assert "not reported" in result
+
+
+@pytest.mark.asyncio
+async def test_system_prune_api_error():
+    from mcp_synology_container.dsm_client import SynologyError
+    from mcp_synology_container.modules.system import register_system
+
+    client = AsyncMock()
+
+    async def mock_request(api, method, **kwargs):
+        if api == "SYNO.Docker.Image":
+            return SAMPLE_IMAGES
+        if api == "SYNO.Docker.Container":
+            return SAMPLE_CONTAINERS
+        if api == "SYNO.Docker.Utils" and method == "prune":
+            raise SynologyError("prune failed", code=100)
+        return {}
+
+    client.request.side_effect = mock_request
+    mcp, tools = make_mock_mcp()
+    register_system(mcp, make_config(), client)
+
+    result = await tools["system_prune"](confirmed=True)
+    assert "Error" in result