#!/usr/bin/env python3
"""
Looker Admin MCP - Alert management tools.
Provides alert CRUD and subscription operations.
"""
import logging
from typing import Dict, Any, Optional, List
from looker_sdk import models40 as models
from looker_admin_mcp.server.looker import handle_sdk_errors, get_sdk
logger = logging.getLogger('looker-admin-mcp.tools.alerts')
@handle_sdk_errors
async def search_alerts(
dashboard_element_id: Optional[int] = None,
user_id: Optional[int] = None,
disabled: Optional[bool] = None,
all_owners: Optional[bool] = False
) -> Dict[str, Any]:
"""
Search alerts with various filters.
Args:
dashboard_element_id: Filter by dashboard element ID
user_id: Filter by owner user ID
disabled: Filter by disabled status (True/False)
all_owners: If True, return alerts for all owners (requires admin)
"""
sdk = get_sdk()
logger.info(f"Searching alerts (disabled={disabled}, all_owners={all_owners})")
alerts = sdk.search_alerts(
dashboard_element_id=dashboard_element_id,
user_id=user_id,
disabled=disabled,
all_owners=all_owners
)
logger.info(f"Found {len(alerts)} alerts")
return {
"alerts": [{
"id": a.id,
"custom_title": a.custom_title,
"comparison_type": a.comparison_type,
"threshold": a.threshold,
"dashboard_element_id": a.dashboard_element_id,
"is_disabled": a.is_disabled,
"is_public": a.is_public,
"cron": a.cron,
"owner_id": a.owner_id,
} for a in alerts],
"count": len(alerts),
"status": "success"
}
@handle_sdk_errors
async def get_alert(alert_id: int) -> Dict[str, Any]:
"""
Get details of a specific alert.
Args:
alert_id: The ID of the alert
"""
sdk = get_sdk()
logger.info(f"Getting alert {alert_id}")
alert = sdk.get_alert(alert_id)
logger.info(f"Retrieved alert: {alert.custom_title or alert_id}")
return {
"id": alert.id,
"custom_title": alert.custom_title,
"description": alert.description,
"comparison_type": alert.comparison_type,
"threshold": alert.threshold,
"field": {
"title": alert.field.title if alert.field else None,
"name": alert.field.name if alert.field else None,
} if alert.field else None,
"dashboard_element_id": alert.dashboard_element_id,
"is_disabled": alert.is_disabled,
"is_public": alert.is_public,
"cron": alert.cron,
"owner_id": alert.owner_id,
"followed": alert.followed,
"followable": alert.followable,
"destinations": [{
"destination_type": d.destination_type,
"email_address": d.email_address,
} for d in (alert.destinations or [])],
"status": "success"
}
@handle_sdk_errors
async def create_alert(
comparison_type: str,
threshold: float,
dashboard_element_id: int,
cron: str = "0 * * * *",
custom_title: Optional[str] = None,
description: Optional[str] = None,
email_addresses: Optional[str] = None,
is_disabled: bool = False,
is_public: bool = False
) -> Dict[str, Any]:
"""
Create a new alert.
Args:
comparison_type: Comparison type (LESS_THAN, GREATER_THAN, EQUAL_TO)
threshold: Threshold value for the alert
dashboard_element_id: ID of the dashboard element to monitor
cron: Cron expression for check frequency (default: hourly)
custom_title: Custom title for the alert (optional)
description: Description of the alert (optional)
email_addresses: Comma-separated list of email addresses (optional)
is_disabled: Whether alert is disabled (default: False)
is_public: Whether alert is public (default: False)
"""
sdk = get_sdk()
logger.info(f"Creating alert for dashboard element {dashboard_element_id}")
# Build destinations
destinations = []
if email_addresses:
for email in email_addresses.split(','):
destinations.append(models.AlertDestination(
destination_type="email",
email_address=email.strip()
))
alert_body = models.WriteAlert(
comparison_type=comparison_type,
threshold=threshold,
dashboard_element_id=dashboard_element_id,
cron=cron,
custom_title=custom_title,
description=description,
is_disabled=is_disabled,
is_public=is_public,
destinations=destinations if destinations else None
)
created_alert = sdk.create_alert(body=alert_body)
logger.info(f"Created alert {created_alert.id}")
return {
"id": created_alert.id,
"custom_title": created_alert.custom_title,
"comparison_type": created_alert.comparison_type,
"threshold": created_alert.threshold,
"dashboard_element_id": created_alert.dashboard_element_id,
"cron": created_alert.cron,
"is_disabled": created_alert.is_disabled,
"status": "success",
"message": f"Alert created successfully"
}
@handle_sdk_errors
async def update_alert(
alert_id: int,
threshold: Optional[float] = None,
cron: Optional[str] = None,
is_disabled: Optional[bool] = None,
custom_title: Optional[str] = None,
description: Optional[str] = None
) -> Dict[str, Any]:
"""
Update an existing alert.
Args:
alert_id: The ID of the alert to update
threshold: New threshold value (optional)
cron: New cron expression (optional)
is_disabled: Enable/disable the alert (optional)
custom_title: New custom title (optional)
description: New description (optional)
"""
sdk = get_sdk()
logger.info(f"Updating alert {alert_id}")
# Get existing alert to preserve required fields
existing = sdk.get_alert(alert_id)
update_fields = {
'comparison_type': existing.comparison_type,
'threshold': threshold if threshold is not None else existing.threshold,
'dashboard_element_id': existing.dashboard_element_id,
}
if cron is not None:
update_fields['cron'] = cron
else:
update_fields['cron'] = existing.cron
if is_disabled is not None:
update_fields['is_disabled'] = is_disabled
if custom_title is not None:
update_fields['custom_title'] = custom_title
if description is not None:
update_fields['description'] = description
alert_body = models.WriteAlert(**update_fields)
updated_alert = sdk.update_alert(alert_id, body=alert_body)
logger.info(f"Updated alert {alert_id}")
return {
"id": updated_alert.id,
"custom_title": updated_alert.custom_title,
"threshold": updated_alert.threshold,
"cron": updated_alert.cron,
"is_disabled": updated_alert.is_disabled,
"status": "success",
"message": f"Alert {alert_id} updated successfully"
}
@handle_sdk_errors
async def delete_alert(alert_id: int, confirm: bool = False) -> Dict[str, Any]:
"""
Delete an alert. Requires confirm=True to execute.
Args:
alert_id: The ID of the alert to delete
confirm: Must be True to execute the deletion
"""
if not confirm:
return {
"error": "Destructive operation requires confirm=True",
"warning": f"This will permanently delete alert {alert_id}. Set confirm=True to proceed.",
"alert_id": alert_id
}
sdk = get_sdk()
logger.warning(f"Deleting alert {alert_id} (confirmed)")
try:
alert = sdk.get_alert(alert_id)
alert_title = alert.custom_title or f"alert_{alert_id}"
except Exception:
alert_title = f"alert_{alert_id}"
sdk.delete_alert(alert_id)
logger.warning(f"Deleted alert {alert_id}")
return {
"status": "success",
"deleted_alert_id": alert_id,
"deleted_alert_title": alert_title,
"message": f"Alert {alert_id} has been permanently deleted"
}
@handle_sdk_errors
async def follow_alert(alert_id: int) -> Dict[str, Any]:
"""
Follow an alert to receive notifications.
Args:
alert_id: The ID of the alert to follow
"""
sdk = get_sdk()
logger.info(f"Following alert {alert_id}")
sdk.follow_alert(alert_id)
logger.info(f"Now following alert {alert_id}")
return {
"alert_id": alert_id,
"status": "success",
"message": f"Now following alert {alert_id}"
}
@handle_sdk_errors
async def unfollow_alert(alert_id: int) -> Dict[str, Any]:
"""
Unfollow an alert to stop receiving notifications.
Args:
alert_id: The ID of the alert to unfollow
"""
sdk = get_sdk()
logger.info(f"Unfollowing alert {alert_id}")
sdk.unfollow_alert(alert_id)
logger.info(f"Unfollowed alert {alert_id}")
return {
"alert_id": alert_id,
"status": "success",
"message": f"Unfollowed alert {alert_id}"
}
