#!/usr/bin/env python3
"""
Looker Admin MCP - Scheduled plans management tools.
Provides scheduled plan CRUD and execution 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.schedules')
@handle_sdk_errors
async def list_scheduled_plans(
user_id: Optional[int] = None,
all_users: Optional[bool] = False
) -> Dict[str, Any]:
"""
List all scheduled plans, optionally filtered by user.
Args:
user_id: Filter by user ID (optional)
all_users: If True, return schedules for all users (requires admin)
"""
sdk = get_sdk()
logger.info(f"Listing scheduled plans (user_id={user_id}, all_users={all_users})")
plans = sdk.all_scheduled_plans(
user_id=user_id,
all_users=all_users
)
logger.info(f"Retrieved {len(plans)} scheduled plans")
return {
"scheduled_plans": [{
"id": p.id,
"name": p.name,
"title": p.title,
"user_id": p.user_id,
"dashboard_id": p.dashboard_id,
"look_id": p.look_id,
"crontab": p.crontab,
"timezone": p.timezone,
"enabled": p.enabled,
"next_run_at": p.next_run_at,
"last_run_at": p.last_run_at,
} for p in plans],
"count": len(plans),
"status": "success"
}
@handle_sdk_errors
async def get_scheduled_plans_for_dashboard(dashboard_id: int) -> Dict[str, Any]:
"""
Get all scheduled plans for a specific dashboard.
Args:
dashboard_id: The ID of the dashboard
"""
sdk = get_sdk()
logger.info(f"Getting scheduled plans for dashboard {dashboard_id}")
plans = sdk.scheduled_plans_for_dashboard(dashboard_id, all_users=True)
logger.info(f"Dashboard {dashboard_id} has {len(plans)} scheduled plans")
return {
"dashboard_id": dashboard_id,
"scheduled_plans": [{
"id": p.id,
"name": p.name,
"user_id": p.user_id,
"crontab": p.crontab,
"enabled": p.enabled,
} for p in plans],
"count": len(plans),
"status": "success"
}
@handle_sdk_errors
async def get_scheduled_plans_for_look(look_id: int) -> Dict[str, Any]:
"""
Get all scheduled plans for a specific Look.
Args:
look_id: The ID of the Look
"""
sdk = get_sdk()
logger.info(f"Getting scheduled plans for look {look_id}")
plans = sdk.scheduled_plans_for_look(look_id, all_users=True)
logger.info(f"Look {look_id} has {len(plans)} scheduled plans")
return {
"look_id": look_id,
"scheduled_plans": [{
"id": p.id,
"name": p.name,
"user_id": p.user_id,
"crontab": p.crontab,
"enabled": p.enabled,
} for p in plans],
"count": len(plans),
"status": "success"
}
@handle_sdk_errors
async def get_scheduled_plan(scheduled_plan_id: int) -> Dict[str, Any]:
"""
Get details of a specific scheduled plan.
Args:
scheduled_plan_id: The ID of the scheduled plan
"""
sdk = get_sdk()
logger.info(f"Getting scheduled plan {scheduled_plan_id}")
plan = sdk.scheduled_plan(scheduled_plan_id)
logger.info(f"Retrieved scheduled plan: {plan.name}")
return {
"id": plan.id,
"name": plan.name,
"title": plan.title,
"user_id": plan.user_id,
"dashboard_id": plan.dashboard_id,
"look_id": plan.look_id,
"crontab": plan.crontab,
"timezone": plan.timezone,
"enabled": plan.enabled,
"run_as_recipient": plan.run_as_recipient,
"include_links": plan.include_links,
"scheduled_plan_destinations": [{
"id": d.id,
"type": d.type,
"format": d.format,
"address": d.address,
} for d in (plan.scheduled_plan_destination or [])],
"created_at": plan.created_at,
"updated_at": plan.updated_at,
"next_run_at": plan.next_run_at,
"last_run_at": plan.last_run_at,
"status": "success"
}
@handle_sdk_errors
async def create_scheduled_plan(
name: str,
crontab: str,
dashboard_id: Optional[int] = None,
look_id: Optional[int] = None,
timezone: str = "America/Los_Angeles",
email_addresses: Optional[str] = None,
enabled: bool = True
) -> Dict[str, Any]:
"""
Create a new scheduled plan for a dashboard or look.
Args:
name: Name of the scheduled plan
crontab: Cron expression for schedule (e.g., '0 9 * * *' for 9am daily)
dashboard_id: Dashboard ID (required if look_id not provided)
look_id: Look ID (required if dashboard_id not provided)
timezone: Timezone for the schedule (default: America/Los_Angeles)
email_addresses: Comma-separated list of email addresses
enabled: Whether the schedule is enabled (default: True)
"""
if not dashboard_id and not look_id:
return {"error": "Either dashboard_id or look_id must be provided"}
sdk = get_sdk()
logger.info(f"Creating scheduled plan: {name}")
# Build destinations
destinations = []
if email_addresses:
for email in email_addresses.split(','):
destinations.append(models.ScheduledPlanDestination(
type="email",
address=email.strip(),
format="inline_visualizations"
))
plan_body = models.WriteScheduledPlan(
name=name,
dashboard_id=dashboard_id,
look_id=look_id,
crontab=crontab,
timezone=timezone,
enabled=enabled,
scheduled_plan_destination=destinations if destinations else None
)
created_plan = sdk.create_scheduled_plan(body=plan_body)
logger.info(f"Created scheduled plan {created_plan.id}: {name}")
return {
"id": created_plan.id,
"name": created_plan.name,
"crontab": created_plan.crontab,
"timezone": created_plan.timezone,
"enabled": created_plan.enabled,
"dashboard_id": created_plan.dashboard_id,
"look_id": created_plan.look_id,
"status": "success",
"message": f"Scheduled plan '{name}' created successfully"
}
@handle_sdk_errors
async def update_scheduled_plan(
scheduled_plan_id: int,
name: Optional[str] = None,
crontab: Optional[str] = None,
timezone: Optional[str] = None,
enabled: Optional[bool] = None
) -> Dict[str, Any]:
"""
Update an existing scheduled plan.
Args:
scheduled_plan_id: The ID of the scheduled plan to update
name: New name (optional)
crontab: New cron expression (optional)
timezone: New timezone (optional)
enabled: Enable/disable the schedule (optional)
"""
sdk = get_sdk()
logger.info(f"Updating scheduled plan {scheduled_plan_id}")
update_fields = {}
if name is not None:
update_fields['name'] = name
if crontab is not None:
update_fields['crontab'] = crontab
if timezone is not None:
update_fields['timezone'] = timezone
if enabled is not None:
update_fields['enabled'] = enabled
plan_body = models.WriteScheduledPlan(**update_fields)
updated_plan = sdk.update_scheduled_plan(scheduled_plan_id, body=plan_body)
logger.info(f"Updated scheduled plan {scheduled_plan_id}")
return {
"id": updated_plan.id,
"name": updated_plan.name,
"crontab": updated_plan.crontab,
"enabled": updated_plan.enabled,
"status": "success",
"message": f"Scheduled plan {scheduled_plan_id} updated successfully"
}
@handle_sdk_errors
async def delete_scheduled_plan(scheduled_plan_id: int, confirm: bool = False) -> Dict[str, Any]:
"""
Delete a scheduled plan. Requires confirm=True to execute.
Args:
scheduled_plan_id: The ID of the scheduled plan 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 scheduled plan {scheduled_plan_id}. Set confirm=True to proceed.",
"scheduled_plan_id": scheduled_plan_id
}
sdk = get_sdk()
logger.warning(f"Deleting scheduled plan {scheduled_plan_id} (confirmed)")
try:
plan = sdk.scheduled_plan(scheduled_plan_id)
plan_name = plan.name
except Exception:
plan_name = f"plan_{scheduled_plan_id}"
sdk.delete_scheduled_plan(scheduled_plan_id)
logger.warning(f"Deleted scheduled plan {scheduled_plan_id}")
return {
"status": "success",
"deleted_plan_id": scheduled_plan_id,
"deleted_plan_name": plan_name,
"message": f"Scheduled plan {scheduled_plan_id} has been permanently deleted"
}
@handle_sdk_errors
async def run_scheduled_plan_once(scheduled_plan_id: int) -> Dict[str, Any]:
"""
Run a scheduled plan immediately (one-time execution).
Args:
scheduled_plan_id: The ID of the scheduled plan to run
"""
sdk = get_sdk()
logger.info(f"Running scheduled plan {scheduled_plan_id} once")
result = sdk.scheduled_plan_run_once_by_id(scheduled_plan_id)
logger.info(f"Triggered one-time run for scheduled plan {scheduled_plan_id}")
return {
"scheduled_plan_id": scheduled_plan_id,
"status": "success",
"message": f"Scheduled plan {scheduled_plan_id} has been triggered for immediate execution"
}
