"""Notification management. Provides the `unraid_notifications` tool with 9 actions for viewing, creating, archiving, and deleting system notifications. """ from typing import Any, Literal, get_args from fastmcp import Context as _Context from fastmcp import FastMCP from ..config.logging import logger from ..core.client import make_graphql_request from ..core.exceptions import CredentialsNotConfiguredError as _CredErr from ..core.exceptions import ToolError, tool_error_handler from ..core.setup import elicit_and_configure as _elicit # Re-export at module scope so tests can patch "unraid_mcp.tools.notifications.elicit_and_configure" # and "unraid_mcp.tools.notifications.CredentialsNotConfiguredError" elicit_and_configure = _elicit CredentialsNotConfiguredError = _CredErr Context = _Context QUERIES: dict[str, str] = { "overview": """ query GetNotificationsOverview { notifications { overview { unread { info warning alert total } archive { info warning alert total } } } } """, "list": """ query ListNotifications($filter: NotificationFilter!) { notifications { list(filter: $filter) { id title subject description importance link type timestamp formattedTimestamp } } } """, "warnings": """ query GetWarningsAndAlerts { notifications { warningsAndAlerts { id title subject description importance type timestamp } } } """, } MUTATIONS: dict[str, str] = { "create": """ mutation CreateNotification($input: NotificationData!) { createNotification(input: $input) { id title importance } } """, "archive": """ mutation ArchiveNotification($id: PrefixedID!) { archiveNotification(id: $id) { id title importance } } """, "unread": """ mutation UnreadNotification($id: PrefixedID!) { unreadNotification(id: $id) { id title importance } } """, "delete": """ mutation DeleteNotification($id: PrefixedID!, $type: NotificationType!) { deleteNotification(id: $id, type: $type) { unread { info warning alert total } archive { info warning alert total } } } """, "delete_archived": """ mutation DeleteArchivedNotifications { deleteArchivedNotifications { unread { info warning alert total } archive { info warning alert total } } } """, "archive_all": """ mutation ArchiveAllNotifications($importance: NotificationImportance) { archiveAll(importance: $importance) { unread { info warning alert total } archive { info warning alert total } } } """, "archive_many": """ mutation ArchiveNotifications($ids: [PrefixedID!]!) { archiveNotifications(ids: $ids) { unread { info warning alert total } archive { info warning alert total } } } """, "create_unique": """ mutation NotifyIfUnique($input: NotificationData!) { notifyIfUnique(input: $input) { id title importance } } """, "unarchive_many": """ mutation UnarchiveNotifications($ids: [PrefixedID!]!) { unarchiveNotifications(ids: $ids) { unread { info warning alert total } archive { info warning alert total } } } """, "unarchive_all": """ mutation UnarchiveAll($importance: NotificationImportance) { unarchiveAll(importance: $importance) { unread { info warning alert total } archive { info warning alert total } } } """, "recalculate": """ mutation RecalculateOverview { recalculateOverview { unread { info warning alert total } archive { info warning alert total } } } """, } DESTRUCTIVE_ACTIONS = {"delete", "delete_archived"} ALL_ACTIONS = set(QUERIES) | set(MUTATIONS) _VALID_IMPORTANCE = {"ALERT", "WARNING", "INFO"} NOTIFICATION_ACTIONS = Literal[ "overview", "list", "warnings", "create", "archive", "unread", "delete", "delete_archived", "archive_all", "archive_many", "create_unique", "unarchive_many", "unarchive_all", "recalculate", ] if set(get_args(NOTIFICATION_ACTIONS)) != ALL_ACTIONS: _missing = ALL_ACTIONS - set(get_args(NOTIFICATION_ACTIONS)) _extra = set(get_args(NOTIFICATION_ACTIONS)) - ALL_ACTIONS raise RuntimeError( f"NOTIFICATION_ACTIONS and ALL_ACTIONS are out of sync. " f"Missing from Literal: {_missing or 'none'}. Extra in Literal: {_extra or 'none'}" ) def register_notifications_tool(mcp: FastMCP) -> None: """Register the unraid_notifications tool with the FastMCP instance.""" @mcp.tool() async def unraid_notifications( action: NOTIFICATION_ACTIONS, confirm: bool = False, notification_id: str | None = None, notification_ids: list[str] | None = None, notification_type: str | None = None, importance: str | None = None, offset: int = 0, limit: int = 20, list_type: str = "UNREAD", title: str | None = None, subject: str | None = None, description: str | None = None, ctx: Context | None = None, ) -> dict[str, Any]: """Manage Unraid system notifications. Actions: overview - Notification counts by severity (unread/archive) list - List notifications with filtering (list_type=UNREAD/ARCHIVE, importance=INFO/WARNING/ALERT) warnings - Get deduplicated unread warnings and alerts create - Create notification (requires title, subject, description, importance) archive - Archive a notification (requires notification_id) unread - Mark notification as unread (requires notification_id) delete - Delete a notification (requires notification_id, notification_type, confirm=True) delete_archived - Delete all archived notifications (requires confirm=True) archive_all - Archive all notifications (optional importance filter) archive_many - Archive multiple notifications by ID (requires notification_ids) create_unique - Create notification only if no equivalent unread exists (requires title, subject, description, importance) unarchive_many - Move notifications back to unread (requires notification_ids) unarchive_all - Move all archived notifications to unread (optional importance filter) recalculate - Recompute overview counts from disk """ 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.") # Validate enum parameters before dispatching to GraphQL (SEC-M04). # Invalid values waste a rate-limited request and may leak schema details in errors. valid_list_types = frozenset({"UNREAD", "ARCHIVE"}) valid_importance = frozenset({"INFO", "WARNING", "ALERT"}) valid_notif_types = frozenset({"UNREAD", "ARCHIVE"}) if list_type.upper() not in valid_list_types: raise ToolError( f"Invalid list_type '{list_type}'. Must be one of: {sorted(valid_list_types)}" ) if importance is not None and importance.upper() not in valid_importance: raise ToolError( f"Invalid importance '{importance}'. Must be one of: {sorted(valid_importance)}" ) if notification_type is not None and notification_type.upper() not in valid_notif_types: raise ToolError( f"Invalid notification_type '{notification_type}'. " f"Must be one of: {sorted(valid_notif_types)}" ) with tool_error_handler("notifications", action, logger): logger.info(f"Executing unraid_notifications action={action}") if action == "overview": try: data = await make_graphql_request(QUERIES["overview"]) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(QUERIES["overview"]) notifications = data.get("notifications") or {} return dict(notifications.get("overview") or {}) if action == "list": filter_vars: dict[str, Any] = { "type": list_type.upper(), "offset": offset, "limit": limit, } if importance: filter_vars["importance"] = importance.upper() try: data = await make_graphql_request(QUERIES["list"], {"filter": filter_vars}) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(QUERIES["list"], {"filter": filter_vars}) notifications = data.get("notifications", {}) return {"notifications": notifications.get("list", [])} if action == "warnings": try: data = await make_graphql_request(QUERIES["warnings"]) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(QUERIES["warnings"]) notifications = data.get("notifications", {}) return {"warnings": notifications.get("warningsAndAlerts", [])} if action == "create": if title is None or subject is None or description is None or importance is None: raise ToolError("create requires title, subject, description, and importance") if importance.upper() not in _VALID_IMPORTANCE: raise ToolError( f"importance must be one of: {', '.join(sorted(_VALID_IMPORTANCE))}. " f"Got: '{importance}'" ) if len(title) > 200: raise ToolError(f"title must be at most 200 characters (got {len(title)})") if len(subject) > 500: raise ToolError(f"subject must be at most 500 characters (got {len(subject)})") if len(description) > 2000: raise ToolError( f"description must be at most 2000 characters (got {len(description)})" ) input_data = { "title": title, "subject": subject, "description": description, "importance": importance.upper(), } try: data = await make_graphql_request(MUTATIONS["create"], {"input": input_data}) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["create"], {"input": input_data}) notification = data.get("createNotification") if notification is None: raise ToolError("Notification creation failed: server returned no data") return {"success": True, "notification": notification} if action in ("archive", "unread"): if not notification_id: raise ToolError(f"notification_id is required for '{action}' action") try: data = await make_graphql_request(MUTATIONS[action], {"id": notification_id}) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS[action], {"id": notification_id}) return {"success": True, "action": action, "data": data} if action == "delete": if not notification_id or not notification_type: raise ToolError("delete requires notification_id and notification_type") _del_vars = {"id": notification_id, "type": notification_type.upper()} try: data = await make_graphql_request(MUTATIONS["delete"], _del_vars) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["delete"], _del_vars) return {"success": True, "action": "delete", "data": data} if action == "delete_archived": try: data = await make_graphql_request(MUTATIONS["delete_archived"]) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["delete_archived"]) return {"success": True, "action": "delete_archived", "data": data} if action == "archive_all": variables: dict[str, Any] | None = None if importance: variables = {"importance": importance.upper()} try: data = await make_graphql_request(MUTATIONS["archive_all"], variables) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["archive_all"], variables) return {"success": True, "action": "archive_all", "data": data} if action == "archive_many": if not notification_ids: raise ToolError("notification_ids is required for 'archive_many' action") try: data = await make_graphql_request( MUTATIONS["archive_many"], {"ids": notification_ids} ) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request( MUTATIONS["archive_many"], {"ids": notification_ids} ) return {"success": True, "action": "archive_many", "data": data} if action == "create_unique": if title is None or subject is None or description is None or importance is None: raise ToolError( "create_unique requires title, subject, description, and importance" ) if importance.upper() not in _VALID_IMPORTANCE: raise ToolError( f"importance must be one of: {', '.join(sorted(_VALID_IMPORTANCE))}. " f"Got: '{importance}'" ) if len(title) > 200: raise ToolError(f"title must be at most 200 characters (got {len(title)})") if len(subject) > 500: raise ToolError(f"subject must be at most 500 characters (got {len(subject)})") if len(description) > 2000: raise ToolError( f"description must be at most 2000 characters (got {len(description)})" ) input_data = { "title": title, "subject": subject, "description": description, "importance": importance.upper(), } try: data = await make_graphql_request( MUTATIONS["create_unique"], {"input": input_data} ) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request( MUTATIONS["create_unique"], {"input": input_data} ) notification = data.get("notifyIfUnique") if notification is None: return {"success": True, "duplicate": True, "data": None} return {"success": True, "duplicate": False, "data": notification} if action == "unarchive_many": if not notification_ids: raise ToolError("notification_ids is required for 'unarchive_many' action") try: data = await make_graphql_request( MUTATIONS["unarchive_many"], {"ids": notification_ids} ) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request( MUTATIONS["unarchive_many"], {"ids": notification_ids} ) return {"success": True, "action": "unarchive_many", "data": data} if action == "unarchive_all": vars_: dict[str, Any] | None = None if importance: vars_ = {"importance": importance.upper()} try: data = await make_graphql_request(MUTATIONS["unarchive_all"], vars_) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["unarchive_all"], vars_) return {"success": True, "action": "unarchive_all", "data": data} if action == "recalculate": try: data = await make_graphql_request(MUTATIONS["recalculate"]) except CredentialsNotConfiguredError: configured = await elicit_and_configure(ctx) if not configured: raise ToolError( "Credentials required. Run `unraid_health action=setup` to configure." ) data = await make_graphql_request(MUTATIONS["recalculate"]) return {"success": True, "action": "recalculate", "data": data} raise ToolError(f"Unhandled action '{action}' — this is a bug") logger.info("Notifications tool registered successfully")