unraid-mcp/unraid_mcp/tools/settings.py

"""System settings, time, UPS, and remote access mutations.

Provides the `unraid_settings` tool with 9 actions for updating system
configuration, time settings, UPS, API settings, and Unraid Connect.
"""

from typing import Any, Literal, get_args

from fastmcp import FastMCP

from ..config.logging import logger
from ..core.client import make_graphql_request
from ..core.exceptions import ToolError, tool_error_handler


MUTATIONS: dict[str, str] = {
    "update": """
        mutation UpdateSettings($input: JSON!) {
          updateSettings(input: $input) { restartRequired values warnings }
        }
    """,
    "update_temperature": """
        mutation UpdateTemperatureConfig($input: TemperatureConfigInput!) {
          updateTemperatureConfig(input: $input)
        }
    """,
    "update_time": """
        mutation UpdateSystemTime($input: UpdateSystemTimeInput!) {
          updateSystemTime(input: $input) { currentTime timeZone useNtp ntpServers }
        }
    """,
    "configure_ups": """
        mutation ConfigureUps($config: UPSConfigInput!) {
          configureUps(config: $config)
        }
    """,
    "update_api": """
        mutation UpdateApiSettings($input: ConnectSettingsInput!) {
          updateApiSettings(input: $input) { accessType forwardType port }
        }
    """,
    "connect_sign_in": """
        mutation ConnectSignIn($input: ConnectSignInInput!) {
          connectSignIn(input: $input)
        }
    """,
    "connect_sign_out": """
        mutation ConnectSignOut {
          connectSignOut
        }
    """,
    "setup_remote_access": """
        mutation SetupRemoteAccess($input: SetupRemoteAccessInput!) {
          setupRemoteAccess(input: $input)
        }
    """,
    "enable_dynamic_remote_access": """
        mutation EnableDynamicRemoteAccess($input: EnableDynamicRemoteAccessInput!) {
          enableDynamicRemoteAccess(input: $input)
        }
    """,
}

DESTRUCTIVE_ACTIONS = {"configure_ups", "setup_remote_access", "enable_dynamic_remote_access"}
ALL_ACTIONS = set(MUTATIONS)

SETTINGS_ACTIONS = Literal[
    "update",
    "update_temperature",
    "update_time",
    "configure_ups",
    "update_api",
    "connect_sign_in",
    "connect_sign_out",
    "setup_remote_access",
    "enable_dynamic_remote_access",
]

if set(get_args(SETTINGS_ACTIONS)) != ALL_ACTIONS:
    _missing = ALL_ACTIONS - set(get_args(SETTINGS_ACTIONS))
    _extra = set(get_args(SETTINGS_ACTIONS)) - ALL_ACTIONS
    raise RuntimeError(
        f"SETTINGS_ACTIONS and ALL_ACTIONS are out of sync. "
        f"Missing from Literal: {_missing or 'none'}. Extra in Literal: {_extra or 'none'}"
    )


def register_settings_tool(mcp: FastMCP) -> None:
    """Register the unraid_settings tool with the FastMCP instance."""

    @mcp.tool()
    async def unraid_settings(
        action: SETTINGS_ACTIONS,
        confirm: bool = False,
        settings_input: dict[str, Any] | None = None,
        temperature_config: dict[str, Any] | None = None,
        time_zone: str | None = None,
        use_ntp: bool | None = None,
        ntp_servers: list[str] | None = None,
        manual_datetime: str | None = None,
        ups_config: dict[str, Any] | None = None,
        access_type: str | None = None,
        forward_type: str | None = None,
        port: int | None = None,
        api_key: str | None = None,
        username: str | None = None,
        email: str | None = None,
        avatar: str | None = None,
        access_url_type: str | None = None,
        access_url_name: str | None = None,
        access_url_ipv4: str | None = None,
        access_url_ipv6: str | None = None,
        dynamic_enabled: bool | None = None,
    ) -> dict[str, Any]:
        """Update Unraid system settings, time, UPS, and remote access configuration.

        Actions:
          update - Update system settings (requires settings_input dict)
          update_temperature - Update temperature sensor config (requires temperature_config dict)
          update_time - Update time/timezone/NTP (requires at least one of: time_zone, use_ntp, ntp_servers, manual_datetime)
          configure_ups - Configure UPS monitoring (requires ups_config dict, confirm=True)
          update_api - Update API/Connect settings (requires at least one of: access_type, forward_type, port)
          connect_sign_in - Sign in to Unraid Connect (requires api_key)
          connect_sign_out - Sign out from Unraid Connect
          setup_remote_access - Configure remote access (requires access_type, confirm=True)
          enable_dynamic_remote_access - Enable/disable dynamic remote access (requires access_url_type, dynamic_enabled, confirm=True)
        """
        if action not in ALL_ACTIONS:
            raise ToolError(f"Invalid action '{action}'. Must be one of: {sorted(ALL_ACTIONS)}")

        if action in DESTRUCTIVE_ACTIONS and not confirm:
            raise ToolError(f"Action '{action}' is destructive. Set confirm=True to proceed.")

        with tool_error_handler("settings", action, logger):
            logger.info(f"Executing unraid_settings action={action}")

            if action == "update":
                if settings_input is None:
                    raise ToolError("settings_input is required for 'update' action")
                data = await make_graphql_request(MUTATIONS["update"], {"input": settings_input})
                return {"success": True, "action": "update", "data": data.get("updateSettings")}

            if action == "update_temperature":
                if temperature_config is None:
                    raise ToolError("temperature_config is required for 'update_temperature' action")
                data = await make_graphql_request(
                    MUTATIONS["update_temperature"], {"input": temperature_config}
                )
                return {"success": True, "action": "update_temperature", "result": data.get("updateTemperatureConfig")}

            if action == "update_time":
                time_input: dict[str, Any] = {}
                if time_zone is not None:
                    time_input["timeZone"] = time_zone
                if use_ntp is not None:
                    time_input["useNtp"] = use_ntp
                if ntp_servers is not None:
                    time_input["ntpServers"] = ntp_servers
                if manual_datetime is not None:
                    time_input["manualDateTime"] = manual_datetime
                if not time_input:
                    raise ToolError(
                        "update_time requires at least one of: time_zone, use_ntp, ntp_servers, manual_datetime"
                    )
                data = await make_graphql_request(MUTATIONS["update_time"], {"input": time_input})
                return {"success": True, "action": "update_time", "data": data.get("updateSystemTime")}

            if action == "configure_ups":
                if ups_config is None:
                    raise ToolError("ups_config is required for 'configure_ups' action")
                data = await make_graphql_request(MUTATIONS["configure_ups"], {"config": ups_config})
                return {"success": True, "action": "configure_ups", "result": data.get("configureUps")}

            if action == "update_api":
                api_input: dict[str, Any] = {}
                if access_type is not None:
                    api_input["accessType"] = access_type
                if forward_type is not None:
                    api_input["forwardType"] = forward_type
                if port is not None:
                    api_input["port"] = port
                if not api_input:
                    raise ToolError(
                        "update_api requires at least one of: access_type, forward_type, port"
                    )
                data = await make_graphql_request(MUTATIONS["update_api"], {"input": api_input})
                return {"success": True, "action": "update_api", "data": data.get("updateApiSettings")}

            if action == "connect_sign_in":
                if not api_key:
                    raise ToolError("api_key is required for 'connect_sign_in' action")
                sign_in_input: dict[str, Any] = {"apiKey": api_key}
                if username or email:
                    user_info: dict[str, Any] = {}
                    if username:
                        user_info["preferred_username"] = username
                    if email:
                        user_info["email"] = email
                    if avatar:
                        user_info["avatar"] = avatar
                    sign_in_input["userInfo"] = user_info
                data = await make_graphql_request(
                    MUTATIONS["connect_sign_in"], {"input": sign_in_input}
                )
                return {"success": True, "action": "connect_sign_in", "result": data.get("connectSignIn")}

            if action == "connect_sign_out":
                data = await make_graphql_request(MUTATIONS["connect_sign_out"])
                return {"success": True, "action": "connect_sign_out", "result": data.get("connectSignOut")}

            if action == "setup_remote_access":
                if not access_type:
                    raise ToolError("access_type is required for 'setup_remote_access' action")
                remote_input: dict[str, Any] = {"accessType": access_type}
                if forward_type is not None:
                    remote_input["forwardType"] = forward_type
                if port is not None:
                    remote_input["port"] = port
                data = await make_graphql_request(
                    MUTATIONS["setup_remote_access"], {"input": remote_input}
                )
                return {"success": True, "action": "setup_remote_access", "result": data.get("setupRemoteAccess")}

            if action == "enable_dynamic_remote_access":
                if not access_url_type:
                    raise ToolError(
                        "access_url_type is required for 'enable_dynamic_remote_access' action"
                    )
                if dynamic_enabled is None:
                    raise ToolError(
                        "dynamic_enabled is required for 'enable_dynamic_remote_access' action"
                    )
                url_input: dict[str, Any] = {"type": access_url_type}
                if access_url_name is not None:
                    url_input["name"] = access_url_name
                if access_url_ipv4 is not None:
                    url_input["ipv4"] = access_url_ipv4
                if access_url_ipv6 is not None:
                    url_input["ipv6"] = access_url_ipv6
                data = await make_graphql_request(
                    MUTATIONS["enable_dynamic_remote_access"],
                    {"input": {"url": url_input, "enabled": dynamic_enabled}},
                )
                return {"success": True, "action": "enable_dynamic_remote_access", "result": data.get("enableDynamicRemoteAccess")}

            raise ToolError(f"Unhandled action '{action}' — this is a bug")

    logger.info("Settings tool registered successfully")