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

"""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)