mirror of
https://github.com/jmagar/unraid-mcp.git
synced 2026-03-23 04:29:17 -07:00
4b43c47091
settings.py: drop update_temperature, update_time, update_api, connect_sign_in, connect_sign_out, setup_remote_access, enable_dynamic_remote_access, update_ssh — all 8 reference mutations confirmed absent from Unraid API v4.29.2. Keep update + configure_ups. info.py: drop update_server (updateServerIdentity not in Mutation type) and update_ssh (duplicate of removed settings action). MUTATIONS is now empty; DESTRUCTIVE_ACTIONS is now an empty set. notifications.py: drop create_unique (notifyIfUnique not in Mutation type). Tests: remove corresponding test classes, add parametrized regression tests asserting removed actions are not in each tool's Literal type, update KNOWN_DESTRUCTIVE and _DESTRUCTIVE_TEST_CASES in safety audit, update schema coverage assertions. 858 tests passing, 0 failures.
302 lines
12 KiB
Python
302 lines
12 KiB
Python
"""Notification management.
|
|
|
|
Provides the `unraid_notifications` tool with 13 actions for viewing,
|
|
creating, archiving, and deleting system notifications.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
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
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
}
|
|
|
|
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 }
|
|
}
|
|
}
|
|
""",
|
|
"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",
|
|
"create",
|
|
"archive",
|
|
"unread",
|
|
"delete",
|
|
"delete_archived",
|
|
"archive_all",
|
|
"archive_many",
|
|
"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,
|
|
) -> 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)
|
|
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)
|
|
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":
|
|
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()
|
|
data = await make_graphql_request(QUERIES["list"], {"filter": filter_vars})
|
|
notifications = data.get("notifications", {})
|
|
return {"notifications": notifications.get("list", [])}
|
|
|
|
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(),
|
|
}
|
|
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")
|
|
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()}
|
|
data = await make_graphql_request(MUTATIONS["delete"], _del_vars)
|
|
return {"success": True, "action": "delete", "data": data}
|
|
|
|
if action == "delete_archived":
|
|
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()}
|
|
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")
|
|
data = await make_graphql_request(
|
|
MUTATIONS["archive_many"], {"ids": notification_ids}
|
|
)
|
|
return {"success": True, "action": "archive_many", "data": data}
|
|
|
|
if action == "unarchive_many":
|
|
if not notification_ids:
|
|
raise ToolError("notification_ids is required for 'unarchive_many' action")
|
|
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()}
|
|
data = await make_graphql_request(MUTATIONS["unarchive_all"], vars_)
|
|
return {"success": True, "action": "unarchive_all", "data": data}
|
|
|
|
if action == "recalculate":
|
|
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")
|