Scheduler Module

Deployment scheduling with provider-agnostic design

Scheduler Module

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

Since v4.6.122, the execution primitives (ScheduledAgentExecutor, ShellConditionGate, JobResult) live in praisonai-bot. Import them from praisonai_bot.scheduler.* in new code. Safety primitives (RunPolicy, PromptScanResult) stay in the wrapper at praisonai.scheduler.*. The old executor/gate paths remain importable as backward-compatibility shims.

Import

# 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

from praisonai.scheduler.executor import ScheduledAgentExecutor and from praisonai.scheduler.condition_gate import ShellConditionGate still work as backward-compatibility shims when the praisonai wrapper is installed.

Quick Example

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

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.

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.

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

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.

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

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

An at: string that already carries an offset (at:2026-03-01T09:00:00+05:30, at:2026-03-01T09:00:00Z) is unaffected — the offset in the string always wins. See Schedule Tools → Timezones.

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)

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:

1. The tz= argument passed to parse_schedule / schedule_add (per-schedule).
2. 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().
3. The PRAISONAI_SCHEDULE_TIMEZONE environment variable.
4. 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:

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:

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:

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.

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

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:

from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult

from praisonai.scheduler.executor import ScheduledAgentExecutor works as a backward-compatible shim when the praisonai wrapper is installed alongside praisonai-bot.

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

Atomic claims: When the backing store supports claim_due, each due job is reserved under a cross-process lock so it fires at most once across all tickers/processes/hosts. Stores without that support fall back to the non-atomic get_due_jobs path.

Example

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:

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.

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

Related

• 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