marcus/mcp-synology-container
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add list_networks, create_network, delete_network tools (Gruppe 4)

Browse Source 
list_networks:
  SYNO.Docker.Network/list → shows name, driver, subnet, gateway,
  attached containers. Response key is "network" (not "networks").

create_network(name, driver, subnet, gateway, ip_range, enable_ipv6, confirmed):
  Dry-run preview without confirmed=True. Passes enable_ipv6 as
  json.dumps(bool) per N4S4 reference. Optional params (subnet,
  gateway, iprange) omitted from request when not provided.

delete_network(name, confirmed):
  Validates network exists and has no attached containers before
  deleting. Clear error listing attached container names if blocked.

15 unit tests covering all paths.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Marcus van Elst
2026-04-13 18:42:10 +02:00
parent 6bdd2bcb6a
commit 4cee16922f
3 changed files with 501 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mcp_synology_container/modules/networks.py
+155
View File
@@ -0,0 +1,155 @@
"""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."
)
params: dict[str, Any] = {
"name": name,
"driver": driver,
"enable_ipv6": json.dumps(enable_ipv6),
}
if subnet is not None:
params["subnet"] = subnet
if gateway is not None:
params["gateway"] = gateway
if ip_range is not None:
params["iprange"] = ip_range
try:
result = await client.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."
)
try:
await client.request("SYNO.Docker.Network", "delete", params={"name": name})
except Exception as e:
return f"Error deleting network '{name}': {e}"
return f"Network '{name}' deleted."
src/mcp_synology_container/server.py
+2
View File
@@ -29,6 +29,7 @@ def create_server(config: AppConfig, client: DsmClient) -> FastMCP:
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.networks import register_networks
from mcp_synology_container.modules.projects import register_projects
from mcp_synology_container.modules.system import register_system
@@ -37,6 +38,7 @@ def create_server(config: AppConfig, client: DsmClient) -> FastMCP:
register_compose(mcp, config, client)
register_images(mcp, config, client)
register_system(mcp, config, client)
register_networks(mcp, config, client)
logger.info("MCP server configured with all tool modules")
return mcp
tests/test_modules/test_networks.py
+344
View File
@@ -0,0 +1,344 @@
"""Tests for modules/networks.py."""
import json
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_NETWORKS = {
"network": [
{
"id": "b741915823aa",
"name": "vault_default",
"driver": "bridge",
"subnet": "172.22.0.0/16",
"gateway": "172.22.0.1",
"iprange": "",
"enable_ipv6": False,
"containers": ["vault"],
},
{
"id": "2fc7ebae3901",
"name": "host",
"driver": "host",
"subnet": "",
"gateway": "",
"iprange": "",
"enable_ipv6": False,
"containers": [],
},
{
"id": "aabbcc112233",
"name": "my_bridge",
"driver": "bridge",
"subnet": "10.10.0.0/24",
"gateway": "10.10.0.1",
"iprange": "",
"enable_ipv6": False,
"containers": [],
},
]
}
# ──────────────────────────────────────────────────────────────────────────────
# list_networks
# ──────────────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_list_networks_shows_all():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["list_networks"]()
assert "vault_default" in result
assert "host" in result
assert "my_bridge" in result
assert "3 total" in result
@pytest.mark.asyncio
async def test_list_networks_shows_subnet_and_gateway():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["list_networks"]()
assert "172.22.0.0/16" in result
assert "172.22.0.1" in result
@pytest.mark.asyncio
async def test_list_networks_shows_attached_containers():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["list_networks"]()
assert "vault" in result
@pytest.mark.asyncio
async def test_list_networks_empty():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {"network": []}
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["list_networks"]()
assert "No networks found" in result
@pytest.mark.asyncio
async def test_list_networks_api_error():
from mcp_synology_container.dsm_client import SynologyError
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.side_effect = SynologyError("API error", code=102)
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["list_networks"]()
assert "Error" in result
# ──────────────────────────────────────────────────────────────────────────────
# create_network
# ──────────────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_create_network_preview():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["create_network"](name="mynet", driver="bridge")
assert "Preview" in result
assert "mynet" in result
assert "confirmed=True" in result
client.request.assert_not_called()
@pytest.mark.asyncio
async def test_create_network_confirmed():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {"id": "deadbeef1234567890"}
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["create_network"](
name="mynet", driver="bridge", subnet="192.168.100.0/24", confirmed=True
)
assert "created" in result
assert "mynet" in result
call = client.request.call_args
params = call.kwargs.get("params") or {}
assert params["name"] == "mynet"
assert params["driver"] == "bridge"
assert params["subnet"] == "192.168.100.0/24"
assert json.loads(params["enable_ipv6"]) is False
@pytest.mark.asyncio
async def test_create_network_with_ipv6():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {"id": "abc123"}
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
await tools["create_network"](name="ipv6net", enable_ipv6=True, confirmed=True)
call = client.request.call_args
params = call.kwargs.get("params") or {}
assert json.loads(params["enable_ipv6"]) is True
@pytest.mark.asyncio
async def test_create_network_optional_params_not_sent():
"""subnet/gateway/ip_range must not appear in params when not provided."""
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {}
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
await tools["create_network"](name="bare", confirmed=True)
call = client.request.call_args
params = call.kwargs.get("params") or {}
assert "subnet" not in params
assert "gateway" not in params
assert "iprange" not in params
@pytest.mark.asyncio
async def test_create_network_api_error():
from mcp_synology_container.dsm_client import SynologyError
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.side_effect = SynologyError("create failed", code=100)
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["create_network"](name="mynet", confirmed=True)
assert "Error" in result
# ──────────────────────────────────────────────────────────────────────────────
# delete_network
# ──────────────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_delete_network_preview():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["delete_network"](name="my_bridge")
assert "Preview" in result
assert "my_bridge" in result
assert "confirmed=True" in result
# Only the list call should have been made, not delete
assert client.request.call_count == 1
assert client.request.call_args.args[1] == "list"
@pytest.mark.asyncio
async def test_delete_network_confirmed():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
async def mock_request(api, method, **kwargs):
if method == "list":
return SAMPLE_NETWORKS
if method == "delete":
return {}
return {}
client.request.side_effect = mock_request
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["delete_network"](name="my_bridge", confirmed=True)
assert "deleted" in result
assert "my_bridge" in result
@pytest.mark.asyncio
async def test_delete_network_not_found():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["delete_network"](name="nonexistent", confirmed=True)
assert "not found" in result
@pytest.mark.asyncio
async def test_delete_network_blocked_by_containers():
"""Deletion must be refused when containers are attached."""
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = SAMPLE_NETWORKS
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
# vault_default has ["vault"] attached
result = await tools["delete_network"](name="vault_default", confirmed=True)
assert "Cannot delete" in result
assert "vault" in result
# delete API must not be called
assert client.request.call_count == 1
assert client.request.call_args.args[1] == "list"
@pytest.mark.asyncio
async def test_delete_network_api_error():
from mcp_synology_container.dsm_client import SynologyError
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
async def mock_request(api, method, **kwargs):
if method == "list":
return SAMPLE_NETWORKS
raise SynologyError("delete failed", code=100)
client.request.side_effect = mock_request
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
result = await tools["delete_network"](name="my_bridge", confirmed=True)
assert "Error" in result