Scheduler Module

Deployment scheduling with provider-agnostic design

Scheduler Module

The Scheduler module provides deployment scheduling capabilities with a provider-agnostic design.

Import

python

# Wrapper-tier (praisonai): deployment scheduler + safety primitives
from praisonai.scheduler import ScheduleParser, DeploymentScheduler
from praisonai.scheduler import RunPolicy, PromptScanResult

# Bot-tier (praisonai-bot): execution primitives — canonical since v4.6.122
from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult
from praisonai_bot.scheduler.condition_gate import ShellConditionGate

Quick Example

python

from praisonai.scheduler import DeploymentScheduler

# Create scheduler
scheduler = DeploymentScheduler()

# Schedule deployment every hour
scheduler.schedule(interval_minutes=60)

# Start the scheduler
scheduler.start()

Classes

`DeploymentScheduler`

Minimal deployment scheduler with provider-agnostic design.

Features:

Simple interval-based scheduling
Thread-safe operation
Extensible deployer factory pattern

python

from praisonai.scheduler import DeploymentScheduler

scheduler = DeploymentScheduler()
scheduler.schedule(interval_minutes=30)
scheduler.start()

# Later, stop the scheduler
scheduler.stop()

`ScheduleParser`

Parses schedule expressions into scheduling parameters. ScheduleParser is also re-exported from praisonai.scheduler.shared and used internally by AgentScheduler / AsyncAgentScheduler.

python

from praisonai.scheduler import ScheduleParser

# Parse cron-like expressions
schedule = ScheduleParser.parse("every 30 minutes")
schedule = ScheduleParser.parse("daily at 09:00")

`DeployerInterface`

Abstract interface for deployers to ensure provider compatibility.

python

from praisonai.scheduler import DeployerInterface

class MyDeployer(DeployerInterface):
    def deploy(self) -> bool:
        """Execute deployment. Returns True on success."""
        # Your deployment logic
        return True

Methods

`DeploymentScheduler.schedule(interval_minutes)`

Schedule deployments at a fixed interval.

Parameters:

interval_minutes (int): Minutes between deployments

`DeploymentScheduler.start()`

Start the scheduler in a background thread.

`DeploymentScheduler.stop()`

Stop the scheduler.

`DeploymentScheduler.is_running()`

Check if the scheduler is currently running.

Returns: bool

Example: Custom Deployer

python

from praisonai.scheduler import DeploymentScheduler, DeployerInterface

class DockerDeployer(DeployerInterface):
    def __init__(self, image_name: str):
        self.image_name = image_name

    def deploy(self) -> bool:
        import subprocess
        try:
            subprocess.run(
                ["docker", "push", self.image_name],
                check=True
            )
            return True
        except subprocess.CalledProcessError:
            return False

# Use custom deployer
scheduler = DeploymentScheduler(deployer=DockerDeployer("myapp:latest"))
scheduler.schedule(interval_minutes=60)
scheduler.start()

Timezones

A naive at: timestamp (no Z, no +HH:MM offset) and every cron: expression resolve their timezone in this order:

Precedence	Source	Where it is set
1	`sched.tz` on the schedule	The `tz` field on `Schedule` / the `tz` argument to `parse_schedule`
2	Instance default (`default_timezone`)	`scheduler.timezone` (or `scheduler.tz`) key in `~/.praisonai/config.yaml`, read into the store and passed to `is_due`
3	`PRAISONAI_SCHEDULE_TIMEZONE`	Process-wide environment variable
4	`UTC`	Fallback when none of the above is set

resolve_schedule_timezone in praisonaiagents.scheduler.due implements the env-var and UTC fallback; is_due(job, now, default_timezone=...) threads the instance default through.

python

from praisonaiagents.scheduler import parse_schedule
from praisonaiagents.scheduler.due import resolve_schedule_timezone, is_due

# Per-schedule tz — the offset the parser stamps on the Schedule
schedule = parse_schedule("at:2026-03-01T09:00:00", tz="America/New_York")

# Env-var / UTC fallback used when no tz is supplied
tzinfo = resolve_schedule_timezone()  # PRAISONAI_SCHEDULE_TIMEZONE, else UTC

yaml

# ~/.praisonai/config.yaml — process-wide instance default
scheduler:
  timezone: Europe/London

ScheduleJob — Pre-Run Gate Fields

Since PR #2238, ScheduleJob (in praisonaiagents.scheduler.models) has two optional fields for the pre-run condition gate:

Field	Type	Default	Description
`pre_run`	`Optional[str]`	`None`	Shell command run before each tick. Exit 0 + stdout → run (stdout seeds the prompt); non-zero → skip
`condition`	`Optional[str]`	`None`	Advisory natural-language label (round-tripped, not enforced by the default gate)

python

from praisonaiagents.scheduler import ScheduleJob, Schedule

job = ScheduleJob(
    name="inbox-watch",
    schedule=Schedule(kind="every", every_seconds=300),
    pre_run="scripts/new_mail.sh",
    condition="new mail",
    message="Summarise these new emails.",
)

Timezones

parse_schedule (in praisonaiagents.scheduler.parser) resolves the timezone for a naive at: timestamp and the clock forms (at 9am, at 17:00) in order:

The tz= argument passed to parse_schedule / schedule_add (per-schedule).
The instance default — the scheduler.timezone (or scheduler.tz) key in config.yaml, which ConfigYamlScheduleStore reads into its _default_timezone and passes to is_due().
The PRAISONAI_SCHEDULE_TIMEZONE environment variable.
The machine’s local zone, with the UTC offset in force on the target date (DST-correct).

An at: string that already carries an offset (+05:30, Z, +00:00) is used verbatim. cron: follows the same 1–3 precedence but falls back to UTC at step 4.

Parse-time stamping vs the `is_due()` fallback

As of PraisonAI #4732, the zone is attached at parse time by praisonaiagents.scheduler.due.localize_wall_clock, so parse_schedule returns a Schedule whose at is already an aware ISO string:

python

from praisonaiagents.scheduler.parser import parse_schedule

parse_schedule("at:2026-07-01T09:00:00", tz="Europe/London").at  # -> "2026-07-01T09:00:00+01:00"
parse_schedule("at:2026-12-01T09:00:00", tz="Europe/London").at  # -> "2026-12-01T09:00:00+00:00"

is_due() keeps a legacy branch for naive strings that did not come through the parser — jobs written before this PR, hand-edited config.yaml entries, or a Schedule(kind="at", at=<naive iso>) built directly. That branch calls the same localize_wall_clock helper, so both paths agree. An unknown IANA zone raises ValueError from either path instead of firing never.

The instance default

The store’s instance default flows through is_due(job, now, default_timezone=...) and applies to naive at: values that omit a per-schedule tz. Set it in config.yaml:

yaml

scheduler:
  timezone: Asia/Kolkata   # naive at: with no per-schedule tz reads/stamps as +05:30

is_due() accepts the same value directly when you evaluate a job yourself:

python

from praisonaiagents.scheduler.due import is_due

is_due(job, now, default_timezone="Asia/Kolkata")

ScheduleStoreProtocol — per-principal scoping

Each store method accepts an optional principal — the resolved end-user identity used to isolate one gateway user’s automations from another’s. None returns everything (global / single-tenant behaviour), so every call is backward-compatible.

Method	Signature	Behavior when `principal` is given
`list`	`list(agent_id=None, principal=None) -> List[ScheduleJob]`	Only jobs where `job.principal == principal` are returned. `None` returns everything (pre-scoping global).
`get_by_name`	`get_by_name(name, principal=None) -> Optional[ScheduleJob]`	A job owned by a different identity is treated as not found (returns `None`) so a user cannot read another’s automation by guessing its name.
`remove_by_name`	`remove_by_name(name, principal=None) -> bool`	A job owned by a different identity is not removed (returns `False`) so a user cannot delete another’s automation.

python

from praisonaiagents.scheduler import get_default_store

store = get_default_store()

alice_jobs = store.list(principal="telegram:12345")     # owner-filtered listing
store.get_by_name("morning-brief", principal="bob")     # -> None if owned by alice
store.remove_by_name("morning-brief", principal="bob")  # -> False; job intact

On FileScheduleStore, add, update, remove, and remove_by_name raise OSError when the write to disk fails, and log_run() is safe across processes — see Async Scheduler → Store write durability.

ScheduleJob.principal (Optional[str] = None) round-trips through to_dict/from_dict and is omitted when None, so existing on-disk store files stay byte-identical.

The same scoping was extended to SuggestionStore — list_pending, accept, and dismiss take a principal and refuse cross-owner mutations, with the pending cap and dedup window measured per owner. See Automation Suggestions → Multi-tenant isolation for the suggestion-store surface.

ScheduledAgentExecutor

ScheduledAgentExecutor lives in the bot tier (praisonai-bot). The canonical import is:

python

from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult

Constructor reference

Parameter	Type	Default	Description
`runner`	`ScheduleRunner`	required	An SDK `ScheduleRunner` instance
`agent_resolver`	`Callable[[Optional[str]], Any]`	required	`(agent_id) -> Agent`; may return `None`
`delivery_handler`	`Optional[Callable[..., Any]]`	`None`	Async `(delivery, text) -> None`; routes to a channel bot’s `send_message()`
`on_success`	`Optional[Callable[..., None]]`	`None`	Called with `(job, result)` on success
`on_failure`	`Optional[Callable[..., None]]`	`None`	Called with `(job, error)` on failure
`run_policy`	`Optional[RunPolicy]`	`None`	Wrapper safety gate — scopes tools, scans prompt, persists audit, delivers failure summary
`condition_resolver`	`None \	False \	callable`	`None`	`None` = auto `ShellConditionGate` when `pre_run` set; `False` = gating disabled; callable `(job) → gate \	None` = custom resolver

Public methods

Method	Return type	Description
`async tick()`	`AsyncIterator[JobResult]`	Check for due jobs and execute them, yielding one `JobResult` per job
`async tick_all()`	`List[JobResult]`	Like `tick()` but collects all results into a list
`async run_loop(interval=15.0, *, max_ticks=None)`	`None`	Convenience loop that calls `tick()` at a fixed interval

Example

python

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
from praisonai_bot.scheduler import ScheduledAgentExecutor

agent = Agent(name="assistant", instructions="You are helpful.")
store = FileScheduleStore()
runner = ScheduleRunner(store)

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda agent_id: agent,
)

import asyncio

async def main():
    async for result in executor.tick():
        print(f"Job {result.job.name}: {result.status} ({result.duration:.1f}s)")

asyncio.run(main())

`condition_resolver` values

Value	Behaviour
`None` (default)	Auto-uses `ShellConditionGate` for jobs that have `pre_run` set
`False`	Gating disabled for all jobs
`callable(job) -> gate \	None`	Custom resolver — return a `JobConditionProtocol` instance or `None` to skip gating

JobConditionProtocol and GateResult

JobConditionProtocol and GateResult are exported from praisonaiagents.scheduler:

python

from praisonaiagents.scheduler import JobConditionProtocol, GateResult

GateResult — dataclass with run: bool, context: Optional[str], reason: Optional[str]
JobConditionProtocol — protocol for custom gate implementations
SchedulerProviderProtocol — trigger backend protocol deciding when jobs fire; ScheduleLoop (aliased InProcessScheduleProvider) is the default. See Scheduler Providers.

`RunPolicy`

Run-scoped guardrail for unattended scheduled agent runs. Scopes the toolset, scans the assembled prompt for injection patterns, persists a durable output audit, and supports fail-closed delivery on failure.

python

from praisonai.scheduler import RunPolicy, PromptScanResult
from praisonai_bot.scheduler import ScheduledAgentExecutor

policy = RunPolicy(
    audit_dir="/var/log/praisonai/runs",
    deliver_on_failure=True,
)

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda _id: agent,
    run_policy=policy,
)

See Scheduled Run Policy for the full reference.

`PromptScanResult`

Return type from RunPolicy.scan_prompt() and custom scanner callables.

Field	Type	Description
`ok`	`bool`	`True` if the prompt passed the scan
`reason`	`Optional[str]`	Human-readable reason when `ok` is `False`

Scheduler Pre-Run Gate – Full user-facing documentation
Scheduled Run Policy – Safety gate for unattended runs
praisonai-bot SDK – Bot-tier package reference
Deploy Module – Deployment functionality
CLI Scheduler – CLI scheduling commands

Scheduler Module