unraid-mcp/unraid_mcp/tools/rclone.py

"""RClone cloud storage remote management.

Provides the `unraid_rclone` tool with 4 actions for managing
cloud storage remotes (S3, Google Drive, Dropbox, FTP, etc.).
"""

from typing import Any, Literal

from fastmcp import FastMCP

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


QUERIES: dict[str, str] = {
    "list_remotes": """
        query ListRCloneRemotes {
          rclone { remotes { name type parameters config } }
        }
    """,
    "config_form": """
        query GetRCloneConfigForm($formOptions: RCloneConfigFormInput) {
          rclone { configForm(formOptions: $formOptions) { id dataSchema uiSchema } }
        }
    """,
}

MUTATIONS: dict[str, str] = {
    "create_remote": """
        mutation CreateRCloneRemote($input: CreateRCloneRemoteInput!) {
          rclone { createRCloneRemote(input: $input) { name type parameters } }
        }
    """,
    "delete_remote": """
        mutation DeleteRCloneRemote($input: DeleteRCloneRemoteInput!) {
          rclone { deleteRCloneRemote(input: $input) }
        }
    """,
}

DESTRUCTIVE_ACTIONS = {"delete_remote"}
ALL_ACTIONS = set(QUERIES) | set(MUTATIONS)

RCLONE_ACTIONS = Literal[
    "list_remotes",
    "config_form",
    "create_remote",
    "delete_remote",
]


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

    @mcp.tool()
    async def unraid_rclone(
        action: RCLONE_ACTIONS,
        confirm: bool = False,
        name: str | None = None,
        provider_type: str | None = None,
        config_data: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Manage RClone cloud storage remotes.

        Actions:
          list_remotes - List all configured remotes
          config_form - Get config form schema (optional provider_type for specific provider)
          create_remote - Create a new remote (requires name, provider_type, config_data)
          delete_remote - Delete a remote (requires name, 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.")

        try:
            logger.info(f"Executing unraid_rclone action={action}")

            if action == "list_remotes":
                data = await make_graphql_request(QUERIES["list_remotes"])
                remotes = data.get("rclone", {}).get("remotes", [])
                return {"remotes": list(remotes) if isinstance(remotes, list) else []}

            if action == "config_form":
                variables: dict[str, Any] = {}
                if provider_type:
                    variables["formOptions"] = {"providerType": provider_type}
                data = await make_graphql_request(QUERIES["config_form"], variables or None)
                form = data.get("rclone", {}).get("configForm", {})
                if not form:
                    raise ToolError("No RClone config form data received")
                return dict(form)

            if action == "create_remote":
                if name is None or provider_type is None or config_data is None:
                    raise ToolError("create_remote requires name, provider_type, and config_data")
                data = await make_graphql_request(
                    MUTATIONS["create_remote"],
                    {"input": {"name": name, "type": provider_type, "config": config_data}},
                )
                remote = data.get("rclone", {}).get("createRCloneRemote")
                if not remote:
                    raise ToolError(
                        f"Failed to create remote '{name}': no confirmation from server"
                    )
                return {
                    "success": True,
                    "message": f"Remote '{name}' created successfully",
                    "remote": remote,
                }

            if action == "delete_remote":
                if not name:
                    raise ToolError("name is required for 'delete_remote' action")
                data = await make_graphql_request(
                    MUTATIONS["delete_remote"], {"input": {"name": name}}
                )
                success = data.get("rclone", {}).get("deleteRCloneRemote", False)
                if not success:
                    raise ToolError(f"Failed to delete remote '{name}'")
                return {
                    "success": True,
                    "message": f"Remote '{name}' deleted successfully",
                }

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

        except ToolError:
            raise
        except Exception as e:
            logger.error(f"Error in unraid_rclone action={action}: {e}", exc_info=True)
            raise ToolError(f"Failed to execute rclone/{action}: {e!s}") from e

    logger.info("RClone tool registered successfully")