""" title: Reminder Scheduler Action author: open-webui author_url: https://github.com/hakki123mrk funding_url: https://github.com/hakki123mrk version: 0.1.0 required_open_webui_version: 0.3.9 """ TYPE = "action" TITLE = "Reminder Scheduler Action" DESCRIPTION = "Schedules a one-off reminder using APScheduler’s date trigger." from pydantic import BaseModel, Field from typing import Optional, Dict, Any import asyncio import atexit from apscheduler.schedulers.background import BackgroundScheduler from datetime import datetime # Create and start a global scheduler instance when this module is imported. scheduler: BackgroundScheduler = BackgroundScheduler() scheduler.start() atexit.register(lambda: scheduler.shutdown()) class Action: class Valves(BaseModel): # Optional configuration options can be defined here. pass def __init__(self): self.valves = self.Valves() def schedule_date(self, func, run_date: datetime, *args, **kwargs) -> str: """ Schedule a function to run once at a specific date/time. :param func: The function to execute. :param run_date: The datetime at which to run the function. :param args: Positional arguments for the function. :param kwargs: Keyword arguments for the function. :return: The unique identifier of the scheduled job. """ job = scheduler.add_job(func, 'date', run_date=run_date, args=args, kwargs=kwargs) return job.id async def run( self, input: Any, params: Dict[str, Any], __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[Dict[str, Any]]: """ Schedules a reminder to be executed once at a specified date/time. Expected parameters (via `params`): - reminder_message (str): The reminder text. - run_date (str): The date/time to trigger the reminder in ISO format (e.g., "2025-05-30T09:00:00"). :param input: Unused input. :param params: Dictionary containing scheduling parameters. :param __user__: (Optional) The user object. :param __event_emitter__: (Optional) A callable to emit events. :param __event_call__: (Optional) A callable to prompt for input. :return: A dictionary containing the scheduled job's ID or an error message. """ reminder_message: Optional[str] = params.get("reminder_message") run_date_str: Optional[str] = params.get("run_date") if not reminder_message or not run_date_str: return {"error": "Missing required parameters: reminder_message and run_date"} try: run_date_obj = datetime.fromisoformat(run_date_str) except Exception as e: return {"error": f"Invalid run_date format: {e}"} # Define the reminder function to be scheduled. def reminder() -> None: print(f"Reminder: {reminder_message}") # (Optional) Extend here to send notifications, emails, etc. # Schedule the reminder using the date trigger. job_id = self.schedule_date(reminder, run_date_obj) if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "Reminder scheduled", "job_id": job_id, "done": True}, } ) return {"job_id": job_id}