Agent Zero 定时任务完全指南：精通 scheduled-tasks 技能与 scheduler 工具

Agent Zero 定时任务完全指南：精通 scheduled-tasks 技能与 scheduler 工具

【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero

导读

本文围绕 Agent Zero 内置的scheduled-tasks技能及其底层的scheduler工具展开，系统讲解 Agent Zero 中定时任务（scheduled）、计划任务（planned）与临时任务（adhoc）的创建、查询、修改、运行与删除全流程。读完本文，你将掌握 cron 风格schedule字段与 ISOplan字段的正确用法、IANA 时区处理机制、专用上下文（dedicated context）与安全规则，并能结合源码理解任务调度器（helpers/task_scheduler.py）的底层实现原理。

一、scheduled-tasks 技能是什么

scheduled-tasks是 Agent Zero 内置技能之一，其说明为：

Use for complex Agent Zero scheduler work, including creating, updating, deleting, running, waiting for, timezone-correcting, or auditing scheduled, planned, and adhoc tasks.

即：凡是涉及 Agent Zero 调度器的复杂工作——创建、更新、删除、运行、等待、时区校正、审计定时/计划/临时任务——都应加载该技能。技能包由两个文件组成：

• skills/scheduled-tasks/SKILL.md：技能主体，拥有调度器动作（action）描述、调度字段（schedule fields）指引、安全规则与示例的“所有权”（Ownership）；
• skills/scheduled-tasks/AGENTS.md：技能的维护型 DOX，规定技能的职责边界与本地契约（Local Contracts）。

同时，Agent Zero 的主提示词 prompts/agent.system.tool.scheduler.md 中明确约定：复杂调度任务应加载scheduled-tasks技能；而在技能加载之前，Agent 也需要遵循其核心规则（先查后建、不递归调度、专用上下文等）。

二、调度器核心工作流：先查后动

技能与系统提示词都强调同一条铁律：

Always inspect existing tasks before creating, updating, deleting, or running one.

在创建（create_*）、更新（update_task）、删除（delete_task）或运行（run_task）任何任务之前，必须先通过find_task_by_name或list_tasks检查已有任务。这一契约在 skills/scheduled-tasks/SKILL.md 与 prompts/agent.system.tool.scheduler.md 中均有明确表述。

其典型调用格式（技能中的示例）：

{ "tool_name": "scheduler", "tool_args": { "action": "find_task_by_name", "name": "daily backup" } }

三、scheduler 工具的动作清单

scheduler工具共支持 10 个动作，每个动作对应 tools/scheduler.py 中SchedulerTool类的一个异步方法（execute方法通过action参数分发到具体实现，见 tools/scheduler.py）。动作及参数如下：

动作	必填参数	可选参数	说明
list_tasks	—	state[]、type[]、next_run_within、next_run_after	按状态、类型、距下次运行时间（分钟）过滤列出任务
find_task_by_name	name	—	按名称模糊查找任务（名称不区分大小写的子串匹配）
show_task	uuid	—	查看单个任务详情
run_task	uuid	context	立即运行任务，可附加上下文
update_task	uuid	name、system_prompt、prompt、attachments[]、schedule、timezone、plan[]、state、dedicated_context	更新任务字段
delete_task	uuid	—	删除任务（破坏性操作，需先按 UUID 定位）
create_scheduled_task	name、system_prompt、prompt	attachments[]、schedule、timezone、dedicated_context	创建按 cron 周期重复的任务
create_adhoc_task	name、system_prompt、prompt	attachments[]、dedicated_context	创建一次性临时任务
create_planned_task	name、system_prompt、prompt	attachments[]、plan[]、dedicated_context	创建按计划时间列表触发的任务
wait_for_task	uuid	—	等待任务完成并返回其结果

如果传入了未知动作，execute会返回错误提示，并列出自上而下的全部受支持动作（tools/scheduler.py）。

3.1 三种任务类型的底层模型

从源码看，三种任务分别对应 helpers/task_scheduler.py 中继承自BaseTask的三个 Pydantic 模型（通过type字段做判别联合，见 helpers/task_scheduler.py）：

• ScheduledTask（type: "scheduled"）：携带TaskSchedule，按 cron 表达式周期触发；
• AdHocTask（type: "adhoc"）：一次性临时任务，额外带有token字段用于身份校验；
• PlannedTask（type: "planned"）：携带TaskPlan（todo/in_progress/done三个时间列表），按指定时刻逐个触发。

任务状态TaskState共有四种枚举值：idle、running、disabled、error（helpers/task_scheduler.py）。任务被创建后默认处于idle；运行中被置为running；运行失败置为error；用户可显式禁用为disabled。

四、Schedule 字段详解：cron 风格而非 ISO

技能明确规定：schedule使用 cron 风格的字段，严禁把 ISO 日期时间塞进schedule（Do not put ISO datetimes intoschedule）。schedule包含以下字段：

• minute（分钟）
• hour（小时）
• day（日）
• month（月）
• weekday（星期）
• timezone（时区）

这些字段在TaskSchedule模型中被定义为 5 个 cron 字段加一个时区字段，并通过to_crontab()拼装为标准 5 段 cron 表达式（helpers/task_scheduler.py）：

def to_crontab(self) -> str: return f"{self.minute} {self.hour} {self.day} {self.month} {self.weekday}"

计划任务的日期时间则放进plan字段，且应为 ISO 字符串，例如2026-05-09T18:25:00。_task_plan_from_input会逐个解析plan列表中的 ISO 时间戳，任何一个无法解析都会返回 "Invalid datetime: ..." 错误（tools/scheduler.py）。

4.1 时区规范

• 使用 IANA 时区名称，例如Europe/Rome；
• 省略timezone时使用当前用户时区；
• 任务创建时的时区解析逻辑位于 tools/scheduler.py：_normalize_timezone会先检查别名（local、user、default、current、current_timezone均映射到当前用户时区），再通过pytz.timezone()校验合法性，非法时区会抛出明确的ValueError提示信息；
• 底层normalize_schedule_timezone也实现了同样的别名归一化与非法时区回退（打印错误日志后回退到用户时区），见 helpers/task_scheduler.py。

测试用例 tests/test_task_scheduler_timezone.py 验证了时区行为：Europe/Rome时区下 5 月 10 日 09:30 的定时任务，其 UTC 下次运行时间为 07:30（test_scheduled_task_next_run_uses_schedule_timezone）；同时验证了"local"这类历史遗留时区别名会被归一化为用户时区（test_scheduled_task_normalizes_legacy_local_timezone）。

4.2 cron 表达式的防幻觉校验

_validate_task_schedule使用正则表达式对拼装出的 crontab 进行校验，注释明确指出“agent might hallucinate”（Agent 可能幻觉生成非法 cron），非法表达式会返回 "Invalid cron expression: ..."（tools/scheduler.py）。此外，_task_schedule_from_input同时支持把字符串形式的schedule（空格分隔的 5 段 cron）解析为字段字典（tools/scheduler.py），因此在工具层与 api/scheduler_task_create.py 的 API 层都兼容字符串与字典两种schedule格式。

五、实战示例：三种创建动作

5.1 单个未来提醒：create_planned_task

对于一次性未来提醒，技能推荐create_planned_task：

{ "action": "create_planned_task", "name": "drink water", "prompt": "Remind the user to drink water.", "plan": ["2026-05-11T09:15:00"], "dedicated_context": true }

plan是 ISO 日期时间数组，任务会在所列时刻逐一触发，每完成一个时间点就将其从todo移入done（见下文TaskPlan进度推进）。PlannedTask.check_schedule()通过plan.should_launch()判断是否有已到期的计划时刻（helpers/task_scheduler.py）。

5.2 周期重复任务：create_scheduled_task

对于周期性、cron 形态的重复任务，使用create_scheduled_task：

{ "action": "create_scheduled_task", "name": "weekday stretch", "prompt": "Remind the user to stretch.", "schedule": { "minute": "15", "hour": "9", "day": "*", "month": "*", "weekday": "1-5", "timezone": "Europe/Rome" }, "dedicated_context": true }

该示例的含义是：每个工作日（周一至周五，weekday: "1-5"）的 09:15（minute: "15"、hour: "9"）在罗马时区触发任务。

技能系统提示词中还给出了“明天罗马时间 9:15”的具体写法（prompts/agent.system.tool.scheduler.md）：

{ "schedule": { "minute": "15", "hour": "9", "day": "11", "month": "5", "weekday": "*", "timezone": "Europe/Rome" } }

5.3 一次性临时任务：create_adhoc_task

create_adhoc_task不携带schedule或plan，创建后由用户显式触发。从源码看，AdHocTask额外生成一个 19 位随机token（helpers/task_scheduler.py），SchedulerTaskList.save()甚至在写盘前专门校验 adhoc 任务的 token 非空，为空时自动补生成（helpers/task_scheduler.py）。

5.4 完整的 JSON 调用结构

技能中的完整示例展示了工具调用包裹结构：

{ "tool_name": "scheduler", "tool_args": { "action": "find_task_by_name", "name": "daily backup" } }

SchedulerTool.execute会先将action参数统一转为小写并把连字符替换为下划线（_current_action，tools/scheduler.py），再做动作分发。

六、安全规则（Safety）

技能 skills/scheduled-tasks/SKILL.md 明确列出四条安全红线：

1. 不要创建递归的任务提示词（Do not create recursive task prompts that schedule more tasks）：即任务自身的 prompt 不应再触发创建新的调度任务，防止任务无限自我繁衍；
2. 不要因为任务已到点就运行它，只有用户要求时才运行（Do not run a task just because it is scheduled; run only if the user asks）：Agent 不应主动抢跑；
3. 新创建的任务默认使用专用上下文，除非dedicated_context显式为false：从源码看，三个创建动作的dedicated_context默认值都是True，为True时context_id=None，之后BaseTask.__init__会把context_id回填为任务自身的uuid（is_dedicated()判断context_id == uuid），为False时则绑定到当前 Agent 的上下文（tools/scheduler.py、helpers/task_scheduler.py）；
4. 破坏性操作必须先用 UUID 定位任务再执行：delete_task、update_task都要求先通过find_task_by_name或list_tasks拿到目标任务的uuid，再以 UUID 操作。

此外，prompts/agent.system.tool.scheduler.md 还补充了一条规则：涉及计划/定时任务时，若用户点名了某个时区，必须把该时区写入timezone字段。

七、源码级原理：任务如何被触发、执行与持久化

7.1 触发机制：tick 轮询

任务到期检测由TaskScheduler.tick()完成——它从任务列表取出所有“到期且空闲”的任务并逐个运行（helpers/task_scheduler.py）：

async def tick(self): for task in await self._tasks.get_due_tasks(): await self._run_task(task)

get_due_tasks()内部先reload()最新状态，再筛选check_schedule()为真且state == IDLE的任务（helpers/task_scheduler.py）。ScheduledTask.check_schedule()用crontab库计算参考时刻到下次执行的秒数是否落在轮询频率（默认 60 秒）窗口内（helpers/task_scheduler.py）。

对外暴露的轮询入口是 API 端点 api/scheduler_tick.py，它支持传入timezone设置本地化、打印任务清单与统计，并返回序列化后的全部任务。

7.2 执行机制：后台 DeferredTask 线程

_run_task将任务的执行包装进DeferredTask线程（thread_name="TaskScheduler"）中异步运行，并登记到_running_deferred_tasks字典以便支持取消（cancel_running_task、cancel_tasks_by_context）（helpers/task_scheduler.py）。执行流程为：

1. 快照检查任务是否存在、是否已在运行（防重复触发）；
2. 原子地把状态置为running（update_task_checked通过验证函数防止竞态条件）；
3. 调用on_run()（PlannedTask会在此把即将执行的时刻从todo移到in_progress）；
4. 获取/创建任务上下文（_get_chat_context），组装带附件的用户消息（附件支持本地路径与 http/https 等 URL），注入system_prompt并调用agent.monologue()运行 Agent；
5. 成功后on_success将状态置回idle并记录last_result；失败后on_error将状态置为error并记录last_result = "ERROR: ..."；
6. on_finish统一刷新updated_at时间戳。

run_task_by_uuid还处理了特殊状态转换：已运行的任务再次运行会报错，禁用任务拒绝运行，error 状态任务会先重置回 idle 再运行（helpers/task_scheduler.py）。

7.3 计划任务的进度推进

TaskPlan维护三个时间列表：todo（待执行）、in_progress（进行中）、done（已完成）。add_todo插入并排序；set_in_progress把时刻从todo移到in_progress；set_done在任务结束后把in_progress移到done（helpers/task_scheduler.py）。PlannedTask.on_run/on_finish将这一推进过程与调度器生命周期挂钩，并强制持久化（helpers/task_scheduler.py）。这样，plan: ["2026-05-11T09:15:00", "2026-05-11T15:00:00"]这样的任务会在两个时刻各触发一次，第二次触发只会在第一次完成后发生。

7.4 持久化：usr/scheduler/tasks.json

所有任务通过SchedulerTaskList单例持久化到usr/scheduler/tasks.json（常量SCHEDULER_FOLDER = "usr/scheduler"，helpers/task_scheduler.py）。SchedulerTaskList.get()在首次访问时若文件不存在会自动创建空任务列表（helpers/task_scheduler.py）。每次增删改都会落盘（save()），每次读取前都会reload()重新校验最新状态；update_task_by_uuid通过“先 reload、再更新、再保存”并配合线程锁实现原子更新，避免多进程并发竞争（helpers/task_scheduler.py）。所有增删改操作后还会调用mark_dirty_all通知状态监控模块（state monitor）刷新界面。

7.5 任务序列化字段

serialize_task输出的标准任务结构包含：uuid、name、state、system_prompt、prompt、attachments、project_name、project_color、created_at、updated_at、last_run、next_run（由 cron/plan 计算）、last_result、context_id、dedicated_context，以及按类型附加的schedule（scheduled）/token（adhoc）/plan（planned）字段（helpers/task_scheduler.py）。list_tasks动作的过滤参数next_run_within、next_run_after正是基于序列化中的get_next_run_minutes()计算值实现的（tools/scheduler.py）。

7.6 WebUI 与 API 层

除 Agent 工具外，调度功能还通过 WebUI API 暴露：SchedulerTaskCreate（api/scheduler_task_create.py）负责 UI 端创建任务，自动区分schedule、plan、adhoc 三种分支，并为 adhoc 任务自动生成 token；SchedulerTaskUpdate（api/scheduler_task_update.py）支持按task_id更新字段、状态与调度，并明确禁止修改项目绑定（"Project changes are not allowed"）。整套 API 均复用helpers/task_scheduler.py的解析与序列化函数，保证工具层与 UI 层行为一致。

八、运行与等待：run_task 与 wait_for_task

• run_task：按 UUID 立即运行任务，可选context参数注入附加上下文（task_context会被拼入任务的 user 消息中）。值得注意的细节是：如果任务运行在当前上下文（非专用上下文），工具会返回break_loop=True以中断当前对话循环，避免同一窗口出现两个对话（tools/scheduler.py）；
• wait_for_task：阻塞等待任务在独立专用上下文中完成，默认超时 300 秒（DEFAULT_WAIT_TIMEOUT）。如果任务运行在调用者自己的上下文中，会直接返回错误 "You can only wait for tasks running in their own dedicated context."。等待期间每秒轮询一次任务状态，超时返回 "Task wait timeout (300 seconds)"（tools/scheduler.py）。等待结束时返回任务 UUID、状态、上次运行时间与结果内容。

九、删除与清理

delete_task是破坏性操作：删除前会检查任务状态，若任务正在运行（running）则先重置其上下文与状态为idle；若任务持有独立的专用上下文，还会同时移除AgentContext与对应的持久化聊天记录（persist_chat.remove_chat），最后从任务列表中移除并落盘（tools/scheduler.py）。这也是技能强调“破坏性操作先按 UUID 定位”的原因——删除不仅影响任务本身，还连带清理其上下文与历史。

十、技能维护约定

根据 skills/scheduled-tasks/AGENTS.md 的约定，该技能由 Agent Zero 开发者负责维护，核心职责包括：

• 保持调度器工具动作、字段名、时区行为的准确性；
• 技能示例必须始终是合法的 JSON 工具参数（Keep examples valid JSON tool arguments）；
• 当调度器工具动作、字段或时区行为变化时，同步更新技能；并手动复查 SKILL.md 中是否存在过期的动作名与日期指引。

技能目录结构验证：scheduled-tasks技能目录下共两个文件（AGENTS.md与SKILL.md），无子 DOX，与 AGENTS.md 中 "No child DOX files" 的声明一致，符合 skills/AGENTS.md 中“每个技能目录必须包含SKILL.md”的全局契约。

结语

scheduled-tasks技能与scheduler工具共同构成了 Agent Zero 的完整任务调度能力：cron 风格的schedule支撑周期任务，ISO 格式的plan支撑精确时刻的一次性提醒，adhoc支撑临时执行，wait_for_task支撑同步等待结果，而“先查后动”、非递归提示、专用上下文与 UUID 定位等安全规则则保证了调度行为的可控与可审计。理解 tools/scheduler.py 与 helpers/task_scheduler.py 的底层实现，可以帮助你在使用调度能力时准确预判其行为，并在扩展或排障时快速定位问题。

【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero