Agent Zero 调度器任务列表 API 深度解析/api/scheduler_tasks_list 的契约、实现与任务序列化全链路【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero本文围绕 Agent Zero 的 api/scheduler_tasks_list.py.dox.md 文档及其实现 api/scheduler_tasks_list.py 展开完整梳理这个“列出调度器全部任务”的 HTTP 端点它的路由注册与鉴权/CSRF 契约、请求与响应结构、时区参数的作用机制以及背后的任务存储、reload 与序列化底层实现。读完后你可以独立调用该端点获取任务清单理解每个任务字段的来源并能定位前端调度器面板是如何消费这份数据的。端点定位与职责边界api/scheduler_tasks_list.py.dox.md 是该端点的“持久化笔记”DOX profile其明确声明的职责是职责Purposescheduler_tasks_list.py独占ownscheduler_tasks_list这个 API 端点负责处理“列出调度器任务”的请求由于api/目录是有意保持扁平flat结构的DOX 文件与实现文件必须保持同步。所有权Ownershipscheduler_tasks_list.py拥有运行时实现.dox.md拥有关于职责、契约、副作用与验证的持久化说明。实现类为SchedulerTasksList继承自ApiHandler核心方法签名为async process(self, input: Input, request: Request) - Output。运行时契约Runtime ContractsHTTP handler 必须继承 helpers/api.py 中的helpers.api.ApiHandlerWebSocket handler 则须继承helpers.ws.WsHandler当请求载荷、鉴权/CSRF 要求、响应结构、路由副作用或 WebSocket 事件契约发生变化时必须同步更新 DOX。该端点已观测到的副作用区域是“调度器状态scheduler state”。导入依赖helpers.api、helpers.localization、helpers.print_style、helpers.task_scheduler以及标准库traceback。关键调用点Key Concepts源码中实际调用的 helper 为scheduler.serialize_all_tasks、Localization.get().set_timezone、scheduler.reload、PrintStyle.error和traceback.format_exc——这也是本文后续逐条展开的五个关键点。路由分发与安全契约POST 鉴权 CSRF在 Agent Zero 中api/目录下的每个文件对应一个端点。helpers/api.py 中注册的通用分发规则是app.add_url_rule( /api/path:path, api_dispatch, _dispatch, methods[GET, POST, PUT, PATCH, DELETE], )请求/api/scheduler_tasks_list时_dispatch会先在内置目录api/scheduler_tasks_list.py中加载第一个ApiHandler子类插件端点则位于plugins/name/api/下见 helpers/api.py并按 handler 类上声明的安全开关依次包裹装饰器helpers/api.pycsrf_protect当requires_csrf()为真、requires_api_key、requires_auth、requires_loopback。SchedulerTasksList本身没有覆写任何安全开关因此完全采用 ApiHandler 的默认值类方法默认值对本端点的含义get_methods()[POST]仅接受 POST 请求其他方法返回 405requires_auth()True若服务端配置了登录凭据未登录请求会被重定向到登录页requires_csrf()跟随requires_auth()即True请求必须携带有效的X-CSRF-Token头或 CSRF cookie否则返回 403requires_api_key()False不强制X-API-KEYrequires_loopback()False不限制只能回环地址访问对比/api/scheduler_tick显式要求 loopback见 api/scheduler_tick.pySchedulerTasksList是纯查询型端点没有任何写操作副作用DOX 中的工作指引Work Guidance也要求“除非端点契约明确变更否则保留鉴权、CSRF、loopback 与 API key 检查”并在需要非 JSON 响应文件、重定向、特定状态码时使用helpers.api.Response返回。实现逐行解析端点全部实现见 api/scheduler_tasks_list.pyclass SchedulerTasksList(ApiHandler): async def process(self, input: Input, request: Request) - Output: List all tasks in the scheduler with their types try: # Get timezone from input (do not set if not provided, # we then rely on poll() to set it) if timezone : input.get(timezone, None): Localization.get().set_timezone(timezone) # Get task scheduler scheduler TaskScheduler.get() await scheduler.reload() # Use the schedulers convenience method for task serialization tasks_list scheduler.serialize_all_tasks() return {ok: True, tasks: tasks_list} except Exception as e: PrintStyle.error(fFailed to list tasks: {str(e)} {traceback.format_exc()}) return {ok: False, error: fFailed to list tasks: {str(e)} {traceback.format_exc()}, tasks: []}执行链路可以拆成四步时区协商从请求体读取可选的timezone字段。源码注释明确说明“未提供时不做设置此时依赖poll()机制来设置”——即 WebUI 的轮询请求会周期性带上用户时区从而保证服务端单例的时区状态与用户一致。Localization是一个全局单例set_timezone定义在 helpers/localization.py它影响后续所有“naive datetime 归一化”和 ISO 序列化结果。获取调度器单例TaskScheduler.get()返回进程内唯一的 TaskScheduler 实例惰性单例首次构造时通过SchedulerTaskList.get()载入任务存储。强制 reloadawait scheduler.reload()会重新从磁盘读取usr/scheduler/tasks.json目录常量SCHEDULER_FOLDER usr/scheduler见 helpers/task_scheduler.py把文件内容反序列化并整体替换内存中的任务列表SchedulerTaskList.reload。这一步是保证列表视图“所见即磁盘最新状态”的关键即使其他进程CLI、插件、tick 协程刚改写了任务文件本端点也不会返回过期数据。序列化返回scheduler.serialize_all_tasks()委托给模块级函数serialize_tasks(self.get_tasks())helpers/task_scheduler.py逐任务产出面向前端的字典。响应结构成功时返回 HTTP 200 与如下 JSON{ok: true, tasks: [ /* serialize_task() 产出的任务对象列表 */ ]}失败时不抛 HTTP 500而是同样返回 200 与{ok: false, error: Failed to list tasks: 异常信息 完整 traceback, tasks: []}同时通过PrintStyle.error把错误与traceback.format_exc()打印到服务端控制台。也就是说调用方应以 JSON 中的ok字段判断成败而不是仅看 HTTP 状态码。调用示例端点为 POST JSON最小请求体可以为空对象时区可选curl -X POST http://host:port/api/scheduler_tasks_list \ -H Content-Type: application/json \ -H X-CSRF-Token: csrf_token \ -d {timezone: Asia/Shanghai}若服务端启用了登录配置了用户口令还需要携带有效的会话 Cookietimezone为 IANA 时区名。注意normalize_schedule_timezone对任务调度时区还支持local、user、default、current、current_timezone等别名helpers/task_scheduler.py但请求体里这个timezone是直接传给Localization.set_timezone的属于用户显示时区协商二者不要混淆。任务序列化每个字段从哪里来tasks数组中每个元素由 serialize_task 生成字段可按三类理解通用字段所有任务类型共有字段来源说明uuidBaseTask.uuid任务唯一 ID由guids.generate_id()生成name/system_prompt/prompt任务定义任务名称、执行时的系统提示词、用户提示词stateTaskState枚举idle/running/disabled/error四态helpers/task_scheduler.pyattachmentslist[str]附件文件路径或 URLproject_name/project_color任务定义任务绑定的项目created_at/updated_at/last_run时间戳经serialize_datetime转换为用户时区后的 ISO 字符串next_runtask.get_next_run()下次运行时间见下last_result任务最近一次执行结果成功时为结果文本失败时为ERROR: ...context_id/dedicated_contextBaseTask.context_id任务复用的 Agent 上下文 IDdedicated_context表示任务拥有独占上下文helpers/task_scheduler.pyproject派生{name, color}嵌套对象方便前端直接渲染类型专属字段——这正是 DOX 文档标题所说“list all tasks ... with their types”的含义if isinstance(task, ScheduledTask): task_dict[type] scheduled task_dict[schedule] serialize_task_schedule(task.schedule) elif isinstance(task, AdHocTask): task_dict[type] adhoc task_dict[token] adhoc_task.token else: task_dict[type] planned task_dict[plan] serialize_task_plan(planned_task.plan)scheduledschedule为{minute, hour, day, month, weekday, timezone}六个 cron 字段TaskScheduleto_crontab()可拼成标准 crontab 表达式adhoc多一个 19 位随机数字字符串token用于一次性/外部触发场景plannedplan为{todo: [...], in_progress: dt|null, done: [...]}的待办时间线TaskPlantodo升序、done按最新在前排序。next_run的计算逻辑因类型而异ScheduledTask用py-crontab的CronTab.next()在任务自身时区下求下一个触发点再转回 UTChelpers/task_scheduler.pyPlannedTask直接取plan.todo[0]helpers/task_scheduler.pyAdHocTask继承BaseTask返回None因此序列化后next_run为null。时区处理的完整闭环DOX 文档列出的关键概念Localization.get().set_timezone之所以重要是因为序列化中的每个时间字段都经过 serialize_datetimedef serialize_datetime(dt): # Use the Localization singleton for timezone conversion and serialization return Localization.get().serialize_datetime(dt)即任务内部统一保存时区感知aware的 datetime输出前由Localization单例helpers/localization.py 的serialize_datetime转换到“当前用户时区”。这就解释了端点的两个设计请求体里的timezone让调用方通常是 WebUI显式声明显示时区使created_at、last_run、next_run、plan中所有时间按用户本地时间展示不提供时“依赖poll()来设置”——WebUI 轮询流会持续维持时区状态避免每次列表请求都强依赖参数。与之呼应的是任务级时区ScheduledTask.schedule.timezone记录的是调度触发时区序列化前会被normalize_schedule_timezone归一化未知时区名会回退到用户当前时区并打印告警helpers/task_scheduler.py。仓库中针对该时区机制的行为测试见 tests/test_task_scheduler_timezone.py——这也正是 DOX“Verification”一节所建议的路径该端点没有按名字直接对应的专属测试应选择最近的调度器行为测试或做一次聚焦的冒烟验证“Run endpoint-specific or API/WebSocket tests for changed behavior; smoke-test browser callers when no focused test exists”。存储模型与 reload 语义理解本端点离不开其背后的存储结构helpers/task_scheduler.py持久化文件为usr/scheduler/tasks.json运行期生成顶层模型SchedulerTaskList以type字段作为 Pydantic 判别式discriminator反序列化为三种任务类型之一SchedulerTaskList.get()是进程级单例文件不存在时创建空列表并落盘存在时整体model_validate_json载入此后每次get()都会触发一次reload()TaskScheduler.get()也是单例构造时挂接SchedulerTaskList单例与一个绿色PrintStyle打印器。因此scheduler_tasks_list端点中await scheduler.reload()这一步的意义在于列表端点是“读路径”但 Agent Zero 的任务文件同时被调度 tick、任务增删改等多个写路径触碰reload 保证读到的不是内存缓存的旧快照。整个读路径是只读的——它不更新任何任务状态、不落盘这与 DOX 中“Observed side-effect areas: scheduler state”的措辞一致这里指与调度器共享状态空间而非写入。与相邻端点的关系也由此清晰/api/scheduler_tickapi/scheduler_tick.py驱动scheduler.tick()执行到期任务TaskScheduler.tick且它要求 loopback、不要求鉴权与 CSRF定位是本机内部心跳而本端点是对外登录用户暴露的只读视图配套的还有 api/scheduler_task_create.py、api/scheduler_task_update.py、api/scheduler_task_delete.py、api/scheduler_task_run.py 四个写端点。前端消费方WebUI 调度器面板仓库中的真实调用方是 WebUI 的调度器组件 webui/components/modals/scheduler/scheduler-store.jsasync listTasks() { const result await callSchedulerEndpoint( /scheduler_tasks_list, { timezone: getUserTimezone() }, Failed to fetch tasks ); if (!result.ok) return { ok: false, error: result.error }; const rawTasks Array.isArray(result.data?.tasks) ? result.data.tasks : []; const normalized rawTasks .filter(ensureTaskValidity) .map((task) normalizeTaskFromBackend(task)) .filter(Boolean); return { ok: true, tasks: normalized }; }可以看到前端的实际行为与后端契约一一对应POST JSON、携带timezone: getUserTimezone()对应前文的时区协商、按result.data.tasks解析任务数组再经过ensureTaskValidity过滤与normalizeTaskFromBackend归一化后渲染。DOX 的 Work Guidance 中“payload 形状变化时须同步更新前端调用方、插件调用方与测试”的要求正是以这条链路为准。变更该端点时的验证清单综合 DOX 与源码若你要修改scheduler_tasks_list例如调整响应字段或时区语义可依据以下事实基线契约基线POST、鉴权 CSRF、非 loopback 限制、{ok, tasks}/{ok, error, tasks}两种响应形状——均在 api/scheduler_tasks_list.py 与 helpers/api.py 的默认值中固定数据基线任务字段集由 serialize_task 定义时区字段语义由 normalize_schedule_timezone 与Localization单例决定一致性检查SchedulerTaskList.save()内置了 adhoc 任务空 token 的检测与自动修复helpers/task_scheduler.pydeserialize_task同样会补生成空 tokenhelpers/task_scheduler.py——列表端点读到的 adhoc 任务token因此不应为 null验证方式优先跑 tests/test_task_scheduler_timezone.py 等调度器行为测试无专属端点测试时按 DOX 建议对浏览器调用方做冒烟验证同步义务请求载荷、鉴权/CSRF 要求、响应结构或副作用任一变化都须同步更新 api/scheduler_tasks_list.py.dox.md 与 api/scheduler_tasks_list.py两者被 DOX 明确约定为成对维护。小结/api/scheduler_tasks_list是 Agent Zero 任务调度体系中唯一的“只读全景”端点它以极小的实现体量时区协商 → 单例获取 → 强制 reload → 序列化换取了与磁盘状态严格一致的任务视图其安全契约完全继承ApiHandler默认POST 鉴权 CSRF其响应字段则由serialize_task按scheduled/adhoc/planned三态任务类型统一产出。理解了这条链路就同时理解了 WebUI 调度器面板的数据来源以及scheduler_tick、任务增删改端点与usr/scheduler/tasks.json存储之间的协作方式。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考