How to Manage Scheduling in Agent Zero: A Complete Guide to Task Automation
Agent Zero includes a built-in task scheduler that supports cron-style recurring tasks, ad-hoc one-off executions, and planned timestamp-based tasks, all manageable via Python API, CLI tools, or HTTP endpoints.
Agent Zero is an open-source AI agent framework that ships with a lightweight, file-backed task scheduler for automating background operations. Whether you need to run maintenance scripts on a schedule, trigger one-off agent workflows, or plan tasks for specific future timestamps, the scheduling system in Agent Zero provides thread-safe persistence and multiple interfaces for management.
Understanding the TaskScheduler Architecture
The scheduling system centers around the TaskScheduler class, implemented as a singleton in python/helpers/task_scheduler.py. This design ensures that only one scheduler instance manages the task queue across the entire application lifecycle.
To access the global instance, use the factory method:
from python.helpers.task_scheduler import TaskScheduler
scheduler = TaskScheduler.get()
The singleton holds two critical components:
- A
SchedulerTaskListthat manages persistence of tasks tousr/scheduler/tasks.json - A registry of currently running
DeferredTaskobjects tracking active executions
Persistence and Thread Safety
All tasks are stored as JSON in usr/scheduler/tasks.json. The SchedulerTaskList.get() method (lines 70-80 of task_scheduler.py) handles lazy initialization, creating the file if it does not exist.
Thread-safe updates are guaranteed through SchedulerTaskList.update_task_by_uuid() (lines 41-71), which acquires an internal lock before writing to disk. This prevents race conditions when multiple agents or API calls modify the schedule simultaneously.
Task Types and Models
Agent Zero supports three distinct task types, all inheriting from BaseTask in python/helpers/task_scheduler.py. Each type supports lifecycle hooks (on_run, on_success, on_error, on_finish) and common fields including uuid, name, state, system_prompt, and prompt.
ScheduledTask (Cron-Style Recurring)
ScheduledTask implements cron-style scheduling through a TaskSchedule object containing minute, hour, day, month, and weekday fields. When the scheduler tick matches the pattern, the task executes.
Source: ScheduledTask class definition, lines 88-104 of task_scheduler.py.
AdHocTask (One-Off Execution)
AdHocTask represents single-shot tasks triggered manually or by external events. It extends BaseTask with a unique token field for idempotency checks.
Source: AdHocTask class, lines 40-48 of task_scheduler.py.
PlannedTask (Timestamp-Based)
PlannedTask executes at specific UTC timestamps defined in a TaskPlan structure. It tracks state transitions through todo, in_progress, and done phases, making it ideal for deadline-driven workflows.
Source: PlannedTask class, lines 70-78 of task_scheduler.py.
The Scheduler Tick Loop
Task execution is driven by the TaskScheduler.tick() method, invoked periodically by the background job loop in python/helpers/job_loop.py (default interval: every minute).
The tick process follows this sequence:
-
Fetch due tasks via
SchedulerTaskList.get_due_tasks()(lines 84-90), which filters for tasks wherecheck_schedule()returnsTrueand state isIDLE. -
Execute via
_run_task()(lines 11-57), which:- Atomically marks the task
RUNNINGusingupdate_task_checked - Creates or reuses an
AgentContextfor the task - Streams the user-provided prompt to a fresh agent instance
- Persists results and transitions state back to
IDLE(orERRORon exception)
- Atomically marks the task
The background loop ensures that scheduled tasks run even when no user is actively interacting with the agent, making the system suitable for production automation.
Managing Tasks with SchedulerTool
For programmatic control, Agent Zero provides SchedulerTool in python/tools/scheduler.py, which exposes high-level methods that map to the core scheduler API.
Creating a Scheduled Task
from python.tools.scheduler import SchedulerTool
from agent import Agent
# Assume `agent` is a live Agent instance
scheduler = SchedulerTool(agent=Agent())
await scheduler.create_scheduled_task(
name="nightly-backup",
system_prompt="You are a backup script executor",
prompt="Run `tar -czf /backups/daily.tar.gz /mydata` and report success.",
schedule={"minute": "0", "hour": "2", "day": "*", "month": "*", "weekday": "*"},
dedicated_context=True,
)
This creates a cron-style task running daily at 02:00 UTC. The dedicated_context=True flag forces a fresh chat context for isolation.
Running Tasks Immediately
await scheduler.run_task(uuid="123e4567-e89b-12d3-a456-426614174000")
The tool verifies the task isn't already running, then invokes TaskScheduler.run_task_by_uuid() and returns a Response when the agent completes.
Waiting for Completion
await scheduler.wait_for_task(uuid="123e4567-e89b-12d3-a456-426614174000")
Blocks for up to 5 minutes (configurable) until the task state transitions from RUNNING, returning a formatted summary of results.
HTTP API Endpoints for Remote Scheduling
Agent Zero exposes the scheduler functionality via HTTP-style endpoints in python/api/, enabling remote management without direct Python imports.
| Endpoint | Method | Purpose | Implementation |
|---|---|---|---|
/scheduler/tasks |
GET | List all tasks | scheduler_tasks_list.py (lines 12-23) |
/scheduler/task/create |
POST | Create scheduled/ad-hoc/planned task | scheduler_task_create.py (lines 15-46) |
/scheduler/task/run |
POST | Execute task by UUID/name | scheduler_task_run.py (lines 13-35) |
/scheduler/task/update |
PATCH | Modify fields (cron, state, plan) | scheduler_task_update.py (lines 12-30) |
/scheduler/task/delete |
DELETE | Remove task | scheduler_task_delete.py (lines 11-28) |
/scheduler/tick |
POST | Manually trigger scheduler tick | scheduler_tick.py (lines 5-14) |
These endpoints forward to the same TaskScheduler methods used by the Python tool, ensuring consistent behavior across interfaces.
Example API Usage
# List all tasks
curl -X GET https://your-agent-host/api/scheduler/tasks
# Create a new scheduled task
curl -X POST https://your-agent-host/api/scheduler/task/create \
-H "Content-Type: application/json" \
-d '{
"name":"hourly-report",
"system_prompt":"You are a reporting bot",
"prompt":"Generate the sales report for the last hour.",
"schedule":{"minute":"*/60","hour":"*","day":"*","month":"*","weekday":"*"},
"dedicated_context":false
}'
# Trigger the scheduler tick (normally run automatically)
curl -X POST https://your-agent-host/api/scheduler/tick
Background Job Loop and Persistence
The scheduler relies on python/helpers/job_loop.py to drive the tick loop. This background thread wakes every minute (configurable) and invokes TaskScheduler.tick(), ensuring that cron-style tasks trigger even when the main agent is idle.
State changes are persisted through python/helpers/state_monitor_integration.py, which marks the UI dirty when tasks change, triggering front-end refreshes. Chat contexts for scheduled tasks are temporarily saved via python/helpers/persist_chat.py, allowing tasks to resume with full conversation history if interrupted.
Summary
- Agent Zero provides a built-in task scheduler supporting cron-style, ad-hoc, and planned task types.
- The
TaskSchedulersingleton inpython/helpers/task_scheduler.pymanages persistence tousr/scheduler/tasks.jsonwith thread-safe updates. - Three task models—
ScheduledTask,AdHocTask, andPlannedTask—inherit fromBaseTaskand support lifecycle hooks. - The tick loop (driven by
python/helpers/job_loop.py) checks for due tasks every minute and executes them via_run_task(). - Manage tasks programmatically via
SchedulerTool(python/tools/scheduler.py) or remotely through HTTP API endpoints inpython/api/.
Frequently Asked Questions
What file format does Agent Zero use to store scheduled tasks?
Agent Zero persists all scheduled tasks as JSON in usr/scheduler/tasks.json. The SchedulerTaskList class handles atomic writes with internal locking to prevent corruption during concurrent updates.
How does Agent Zero prevent scheduled tasks from running simultaneously?
The TaskScheduler maintains a registry of running DeferredTask objects. Before executing a task, the _run_task() method atomically checks and updates the task state from IDLE to RUNNING. If a task is already running, subsequent calls will skip execution until the state returns to IDLE.
Can I trigger the scheduler manually instead of waiting for the background loop?
Yes. While python/helpers/job_loop.py automatically invokes TaskScheduler.tick() every minute, you can manually trigger a tick via the SchedulerTool or by calling the HTTP endpoint POST /scheduler/tick (implemented in python/api/scheduler_tick.py).
What is the difference between dedicated_context and shared context in scheduled tasks?
When creating a task via SchedulerTool.create_scheduled_task(), setting dedicated_context=True forces the scheduler to create a fresh AgentContext for each execution, ensuring isolation from other conversations. When False, the task may reuse existing contexts depending on the agent configuration, allowing for continuity across related operations.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →