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

Fix get_container_status: clean name + dual-format response handling

Browse Source 
DSM response has two top-level keys:
  details → Docker Engine inspect (State, NetworkSettings, Mounts)
  profile → DSM format (image, port_bindings)

_format_container_detail now reads:
  Status/Running/StartedAt from details.State
  Image               from profile.image
  IP addresses        from details.NetworkSettings.Networks
  Port bindings       from profile.port_bindings
  Mounts              from details.Mounts

Also: debug_container_response tool removed, json/sys imports cleaned up.
27 container tests all passing.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Marcus van Elst
2026-04-14 06:59:50 +02:00
parent 2dbcc0ba5f
commit d9f0e75d0a
2 changed files with 115 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mcp_synology_container/modules/containers.py
+39 -56
View File
@@ -2,10 +2,8 @@
from __future__ import annotations
import json
import logging
import re
import sys
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
@@ -108,29 +106,6 @@ def register_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> N
return "\n".join(lines).rstrip()
@mcp.tool()
async def debug_container_response(container_name: str) -> str:
"""TEMPORARY DEBUG TOOL — returns raw SYNO.Docker.Container/get response as JSON.
Used to inspect the actual DSM response structure so that
get_container_status can be fixed to read the correct fields.
Remove once the real fix is in place.
Args:
container_name: Name of the container to inspect.
"""
clean_name = _strip_hash_prefix(container_name)
try:
data = await client.request(
"SYNO.Docker.Container",
"get",
params={"name": clean_name},
)
except Exception as e:
return f"Error calling SYNO.Docker.Container/get for '{clean_name}': {e}"
return json.dumps(data, indent=2, default=str)
@mcp.tool()
async def get_container_status(container_name: str) -> str:
"""Get detailed status, uptime, and resource usage of a container.
@@ -138,9 +113,7 @@ def register_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> N
Args:
container_name: Name of the container to inspect.
"""
# Strip hash prefix from user input; SYNO.Docker.Container/get accepts
# the clean name directly — do NOT resolve to hash-prefixed name, as
# the endpoint may not accept that form.
# SYNO.Docker.Container/get accepts the clean name (no hash prefix).
clean_name = _strip_hash_prefix(container_name)
try:
data = await client.request(
@@ -151,12 +124,6 @@ def register_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> N
except Exception as e:
return f"Error getting container '{container_name}': {e}"
sys.stderr.write(
f"[get_container_status] name={clean_name!r} "
f"raw response: {json.dumps(data, indent=2, default=str)[:2000]}\n"
)
sys.stderr.flush()
if not data:
return f"Container '{container_name}' not found."
@@ -357,23 +324,17 @@ def _container_in_project(container: dict[str, Any], project_name: str) -> bool:
def _format_container_detail(name: str, data: dict[str, Any]) -> str:
"""Format container inspect data as human-readable text.
DSM may return the container object under a "container" key, or directly
at the top level. Supports both the Docker Engine inspect format
(nested State/Config/HostConfig) and the DSM flat format (lowercase
status/image fields as returned by the list endpoint).
SYNO.Docker.Container/get returns two top-level keys:
- "details": Docker Engine inspect format (State, NetworkSettings, Mounts, …)
- "profile": DSM format (image, port_bindings, …)
"""
# Unwrap "container" wrapper if present (common DSM API pattern)
raw = data["container"] if "container" in data and isinstance(data["container"], dict) else data
details: dict[str, Any] = data.get("details", {}) or {}
profile: dict[str, Any] = data.get("profile", {}) or {}
# Docker Engine inspect format: State / Config / HostConfig
state = raw.get("State", {}) or {}
config = raw.get("Config", {}) or {}
host_config = raw.get("HostConfig", {}) or {}
# DSM flat format fallback (lowercase fields, same as list response)
status_str = state.get("Status") or raw.get("status") or "?"
running = state.get("Running") if "Running" in state else (status_str == "running")
image_str = config.get("Image") or raw.get("image") or "?"
state: dict[str, Any] = details.get("State", {}) or {}
status_str = state.get("Status", "?")
running = state.get("Running", False)
image_str = profile.get("image", "?")
lines = [
f"Container: {name}",
@@ -388,13 +349,35 @@ def _format_container_detail(name: str, data: dict[str, Any]) -> str:
lines.append(f" Finished: {state['FinishedAt']}")
lines.append(f" Exit code: {state.get('ExitCode', '?')}")
memory = host_config.get("Memory", 0)
if memory:
mb = memory // (1024 * 1024)
lines.append(f" Memory limit: {mb} MiB")
# IP addresses from all attached networks
networks: dict[str, Any] = (
details.get("NetworkSettings", {}) or {}
).get("Networks", {}) or {}
for net_name, net in networks.items():
ip = (net or {}).get("IPAddress", "")
if ip:
lines.append(f" IP ({net_name}): {ip}")
env = config.get("Env", []) or []
if env:
lines.append(f" Env vars: {len(env)}")
# Port bindings from DSM profile
port_bindings: list[dict[str, Any]] = profile.get("port_bindings", []) or []
if port_bindings:
lines.append(" Ports:")
for pb in port_bindings:
host = pb.get("host_port", "?")
ctr = pb.get("container_port", "?")
proto = pb.get("type", "tcp")
lines.append(f" {host} → {ctr}/{proto}")
# Mounts from Docker Engine inspect
mounts: list[dict[str, Any]] = details.get("Mounts", []) or []
if mounts:
lines.append(" Mounts:")
for m in mounts:
src = m.get("Source", "?")
dst = m.get("Destination", "?")
mtype = m.get("Type", "")
rw = "" if m.get("RW", True) else " [ro]"
type_tag = f" ({mtype})" if mtype else ""
lines.append(f" {src} → {dst}{type_tag}{rw}")
return "\n".join(lines)
tests/test_modules/test_containers.py
+76 -28
View File
@@ -407,18 +407,48 @@ async def test_container_stats_strips_hash_prefix():
assert "CPU" in result
DSM_CONTAINER_RESPONSE = {
"details": {
"State": {
"Status": "running",
"Running": True,
"StartedAt": "2025-01-01T10:00:00Z",
"FinishedAt": "0001-01-01T00:00:00Z",
"ExitCode": 0,
},
"NetworkSettings": {
"Networks": {
"bridge": {"IPAddress": "172.17.0.2"},
}
},
"Mounts": [
{
"Type": "bind",
"Source": "/volume1/docker/jenkins",
"Destination": "/var/jenkins_home",
"RW": True,
}
],
},
"profile": {
"image": "jenkins/jenkins:2.558-jdk21",
"port_bindings": [
{"host_port": 8080, "container_port": 8080, "type": "tcp"},
{"host_port": 50000, "container_port": 50000, "type": "tcp"},
],
},
}
@pytest.mark.asyncio
async def test_get_container_status_uses_clean_name():
"""get_container_status strips hash prefix and calls get with clean name."""
from mcp_synology_container.modules.containers import register_containers
detail_data = {"State": {"Status": "running", "Running": True}, "Config": {"Image": "jenkins/jenkins:lts"}}
async def mock_request(api, method, **kwargs):
if api == "SYNO.Docker.Container" and method == "get":
# Must be called with the CLEAN name (no hash prefix)
assert kwargs["params"]["name"] == "jenkins"
return detail_data
return DSM_CONTAINER_RESPONSE
return {}
client = AsyncMock()
@@ -427,56 +457,74 @@ async def test_get_container_status_uses_clean_name():
mcp, tools = make_mock_mcp()
register_containers(mcp, make_config(), client)
# User passes hash-prefixed name → should be stripped before get call
# User passes hash-prefixed name → stripped before get call
result = await tools["get_container_status"]("f93cb8b504f7_jenkins")
assert "running" in result
assert "f93cb8b504f7" not in result
@pytest.mark.asyncio
async def test_get_container_status_container_key_unwrap():
"""get_container_status unwraps DSM 'container' wrapper key."""
async def test_get_container_status_details_profile_structure():
"""get_container_status reads status from details.State, image from profile."""
from mcp_synology_container.modules.containers import register_containers
# DSM may wrap the inspect data under a "container" key
wrapped_data = {
"container": {
"State": {"Status": "running", "Running": True},
"Config": {"Image": "jenkins/jenkins:lts"},
}
}
client = AsyncMock()
client.request.return_value = wrapped_data
client.request.return_value = DSM_CONTAINER_RESPONSE
mcp, tools = make_mock_mcp()
register_containers(mcp, make_config(), client)
result = await tools["get_container_status"]("jenkins")
assert "running" in result
assert "jenkins/jenkins:lts" in result
assert "jenkins/jenkins:2.558-jdk21" in result
assert "True" in result # Running field
@pytest.mark.asyncio
async def test_get_container_status_flat_format():
"""get_container_status handles DSM flat format (lowercase status/image)."""
async def test_get_container_status_shows_ip():
"""get_container_status shows IP address from NetworkSettings."""
from mcp_synology_container.modules.containers import register_containers
flat_data = {
"status": "running",
"image": "jenkins/jenkins:lts",
"name": "jenkins",
}
client = AsyncMock()
client.request.return_value = flat_data
client.request.return_value = DSM_CONTAINER_RESPONSE
mcp, tools = make_mock_mcp()
register_containers(mcp, make_config(), client)
result = await tools["get_container_status"]("jenkins")
assert "running" in result
assert "jenkins/jenkins:lts" in result
assert "172.17.0.2" in result
@pytest.mark.asyncio
async def test_get_container_status_shows_ports():
"""get_container_status shows port bindings from profile."""
from mcp_synology_container.modules.containers import register_containers
client = AsyncMock()
client.request.return_value = DSM_CONTAINER_RESPONSE
mcp, tools = make_mock_mcp()
register_containers(mcp, make_config(), client)
result = await tools["get_container_status"]("jenkins")
assert "8080" in result
assert "50000" in result
@pytest.mark.asyncio
async def test_get_container_status_shows_mounts():
"""get_container_status shows mount paths from details.Mounts."""
from mcp_synology_container.modules.containers import register_containers
client = AsyncMock()
client.request.return_value = DSM_CONTAINER_RESPONSE
mcp, tools = make_mock_mcp()
register_containers(mcp, make_config(), client)
result = await tools["get_container_status"]("jenkins")
assert "/volume1/docker/jenkins" in result
assert "/var/jenkins_home" in result
@pytest.mark.asyncio