Initial implementation
This commit is contained in:
@@ -0,0 +1,215 @@
|
||||
"""MCP tools for SYNO.Docker.Container: list, status, logs, exec."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
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_containers(mcp: FastMCP, config: AppConfig, client: DsmClient) -> None:
|
||||
"""Register all container management tools with the MCP server."""
|
||||
|
||||
@mcp.tool()
|
||||
async def list_containers(project_name: str | None = None) -> str:
|
||||
"""List containers, optionally filtered by project name.
|
||||
|
||||
Args:
|
||||
project_name: Optional project name to filter containers.
|
||||
If omitted, lists all containers.
|
||||
"""
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.Docker.Container",
|
||||
"list",
|
||||
params={"limit": "-1", "offset": "0", "type": "all"},
|
||||
)
|
||||
except Exception as e:
|
||||
return f"Error listing containers: {e}"
|
||||
|
||||
containers: list[dict[str, Any]] = data.get("containers", [])
|
||||
if not containers:
|
||||
return "No containers found."
|
||||
|
||||
# Filter by project if specified
|
||||
if project_name:
|
||||
containers = [
|
||||
c for c in containers
|
||||
if c.get("project_name") == project_name
|
||||
or _container_in_project(c, project_name)
|
||||
]
|
||||
if not containers:
|
||||
return f"No containers found for project '{project_name}'."
|
||||
|
||||
lines = [f"Containers ({len(containers)} total):", ""]
|
||||
for container in sorted(containers, key=lambda c: c.get("name", "")):
|
||||
name = container.get("name", "?")
|
||||
state = container.get("status", container.get("state", "?"))
|
||||
image = container.get("image", "?")
|
||||
lines.append(f" {name}")
|
||||
lines.append(f" Status: {state}")
|
||||
lines.append(f" Image: {image}")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines).rstrip()
|
||||
|
||||
@mcp.tool()
|
||||
async def get_container_status(container_name: str) -> str:
|
||||
"""Get detailed status, uptime, and resource usage of a container.
|
||||
|
||||
Args:
|
||||
container_name: Name of the container to inspect.
|
||||
"""
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.Docker.Container",
|
||||
"get",
|
||||
params={"name": container_name},
|
||||
)
|
||||
except Exception as e:
|
||||
return f"Error getting container '{container_name}': {e}"
|
||||
|
||||
if not data:
|
||||
return f"Container '{container_name}' not found."
|
||||
|
||||
return _format_container_detail(container_name, data)
|
||||
|
||||
@mcp.tool()
|
||||
async def get_container_logs(
|
||||
container_name: str,
|
||||
tail: int = 100,
|
||||
keyword: str | None = None,
|
||||
) -> str:
|
||||
"""Get log output from a container.
|
||||
|
||||
Args:
|
||||
container_name: Name of the container.
|
||||
tail: Number of recent log lines to return (default 100).
|
||||
keyword: Optional keyword to filter log lines.
|
||||
"""
|
||||
params: dict[str, Any] = {
|
||||
"name": container_name,
|
||||
"limit": tail,
|
||||
"offset": 0,
|
||||
"sort_dir": "DESC",
|
||||
}
|
||||
if keyword:
|
||||
params["keyword"] = keyword
|
||||
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.Docker.Container.Log",
|
||||
"get",
|
||||
params=params,
|
||||
)
|
||||
except Exception as e:
|
||||
return f"Error getting logs for '{container_name}': {e}"
|
||||
|
||||
logs: list[dict[str, Any]] = data.get("logs", [])
|
||||
if not logs:
|
||||
return f"No logs found for container '{container_name}'."
|
||||
|
||||
total = data.get("total", len(logs))
|
||||
header = f"Logs for {container_name} (showing {len(logs)} of {total}):\n"
|
||||
|
||||
# Logs are returned in DESC order, reverse for chronological display
|
||||
lines = []
|
||||
for entry in reversed(logs):
|
||||
timestamp = entry.get("created", "")
|
||||
stream = entry.get("stream", "")
|
||||
text = entry.get("text", "")
|
||||
stream_tag = f"[{stream}] " if stream else ""
|
||||
lines.append(f"{timestamp} {stream_tag}{text}")
|
||||
|
||||
return header + "\n".join(lines)
|
||||
|
||||
@mcp.tool()
|
||||
async def exec_in_container(
|
||||
container_name: str,
|
||||
command: str,
|
||||
confirmed: bool = False,
|
||||
) -> str:
|
||||
"""Execute a command in a running container.
|
||||
|
||||
This executes a shell command inside the container. Use with caution.
|
||||
Requires confirmation before executing.
|
||||
|
||||
Args:
|
||||
container_name: Name of the container.
|
||||
command: Shell command to execute.
|
||||
confirmed: Must be True to proceed. Set to True to confirm execution.
|
||||
"""
|
||||
if not confirmed:
|
||||
return (
|
||||
f"About to run in container '{container_name}':\n"
|
||||
f" $ {command}\n\n"
|
||||
f"Call this tool again with confirmed=True to proceed."
|
||||
)
|
||||
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.Docker.Container",
|
||||
"exec",
|
||||
params={
|
||||
"name": container_name,
|
||||
"command": command,
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
return f"Error executing command in '{container_name}': {e}"
|
||||
|
||||
output = data.get("output", "")
|
||||
exit_code = data.get("exit_code", 0)
|
||||
|
||||
result_lines = [f"Command executed in '{container_name}':"]
|
||||
result_lines.append(f" Exit code: {exit_code}")
|
||||
if output:
|
||||
result_lines.append(" Output:")
|
||||
result_lines.append(output)
|
||||
|
||||
return "\n".join(result_lines)
|
||||
|
||||
|
||||
def _container_in_project(container: dict[str, Any], project_name: str) -> bool:
|
||||
"""Check if a container belongs to a project based on its labels."""
|
||||
labels = container.get("labels", {}) or {}
|
||||
if isinstance(labels, dict):
|
||||
return labels.get("com.docker.compose.project") == project_name
|
||||
return False
|
||||
|
||||
|
||||
def _format_container_detail(name: str, data: dict[str, Any]) -> str:
|
||||
"""Format container inspect data as human-readable text."""
|
||||
state = data.get("State", {}) or {}
|
||||
config = data.get("Config", {}) or {}
|
||||
host_config = data.get("HostConfig", {}) or {}
|
||||
|
||||
lines = [
|
||||
f"Container: {name}",
|
||||
f" Status: {state.get('Status', '?')}",
|
||||
f" Running: {state.get('Running', False)}",
|
||||
f" Image: {config.get('Image', '?')}",
|
||||
]
|
||||
|
||||
if state.get("StartedAt"):
|
||||
lines.append(f" Started: {state.get('StartedAt')}")
|
||||
if state.get("FinishedAt") and not state.get("Running"):
|
||||
lines.append(f" Finished: {state.get('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")
|
||||
|
||||
env = config.get("Env", []) or []
|
||||
if env:
|
||||
lines.append(f" Env vars: {len(env)}")
|
||||
|
||||
return "\n".join(lines)
|
||||
Reference in New Issue
Block a user