"""Pushover notification tool for MCP server."""
import os
import httpx
from ..constants import (
PUSHOVER_API_URL,
PUSHOVER_PRIORITY_MAX,
PUSHOVER_PRIORITY_MIN,
PUSHOVER_VALID_SOUNDS,
REQUEST_TIMEOUT,
)
from ..exceptions import PushoverError
async def send_notification(
message: str,
title: str | None = None,
url: str | None = None,
url_title: str | None = None,
priority: int = 0,
sound: str | None = None,
) -> dict[str, str | int]:
"""Send a push notification via Pushover.
Args:
message: The notification message body (required).
title: Title shown at top of notification. Defaults to app name.
url: URL to include (tappable in notification).
url_title: Label for the URL instead of showing raw URL.
priority: Priority level from -2 (silent) to 2 (emergency). Default 0.
sound: Notification sound name. See Pushover docs for available sounds.
Returns:
Dictionary with 'status' (1 for success) and 'request' (UUID string).
Raises:
ValueError: If priority is out of range or sound is invalid.
PushoverError: If Pushover API returns an error or credentials are missing.
"""
# Get credentials from environment
api_token = os.environ.get("PUSHOVER_API_TOKEN")
user_key = os.environ.get("PUSHOVER_USER_KEY")
if not api_token:
raise PushoverError("PUSHOVER_API_TOKEN environment variable is not set")
if not user_key:
raise PushoverError("PUSHOVER_USER_KEY environment variable is not set")
# Validate priority
if not PUSHOVER_PRIORITY_MIN <= priority <= PUSHOVER_PRIORITY_MAX:
raise ValueError(
f"Priority must be between {PUSHOVER_PRIORITY_MIN} and "
f"{PUSHOVER_PRIORITY_MAX}, got {priority}"
)
# Validate sound if provided
if sound is not None and sound not in PUSHOVER_VALID_SOUNDS:
raise ValueError(
f"Invalid sound '{sound}'. Valid sounds: {', '.join(PUSHOVER_VALID_SOUNDS)}"
)
# Build form data
data: dict[str, str | int] = {
"token": api_token,
"user": user_key,
"message": message,
}
if title is not None:
data["title"] = title
if url is not None:
data["url"] = url
if url_title is not None:
data["url_title"] = url_title
if priority != 0:
data["priority"] = priority
if sound is not None:
data["sound"] = sound
# Send request
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
response = await client.post(PUSHOVER_API_URL, data=data)
# Parse response
result = response.json()
if response.status_code != 200 or result.get("status") != 1:
errors = result.get("errors", [])
raise PushoverError(
"Pushover API request failed",
status_code=response.status_code,
errors=errors,
)
return {"status": result["status"], "request": result["request"]}
