mcp-synology-container/src/mcp_synology_container/modules/networks.py

"""MCP tools for SYNO.Docker.Network: list, create, delete."""

from __future__ import annotations

import json
import logging
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 register_networks(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
    """Register all network management tools with the MCP server."""

    @mcp.tool()
    async def list_networks() -> str:
        """List all Docker networks with driver, subnet, gateway, and attached containers."""
        try:
            data = await client.request("SYNO.Docker.Network", "list")
        except Exception as e:
            return f"Error listing networks: {e}"

        networks: list[dict[str, Any]] = data.get("network", [])
        if not networks:
            return "No networks found."

        lines = [f"Networks ({len(networks)} total):", ""]
        for net in sorted(networks, key=lambda n: n.get("name", "")):
            name = net.get("name", "?")
            driver = net.get("driver", "?")
            subnet = net.get("subnet") or "—"
            gateway = net.get("gateway") or "—"
            ipv6 = "IPv6" if net.get("enable_ipv6") else ""
            containers: list[str] = net.get("containers") or []

            lines.append(f"  {name}")
            lines.append(f"    Driver:   {driver}{('  ' + ipv6) if ipv6 else ''}")
            lines.append(f"    Subnet:   {subnet}")
            lines.append(f"    Gateway:  {gateway}")
            if containers:
                lines.append(f"    Containers ({len(containers)}): {', '.join(containers)}")
            else:
                lines.append("    Containers: none")
            lines.append("")

        return "\n".join(lines).rstrip()

    @mcp.tool()
    async def create_network(
        name: str,
        driver: str = "bridge",
        subnet: str | None = None,
        gateway: str | None = None,
        ip_range: str | None = None,
        enable_ipv6: bool = False,
        confirmed: bool = False,
    ) -> str:
        """Create a new Docker network.

        Args:
            name: Name for the new network.
            driver: Network driver. Defaults to "bridge".
            subnet: Subnet in CIDR notation (e.g. "172.28.0.0/16").
            gateway: Gateway IP address.
            ip_range: Allocatable IP range in CIDR notation.
            enable_ipv6: Enable IPv6 support. Defaults to False.
            confirmed: Must be True to actually create. Default False shows a preview.
        """
        details = [f"  Name:    {name}", f"  Driver:  {driver}"]
        if subnet:
            details.append(f"  Subnet:  {subnet}")
        if gateway:
            details.append(f"  Gateway: {gateway}")
        if ip_range:
            details.append(f"  IP range:{ip_range}")
        if enable_ipv6:
            details.append("  IPv6:    enabled")

        if not confirmed:
            return (
                "Preview: would create network:\n"
                + "\n".join(details)
                + f"\n\nCall create_network(name={name!r}, confirmed=True) to confirm."
            )

        # DevTools-confirmed POST format: all string/bool params as json.dumps.
        params: dict[str, Any] = {
            "name": json.dumps(name),
            "driver": driver,
            "enable_ipv6": json.dumps(enable_ipv6),
            "disable_masquerade": json.dumps(False),
        }
        if subnet is not None:
            params["subnet"] = json.dumps(subnet)
        if gateway is not None:
            params["gateway"] = json.dumps(gateway)
        if ip_range is not None:
            params["iprange"] = json.dumps(ip_range)

        try:
            result = await client.post_request("SYNO.Docker.Network", "create", params=params)
        except Exception as e:
            return f"Error creating network '{name}': {e}"

        net_id = result.get("id", "")
        id_str = f" (ID: {net_id[:12]})" if net_id else ""
        return f"Network '{name}' created{id_str}."

    @mcp.tool()
    async def delete_network(name: str, confirmed: bool = False) -> str:
        """Delete a Docker network by name.

        Refuses deletion if any container is still attached to the network.

        Args:
            name: Name of the network to delete.
            confirmed: Must be True to actually delete. Default False shows a preview.
        """
        # Fetch network list to validate existence and check for attached containers
        try:
            data = await client.request("SYNO.Docker.Network", "list")
        except Exception as e:
            return f"Error fetching network list: {e}"

        networks: list[dict[str, Any]] = data.get("network", [])
        target = next((n for n in networks if n.get("name") == name), None)

        if target is None:
            return f"Network '{name}' not found."

        attached: list[str] = target.get("containers") or []
        if attached:
            return (
                f"Cannot delete network '{name}': "
                f"{len(attached)} container(s) attached: {', '.join(attached)}"
            )

        if not confirmed:
            driver = target.get("driver", "?")
            subnet = target.get("subnet") or "—"
            return (
                f"Preview: would delete network '{name}' (driver={driver}, subnet={subnet}).\n"
                f"Call delete_network(name={name!r}, confirmed=True) to confirm."
            )

        # DevTools-confirmed POST format: method="remove", full network object
        # as a JSON array in the "networks" parameter.  The object must include
        # all fields from the list response plus "_key" set to the network ID.
        net_id = target.get("id", "")
        network_obj: dict[str, Any] = {
            "containers": target.get("containers") or [],
            "driver": target.get("driver", ""),
            "enable_ipv6": target.get("enable_ipv6", False),
            "ipv6_gateway": target.get("ipv6_gateway", ""),
            "ipv6_subnet": target.get("ipv6_subnet", ""),
            "ipv6_iprange": target.get("ipv6_iprange", ""),
            "gateway": target.get("gateway", ""),
            "id": net_id,
            "iprange": target.get("iprange", ""),
            "name": name,
            "subnet": target.get("subnet", ""),
            "disable_masquerade": target.get("disable_masquerade", False),
            "_key": net_id,
        }

        try:
            await client.post_request(
                "SYNO.Docker.Network",
                "remove",
                params={"networks": json.dumps([network_obj])},
            )
        except Exception as e:
            return f"Error deleting network '{name}': {e}"

        return f"Network '{name}' deleted."