废话开头

AI诞生以来极大的改善了我们的日常生活和工作。作为AI的重度用户我最大的痛点就是AI知识库的更新麻烦以及没有日程提醒。前者通过openwebui的记忆功能及记忆函数解决了。偶然看到# xxtui - 个人消息推送这位佬友的推送API。正好替换了之前被微信封杀掉的其他API，做成了typecho博客的推送插件。然后突然想到，为什么不能让AI获取到日程安排后也推送呢？想到了马上就开始行动，定时推送，我们首先需要一个数据库存储推送的时间和内容，然后需要一个脚本来定时的查看数据库，并在存储的时间进行推送~我立马想到cloudflare大善人 ！D1数据库+worker 完美！ 另外感谢@Throttle @pluto233 两位大佬的帮助和提醒！不然我还一直折腾函数！哈哈

目前情况

如图所示，对于需要准备或行程安排的事件，工具可以根据默认值（例如 30 分钟）或指定的分钟数，自动为发送提前提醒。而对于“N分钟后”这类短时且即时的提醒，它能取消提前量，做到精确提醒~

准备工作

一个opwebui
@Cook_Sleep 佬友的时间感知函数 （必备）
@19850213313 佬友的记忆与函数 （可选，但推荐）
@wqmh 佬友的消息推送API （必备）
cloudflare的账号一个（必备）

创建Cloudflare D1数据库

登录Cloudflare主页后左下角选择存储和数据，创建D1数据库，命名scheduled-posts-db

创建Cloudflare的API

Windows终端无法唤起浏览器进行cloudflare的认证，所以创建API。设置变量来部署worker

点击Cloudflare右上角的人头选择配置文件
选择左侧API令牌
点击右侧创建令牌
选择Workers AI模版创建
确定和下图所有选项一致

最下方是设置有效期一天就行了，用完了可以回来删除。
创建完后复制好令牌打开powershell设置为环境变量

$env:CLOUDFLARE_API_TOKEN="你的令牌"

部署worker项目

提前创建好一个文件夹，在powershell中CD到这个文件夹输入
npm create cloudflare@latest
然后它会你一串问题，请根据我的截图选择

创建好后当前文件夹会产生一个子文件夹 如下图

打开wrangler.jsonc 修改

// wrangler.jsonc

{

  "name": "opwebui", // 项目名称

  "main": "src/index.ts",

  "compatibility_date": "2025-08-25", // 当前日期

  

  "d1_databases": [

    {

      "binding": "DB", // Worker 

      "database_name": "scheduled-posts-db", // D1数据库名称 

      "database_id": "这里填D1数据库ID，在名称旁边" 

    }

  ],

  

  "triggers": {

    // 配置 Cron Trigger (定时触发器)

    // "*/1 * * * *" 表示每1分钟执行一次，方便测试后面我改成9分钟了

    "crons": ["*/1 * * * *"]

  },

  

  // 如果需要其他配置，可以在这里添加

  "vars": {

  }

}

打开src目录下的index.ts 替换成下面的内容

    


interface Env {

  DB: D1Database;

  XXTUI_API_KEY: string;

}

  

// xxtui.com API 配置

const XXTUI_BASE_URL = "https://www.xxtui.com/xxtui/";

  



async function initDb(db: D1Database): Promise<void> {

  await db.exec(`CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, post_payload TEXT NOT NULL, schedule_time TEXT NOT NULL, sent_time TEXT, status TEXT DEFAULT 'pending');`);

  console.log("D1 数据库表 'tasks' 已初始化或已存在。");

}

  

// 添加新任务到数据库 (已移除 channel 参数)

async function addTask(

  db: D1Database,

  content: string,

  scheduleTime: Date

): Promise<void> {

  await initDb(db); 

  

  const payload = {

    content: content,

    from: "AI秘书", // 来源名称

    title: "日程提醒", // 固定标题

  };

  

  await db.prepare(

    "INSERT INTO tasks (post_payload, schedule_time) VALUES (?, ?)"

  )

  .bind(JSON.stringify(payload), scheduleTime.toISOString())

  .run();

  console.log(`任务已添加: '${content.substring(0, 20)}...' @ ${scheduleTime.toISOString()}`);

}

  

// 获取所有已到期且待发送的任务

async function getDueTasks(db: D1Database): Promise<any[]> {

  const now = new Date();

  const { results } = await db.prepare(

    "SELECT id, post_payload, schedule_time FROM tasks WHERE status = 'pending' AND schedule_time <= ?"

  )

  .bind(now.toISOString())

  .all();

  return results || [];

}

  

// 更新任务的状态 (发送成功、失败等)

async function updateTaskStatus(

  db: D1Database,

  taskId: number,

  status: 'pending' | 'sent' | 'failed',

  sentTime: Date | null = null

): Promise<void> {

  if (sentTime) {

    await db.prepare(

      "UPDATE tasks SET status = ?, sent_time = ? WHERE id = ?"

    )

    .bind(status, sentTime.toISOString(), taskId)

    .run();

  } else {

    await db.prepare(

      "UPDATE tasks SET status = ? WHERE id = ?"

    )

    .bind(status, taskId)

    .run();

  }

}

  

// --- 消息发送逻辑 ---

  

// 通过xxtui.com发送消息的实际函数

async function sendXxtuiMessage(apiKey: string, payload: any): Promise<boolean> {

  if (!apiKey) {

    console.error("XXTUI_API_KEY 未设置，无法发送消息。");

    return false;

  }

  

  const url = `${XXTUI_BASE_URL}${apiKey}`; // API Key 在 URL 路径中

  const headers = { "Content-Type": "application/json" }; // 参数在 JSON body 中

  

  try {

    const response = await fetch(url, {

      method: "POST",

      headers: headers,

      body: JSON.stringify(payload),

    });

  

    if (!response.ok) {

      throw new Error(`HTTP error! status: ${response.status}, message: ${await response.text()}`);

    }

  

    console.log(`消息发送成功 (${response.status}): ${response.statusText} - ${await response.text()}`);

    return true;

  } catch (e: any) {

    console.error(`消息发送失败: ${e.message}`);

    return false;

  }

}

  

// --- Cloudflare Worker 入口点 ---

  

export default {

  // fetch 处理外部 HTTP 请求 (例如从 OpenWebUI 后端添加任务)

  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {

    const url = new URL(request.url);

  

    // 监听 /add-task 路径的 POST 请求，用于添加提醒任务

    if (url.pathname === '/add-task' && request.method === 'POST') {

      try {

        const { content, scheduleTimeStr } = await request.json(); // 关键修正: 移除 channel 参数

        // 验证必要参数

        if (!content || !scheduleTimeStr) {

          return new Response("Missing content or scheduleTimeStr in request body.", { status: 400 });

        }

        const scheduleTime = new Date(scheduleTimeStr);

        await addTask(env.DB, content, scheduleTime); // 关键修正: 移除 channel 参数

        return new Response("Task added successfully!", { status: 200 });

      } catch (e: any) {

        console.error("Error adding task from fetch request:", e);

        return new Response(`Error adding task: ${e.message}`, { status: 500 });

      }

    }

  

    return new Response("Hello! This is your simplified Scheduled Posts Worker. POST to /add-task to add reminders.", { status: 200 });

  },

  

  // scheduled 处理定时触发器 (Cron Trigger) 的请求

  async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext): Promise<void> {

    console.log("Cron Trigger 触发，开始检查并处理定时任务...");

    if (!env.XXTUI_API_KEY) {

        console.error("XXTUI_API_KEY 未设置，无法执行推送。");

        return;

    }

  

    try {

      await initDb(env.DB);

  

      const dueTasks = await getDueTasks(env.DB);

      if (dueTasks.length === 0) {

        console.log("没有到期或待发送的任务。");

        return;

      }

  

      console.log(`发现 ${dueTasks.length} 个到期任务。`);

  

      for (const task of dueTasks) {

        try {

          const payload = JSON.parse(task.post_payload);

          console.log(`正在处理任务 ID: ${task.id}, 计划时间: ${task.schedule_time}`);

  

          if (await sendXxtuiMessage(env.XXTUI_API_KEY, payload)) {

            await updateTaskStatus(env.DB, task.id, 'sent', new Date());

            console.log(`任务 ID: ${task.id} 发送成功并已标记为'sent'。`);

          } else {

            await updateTaskStatus(env.DB, task.id, 'failed');

            console.log(`任务 ID: ${task.id} 发送失败并已标记为'failed'。`);

          }

        } catch (e: any) {

          console.error(`处理任务 ID: ${task.id} 时发生错误: ${e.message}`);

          await updateTaskStatus(env.DB, task.id, 'failed');

        }

      }

    } catch (e: any) {

      console.error("Scheduled handler 遇到关键错误:", e);

    }

  },

};

修改后回到终端，在你创建的项目名称目录下（代码里是opwebui）运行

npx wrangler deploy

完成后你就能在cloudflare的worker上看到这个项目了。你可以打开worker日志方便调试。

设置好环境变量

在xxtui - 个人消息推送注册并且生成你想要的key 在worker的设置页面中添加变量 变量名称XXTUI_API_KEY
这里的key 你可以填微信的也可以填钉钉或者短信的，看你自己喜好 XXTUI也提供了聚合KEY

在openwebui中添加工具并设置

打开openwebui
选择-工作空间-工具-新建工具
复制以下代码 并保存

"""
title: Worker Task Scheduler
description: Schedule tasks via worker API with fixed timezone offset and optional early reminders.
requirements: httpx
version: 0.0.5 #
licence: MIT
"""

import logging
import traceback
from datetime import datetime, timedelta
from httpx import AsyncClient
from pydantic import BaseModel, Field
from typing import Optional

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class Tools:
    class Valves(BaseModel):
        worker_url: str = Field(
            default="https://opwebui.worker.workers.dev/add-task",
            description="worker api endpoint url",
        )
        timeout: int = Field(default=30, description="timeout for api request")
        default_offset_hours: int = Field(
            default=8,
            description="Default timezone offset in hours from UTC (e.g., 8 for UTC+8, -5 for UTC-5). Does NOT handle DST.",
        )
        # --- 新增：默认提前提醒分钟数 ---
        default_early_reminder_minutes: int = Field(
            default=30,  # 默认提前 30 分钟
            description="Default minutes to send an early reminder before the main event. Set to 0 to disable by default.",
        )
        # -------------------------------

    def __init__(self):
        self.valves = self.Valves()

    async def add_worker_task(
        self,
        content: str,
        local_schedule_time_str: str,
        __event_emitter__: callable,
        user_offset_hours: Optional[int] = None,
        user_early_reminder_minutes: Optional[int] = None,
        # -------------------------------------
        **kwargs,
    ) -> str:
        """
        add a new task to worker scheduler
        :param content: task content to be scheduled
        :param local_schedule_time_str: scheduled time in local format (e.g., "2025-08-25T19:00:00")
        :param __event_emitter__: callback function for status updates.
        :param user_offset_hours: optional timezone offset in hours from UTC (e.g., 8 for UTC+8, -5 for UTC-5). If not provided, default_offset_hours from Valves will be used. Does NOT handle DST.
        :param user_early_reminder_minutes: optional minutes to send an early reminder before the main event. If not provided, default_early_reminder_minutes from Valves will be used. Set to 0 to disable early reminder.
        :return: result message
        """
        user = kwargs.get("user", {})
        metadata = kwargs.get("metadata", {})

        logger.info(
            "[add_worker_task] User ID: %s, Chat ID: %s, Content: %s, Local Schedule Time: %s, User Offset Hours: %s, Early Reminder Minutes: %s",
            user.get("id"),
            metadata.get("chat_id"),
            content[:50] + "..." if len(content) > 50 else content,
            local_schedule_time_str,
            user_offset_hours,
            user_early_reminder_minutes,  # 记录新参数
        )
        await __event_emitter__(
            {
                "type": "status",
                "data": {
                    "description": "processing time and scheduling worker task(s)",
                    "done": False,
                    "hidden": False,
                },
            }
        )

        effective_offset_hours = (
            user_offset_hours
            if user_offset_hours is not None
            else self.valves.default_offset_hours
        )
        effective_early_reminder_minutes = (
            user_early_reminder_minutes
            if user_early_reminder_minutes is not None
            else self.valves.default_early_reminder_minutes
        )

        try:
            naive_local_dt = datetime.fromisoformat(local_schedule_time_str)
            if naive_local_dt.tzinfo is not None:
                error_msg = "Please provide the schedule time without timezone information (e.g., 'YYYY-MM-DDTHH:MM:SS')."
                await __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": error_msg,
                            "done": True,
                            "hidden": False,
                        },
                    }
                )
                return error_msg
        except ValueError:
            error_msg = f"Invalid local schedule time format: '{local_schedule_time_str}'. Please use ISO format like 'YYYY-MM-DDTHH:MM:SS' (e.g., '2025-08-25T19:00:00')."
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {"description": error_msg, "done": True, "hidden": False},
                }
            )
            return error_msg

        # 将本地时间转换为 UTC
        main_utc_dt = naive_local_dt - timedelta(hours=effective_offset_hours)
        main_schedule_time_str = main_utc_dt.isoformat(timespec="seconds") + "Z"

        client = AsyncClient(timeout=self.valves.timeout)
        all_results = []  # 用于收集所有调度请求的结果

        try:
            # --- 步骤 1: 调度提前提醒 ---
            if effective_early_reminder_minutes > 0:
                early_reminder_local_dt = naive_local_dt - timedelta(
                    minutes=effective_early_reminder_minutes
                )
                early_reminder_utc_dt = early_reminder_local_dt - timedelta(
                    hours=effective_offset_hours
                )
                early_reminder_schedule_time_str = (
                    early_reminder_utc_dt.isoformat(timespec="seconds") + "Z"
                )

                early_reminder_content = f"【提前提醒】您的日程：'{content}' 将在 {effective_early_reminder_minutes} 分钟后开始 (预计 {local_schedule_time_str})。请做好准备！"

                payload_early = {
                    "content": early_reminder_content,
                    "scheduleTimeStr": early_reminder_schedule_time_str,
                }

                await __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": f"scheduling early reminder for {early_reminder_schedule_time_str}",
                            "done": False,
                            "hidden": False,
                        },
                    }
                )
                response_early = await client.post(
                    url=self.valves.worker_url,
                    json=payload_early,
                    headers={"Content-Type": "application/json"},
                )
                response_early.raise_for_status()
                all_results.append(
                    f"Early reminder scheduled ({early_reminder_schedule_time_str}). Response: {response_early.text}"
                )

            # --- 步骤 2: 调度主事件提醒 ---
            payload_main = {
                "content": content,
                "scheduleTimeStr": main_schedule_time_str,
            }
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "description": f"scheduling main task for {main_schedule_time_str}",
                        "done": False,
                        "hidden": False,
                    },
                }
            )
            response_main = await client.post(
                url=self.valves.worker_url,
                json=payload_main,
                headers={"Content-Type": "application/json"},
            )
            response_main.raise_for_status()
            all_results.append(
                f"Main task scheduled ({main_schedule_time_str}). Response: {response_main.text}"
            )

            final_message = "Task(s) scheduled successfully. " + " | ".join(all_results)
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "description": final_message,
                        "done": True,
                        "hidden": False,
                    },
                }
            )
            return final_message

        except Exception as err:
            message = f"failed to schedule task(s): {err}"
            logger.error(f"{message}\n{traceback.format_exc()}")
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {"description": message, "done": True, "hidden": False},
                }
            )
            return message
        finally:
            await client.aclose()

url处填入的cloudflare worker的域名（建议使用自定义域名） 记得添加/add-task路径

第二和三个是超时和时区设置默认即。最后一个是日程提醒的提前时间，单位是分钟。比如你预约9点的看牙 AI会提前30分钟预先提醒一次

创建模型启用工具

同样在工作空间创建模型(模型调用工具的能力很重要！)
勾选上工具，及时间感知 增强记忆等文章开头要求安装的函数
提示词中加入以下这段：

1.  **默认时区：** 使用 'user_offset_hours=8' (北京时间 UTC+8) 作为默认时区，除非用户明确指定其他时区偏移。
2.  **提前提醒 (user_early_reminder_minutes)：**
    *   如果用户指定了“N分钟后”的提醒，且 N 小于 30 分钟（例如“10分钟后提醒我”），则将 'user_early_reminder_minutes' 参数**明确设置为 0**，表示不需要提前提醒，只发送一个准点提醒。
    *   如果用户明确要求“不需要提前提醒”或“提前量为0”，也请将 'user_early_reminder_minutes' 设置为 0。
    *   对于其他需要准备的、时间跨度较长的事件（例如会议、预约、出行），如果用户没有明确指定提前量，则使用工具的默认值（例如 30 分钟）进行提前提醒。
    *   如果用户明确指定了提前量（例如“提前1小时提醒我”），则使用用户指定的分钟数。

请严格按照以上规则进行参数填充。

完成！