Schedule Tools

Agent-centric scheduling — let agents add, list, pause, resume, update, and remove scheduled jobs

from praisonaiagents import Agent
from praisonaiagents.tools import (
    schedule_add, schedule_list, schedule_remove,
    schedule_pause, schedule_resume, schedule_update,
)

agent = Agent(
    name="Reporter",
    instructions="Manage the user's recurring reports — add, list, pause, resume, update, and remove.",
    tools=[
        schedule_add, schedule_list, schedule_remove,
        schedule_pause, schedule_resume, schedule_update,
    ],
)
agent.start("Every weekday at 8am send me a briefing, and pause it on weekends")

The user describes a recurring reminder; the agent registers a schedule job via schedule tools.

Schedule tools let your agents self-schedule reminders, recurring tasks, and one-shot jobs — all via simple tool calls. Optionally gate each tick with a cheap shell check via pre_run so expensive model turns only happen when there’s real work to do. No changes to the Agent class are needed.

Quick Start

1. Simple Usage

   from praisonaiagents import Agent
   from praisonaiagents.tools import schedule_add, schedule_list, schedule_remove
   
   agent = Agent(
       name="assistant",
       instructions="You can set reminders and schedules for the user.",
       tools=[schedule_add, schedule_list, schedule_remove],
   )
   
   agent.start("Remind me to check email every morning at 7am")

2. With Configuration

   from praisonaiagents.tools import schedule_list
   
   print(schedule_list())

The agent will call schedule_add with the appropriate schedule expression, and the job will be persisted to disk.

Available Tools

schedule_add

Add a new scheduled job.

Returns: Confirmation string with the job id.

schedule_list

List all scheduled jobs. Takes no parameters.

Returns: Formatted string listing every job with id, name, schedule, status, and message.

schedule_remove

Remove a scheduled job by name.

Returns: Confirmation or not-found message.

schedule_pause

Pause a schedule by name. Sets enabled=False so every due-check skips the job without deleting it — the counterpart to schedule_resume.

Returns: Confirmation or not-found message.

schedule_resume

Resume a paused schedule by name. Sets enabled=True so the job fires again on its next due tick.

Returns: Confirmation or not-found message.

schedule_update

Update a schedule’s cadence and/or message. Only the fields you pass change; empty values leave the current value untouched. Changing schedule clears last_run_at so the new cadence runs fresh.

Returns: Confirmation or not-found / error message.

The CLI exposes the same surface — see praisonai schedule pause / resume / update.

Schedule Expressions

Pick the format that matches how the job should recur.

Examples

Recurring Schedule

from praisonaiagents import Agent
from praisonaiagents.tools import schedule_add, schedule_list, schedule_remove

agent = Agent(
    name="news-bot",
    instructions="""You help users stay informed.
    When asked, create schedules for news summaries.
    Use schedule_add with cron expressions for precise timing.""",
    tools=[schedule_add, schedule_list, schedule_remove],
)

# Agent will create: schedule_add("morning-news", "cron:0 7 * * *", "Summarize AI news")
agent.start("Send me an AI news summary every morning at 7am")

One-Shot Reminder

# Agent will create: schedule_add("meeting-prep", "in 20 minutes", "Prepare for standup")
agent.start("Remind me in 20 minutes to prepare for standup")

Fire-and-forget vs. continuable

A delivered brief is continuable by default — a reply in the same chat resumes the job’s conversation. For pure alerts, set continuable=False so a reply stays a fresh turn.

from praisonaiagents.tools import schedule_add

# Continuable (default) — reply resumes the conversation with the brief in context
schedule_add(
    name="morning-brief",
    schedule="cron:0 7 * * *",
    message="Summarise the morning news",
    deliver="telegram:123456",
)

# Fire-and-forget — reply stays a brand-new turn
schedule_add(
    name="backup-alert",
    schedule="cron:0 3 * * *",
    message="Backup ok",
    deliver="slack:C012345",
    continuable=False,
)

See Scheduler Delivery → Continuable Delivery for the full seed contract.

List and Manage

# Agent will call schedule_list() and schedule_remove("old-task")
agent.start("Show me my schedules and remove 'old-task'")

One-Shot Job

from praisonaiagents.tools import schedule_add

schedule_add(
    name="deploy-ping",
    schedule="in 10 minutes",
    message="Kick off the deploy check-in",
    deliver="slack:C0123456",
    once=True,   # auto-remove after firing so it does not linger
)

Bounded by Count

Set max_runs to fire a fixed number of times, then auto-remove.

from praisonaiagents.tools import schedule_add

# "Remind me every hour, three times total."
schedule_add(
    name="deploy-recheck",
    schedule="*/1h",
    message="Check deploy status",
    deliver="slack:C0123456",
    max_runs=3,     # retires after the third fire
)

Bounded by Date

Set until to stop a recurring schedule after a wall-clock instant.

from praisonaiagents.tools import schedule_add

# "Every weekday at 8am, until Friday."
schedule_add(
    name="week-standup",
    schedule="cron:0 8 * * 1-5",
    message="Time for standup",
    deliver="telegram:-1001234567",
    tz="Europe/London",
    until="2026-09-25T17:00:00",   # timezone-resolved via tz
)

Combined: count and date

Combine both — whichever condition is met first retires the job.

from praisonaiagents.tools import schedule_add

# "Every hour, up to 10 times, but stop by end of day either way."
schedule_add(
    name="rollout-watch",
    schedule="*/1h",
    message="Check rollout health",
    deliver="slack:C0123456",
    tz="Europe/London",
    max_runs=10,                   # cap the count
    until="2026-09-25T23:59:00",   # and the end date
)

An unparseable until fails safe: the job is not fired and is retired by the claim path rather than running forever.

How Should It Stop?

Pick a stop condition to bound a recurring every / cron job.

Using String Tool Names

agent = Agent(
    name="scheduler",
    tools=[
        "schedule_add", "schedule_list", "schedule_remove",
        "schedule_pause", "schedule_resume", "schedule_update",
    ],
)

Timezones for at: and cron:

Which wall-clock time a schedule uses depends on the timezone it resolves to.

For a naive at: ISO timestamp (no Z, no +HH:MM) and the natural-language clock forms (at 9am, at 17:00), the parser resolves the zone in order:

1. The tz argument passed to the schedule (per-schedule).
2. The instance default (the scheduler.timezone key in config.yaml).
3. The PRAISONAI_SCHEDULE_TIMEZONE environment variable (process-wide default).
4. The machine’s local zone.

The offset attached is the one in force on the target date, so a summer date gets the DST offset and a winter date gets the standard offset. An at: string that already carries an offset (+05:30, Z, +00:00) is used verbatim — the offset in the string always wins.

cron: expressions follow the same 1–3 precedence but fall back to UTC at step 4, unchanged from previous releases. Only the naive-at: path defaults to the local zone.

The stored value is aware

As of PraisonAI #4732, parse_schedule returns a Schedule whose at is already an aware ISO string. A job authored on one machine and later polled by a runner in a different zone fires at the same wall-clock time the person originally typed — the guess is_due() used to make at fire time is gone.

from praisonaiagents.scheduler.parser import parse_schedule

# In Europe/London, in the winter, this stores an offset of +00:00
parse_schedule("at:2026-12-01T09:00:00").at   # -> "2026-12-01T09:00:00+00:00"

# In Europe/London, in the summer, this stores an offset of +01:00
parse_schedule("at:2026-07-01T09:00:00").at   # -> "2026-07-01T09:00:00+01:00"

# An explicit tz overrides the system zone
parse_schedule("at:2026-07-01T09:00:00", tz="America/New_York").at
# -> "2026-07-01T09:00:00-04:00"

# An explicit offset in the string is kept verbatim
parse_schedule("at:2026-07-01T09:00:00+02:00").at
# -> "2026-07-01T09:00:00+02:00"

DST is handled per target date

localize_wall_clock attaches the offset in force on the target date, not the offset in force right now. In Europe/London, at:2026-07-01T09:00:00 stamps +01:00 (BST) even if you parse it in January, and at:2026-12-01T09:00:00 stamps +00:00 (GMT) even if you parse it in July. You do not need to hand-compute offsets in the ISO string.

PRAISONAI_SCHEDULE_TIMEZONE fails loudly on bad names

An unknown IANA zone in PRAISONAI_SCHEDULE_TIMEZONE (or in the tz= argument) now raises ValueError from parse_schedule when it hits an at: or clock form, instead of being accepted silently and never firing. A Schedule(kind="at", at=<naive iso>) built directly hits the same check when is_due() reads it.

# Rejected at add time instead of firing never
PRAISONAI_SCHEDULE_TIMEZONE=Not/A/Zone \
  praisonai schedule add "x" -s "at:2026-07-01T09:00:00"
# -> ValueError: Unknown IANA timezone: 'Not/A/Zone'

Legacy stored jobs

Agent self-management flow

The agent creates, pauses, updates, resumes, and inspects a schedule entirely through tool calls in a single conversation.

Storage

Jobs are persisted to ~/.praisonai/config.yaml under the schedules key by default via ConfigYamlScheduleStore. The store is:

• Thread-safe for multi-agent scenarios
• Atomic writes (tmp + rename) to prevent corruption
• Auto-created on first use
• Auto-migrates legacy jobs.json data on first load

Shared Default Store

Every runtime shares one process-wide store via get_default_store() so a job authored by an agent is claimed by the gateway ticker, not silently dropped.

from praisonaiagents.scheduler import get_default_store, set_default_store

# read the current default
store = get_default_store()

# swap it for a custom backend at process start
class MyStore: ...
set_default_store(MyStore())

praisonaiagents.tools.schedule_tools.set_store(my_store) continues to work and now also repoints the canonical default under the hood.

Custom Store (ScheduleStoreProtocol)

Swap the default file store for any backend that implements ScheduleStoreProtocol:

from praisonaiagents.scheduler import ScheduleStoreProtocol

class MyDatabaseStore:
    """Any object with these methods works."""
    def add(self, job): ...
    def get(self, job_id) -> Optional[Any]: ...
    def list(self, agent_id=None) -> list: ...
    def update(self, job): ...
    def remove(self, job_id) -> bool: ...
    def get_by_name(self, name) -> Optional[Any]: ...
    def remove_by_name(self, name) -> bool: ...

assert isinstance(MyDatabaseStore(), ScheduleStoreProtocol)  # ✅

Inject it at startup so all agent schedule_add/list/remove calls use your store:

from praisonaiagents.tools.schedule_tools import set_store

my_store = MyDatabaseStore()
set_store(my_store)  # All schedule tools now use this store

set_store() now also repoints the process-wide scheduler.get_default_store() so the gateway tick and host bridge pick up the same backend. This is best-effort and logs a warning if the repoint fails; the tool store you passed is always authoritative for the agent tools.

Custom Provider (SchedulerProviderProtocol)

Swap the default in-process poll thread for any backend that decides when to fire:

from praisonaiagents.scheduler import SchedulerProviderProtocol, ScheduleLoop

# The built-in in-process loop is the default provider (also exported as
# InProcessScheduleProvider). Any object with start(on_due, store=...) + stop()
# satisfies the protocol — e.g. a cloud-scheduler webhook, systemd timer, or
# APScheduler cron trigger that calls engine.fire_due() on its own schedule.
engine = ScheduleLoop(on_trigger=handle_job)

# Serverless — fire when the external system pings us, no always-on thread:
def on_webhook():
    engine.fire_due()

See Scheduler Providers for full patterns.

Schedule Runner

The ScheduleRunner checks which jobs are due for execution:

from praisonaiagents.scheduler import ScheduleRunner, get_default_store

store = get_default_store()
runner = ScheduleRunner(store=store)

# Get jobs that are due right now
due_jobs = runner.get_due_jobs()

for job in due_jobs:
    print(f"Due: {job.name} — {job.message}")
    runner.mark_run(job)  # Updates last_run_at

Constructing ConfigYamlScheduleStore() directly still works — the process-wide get_default_store() returns the same class by default and is the recommended way to share one instance with the gateway tick and host bridge.

Hook Events

Schedule lifecycle events are available via the hook system. Both SCHEDULE_ADD and SCHEDULE_REMOVE are emitted by the schedule store, so they fire on a real persist/delete no matter who triggered it.

Both input types are importable from praisonaiagents.hooks.

Observing tool-driven schedules

Register hooks, then call the tools — the store emits the payload for you.

from praisonaiagents.hooks import HookEvent, HookResult, get_default_registry
from praisonaiagents.tools import schedule_add, schedule_remove

registry = get_default_registry()

@registry.on(HookEvent.SCHEDULE_ADD)
def on_add(event_data):
    print(f"ADD: {event_data.job_name} → {event_data.schedule} (enabled={event_data.enabled})")
    return HookResult.allow()

@registry.on(HookEvent.SCHEDULE_REMOVE)
def on_remove(event_data):
    print(f"REMOVE: {event_data.job_name} ({event_data.job_id})")
    return HookResult.allow()

schedule_add(name="daily-report", schedule="daily", message="Send the daily report")
schedule_remove("daily-report")

Execution History

Every scheduled job execution is logged as a RunRecord for auditing:

from praisonaiagents.scheduler import get_default_store

store = get_default_store()

# Get last 10 executions for a job
history = store.get_history("job-abc123", limit=10)

for run in history:
    print(f"{run.job_name}: {run.status} ({run.duration:.1f}s)")
    if run.error:
        print(f"  Error: {run.error}")

What happens on a failed one-shot

A failed once=True job is kept but disabled, so it still shows up in schedule_list with its failed history row. The spent fire is undone, so schedule_resume runs it again instead of retiring it unrun — or schedule_remove it if it is no longer needed. A run lost to a crash is handled separately: see Crash Recovery.

Executing Scheduled Jobs

Schedule tools create and persist jobs, but to actually execute them when they’re due, use ScheduleLoop:

from praisonaiagents.scheduler import ScheduleLoop

def handle_job(job):
    print(f"🔔 {job.name}: {job.message}")

loop = ScheduleLoop(on_trigger=handle_job, tick_seconds=30)
loop.start()

Pre-Run Condition Gate

Gate a scheduled tick on a cheap shell check so no model tokens are spent when there’s nothing to do.

1. Add pre_run to a schedule in bot.yaml

   # bot.yaml
   schedules:
     - name: new-issue-triage
       schedule: "*/5m"
       message: "Triage any new issues from the queue."
       pre_run: "gh issue list --state open --json number,title --search 'created:>1h ago'"

2. Every tick, PraisonAI evaluates pre_run before spending tokens

   Exit code	stdout	Outcome
   0	empty	Job runs with original message; no context added
   0	non-empty	Job runs; stdout appended as context (capped at 8 000 chars)
   non-zero	—	Job skipped; truncated stderr in reason (max 500 chars)
   timed out (>30 s)	—	Job skipped; process group killed; reason: pre-run gate timed out

Real-World Examples

Only triage when new issues exist:

pre_run: "gh issue list --state open --search 'created:>10m ago' --json title"

Only summarise inbox when there’s unread mail:

pre_run: "test -s $HOME/.mail/inbox/unread"

Guard against off-hours runs (Monday–Friday, 09:00–18:00):

pre_run: "test $(date +%u) -lt 6 && test $(date +%H) -ge 9 && test $(date +%H) -lt 18"

Custom Condition Gate

Any object implementing JobConditionProtocol can replace the default shell gate — a Python callable, an MCP probe, a database check.

from praisonaiagents.scheduler.protocols import GateResult, JobConditionProtocol
from praisonai.scheduler.executor import ScheduledAgentExecutor

class DatabaseGate:
    def should_run(self, job) -> GateResult:
        rows = pending_rows()
        if rows == 0:
            return GateResult(run=False, reason="no pending rows")
        return GateResult(run=True, context=f"{rows} rows waiting")

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda agent_id: gateway.get_agent(agent_id),
    condition_resolver=lambda job: DatabaseGate(),
)

Pass condition_resolver=False to disable gating entirely. The default resolver automatically activates ShellConditionGate for any job that has a pre_run value.

BotOS Integration

When using BotOS (multi-platform bot orchestrator), scheduled jobs execute automatically — no ScheduleLoop needed. BotOS runs its own 30-second schedule tick alongside all bots:

• Agents create jobs via schedule_add during conversations
• BotOS detects due jobs every 30 seconds
• The originating agent processes the job message
• Results are delivered back to the originating platform (Telegram, Discord, etc.)

# bot.yaml — schedules work out of the box
agent:
  name: assistant
  tools:
    - schedule_add
    - schedule_list
    - schedule_remove

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}

praisonai bot start --config bot.yaml
# Agent can now self-schedule and BotOS executes + delivers results

Architecture

Schedule tools follow PraisonAI’s core principles:

• Agent-centric — tools, not Agent parameters
• Lazy-loaded — zero import cost until used
• Protocol-driven — ScheduleStoreProtocol makes stores swappable
• No Agent bloat — the Agent class is unchanged
• Thread-safe — safe for multi-agent workflows
• Pluggable — set_store() lets any backend replace the default file store

See Also

Best Practices

Related