Scheduler Providers

Choose when scheduled jobs fire — in-process poll (default), systemd/launchd timer, cloud webhook, cron, or serverless

A scheduler provider decides when to fire. The default polls in-process; swap it for systemd, a cloud scheduler webhook, or a cron trigger without changing what fires.

A provider never decides what fires — it just calls on_due whenever a tick should occur, and the shared runner and store claim and fire due jobs.

Which Provider?

Pick in-process when one host runs all the time; pick an external provider when firing should come from elsewhere.

Quick Start

1. Default — in-process poll (nothing to configure)

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleLoop
   
   agent = Agent(name="assistant", instructions="You set reminders.")
   
   loop = ScheduleLoop(on_trigger=lambda job: agent.start(job.message))
   loop.start()  # zero-arg — same as before

2. External / serverless — event-driven, no always-on thread

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleLoop
   
   agent = Agent(name="assistant", instructions="You set reminders.")
   
   # Reuse the built-in loop as a pure "claim + fire" engine — do NOT call start().
   engine = ScheduleLoop(on_trigger=lambda job: agent.start(job.message))
   
   # In your webhook / cron / systemd-timer handler:
   def on_cloud_scheduler_ping():
       engine.fire_due()  # one tick — no daemon thread

How It Works

The provider owns the trigger; the runner and store own the firing.

on_due is the seam: the provider calls it whenever its trigger fires, and everything downstream — claiming due jobs and firing them — happens in the runner and store.

Provider Options

Two ways to drive firing — one always-on, one event-driven.

Any object with start(on_due, store=...) and stop() satisfies SchedulerProviderProtocol.

from praisonaiagents.scheduler import (
    SchedulerProviderProtocol,
    InProcessScheduleProvider,
    ScheduleLoop,
)

assert InProcessScheduleProvider is ScheduleLoop
assert isinstance(ScheduleLoop(on_trigger=lambda job: None), SchedulerProviderProtocol)

Sharing the store across providers

When a webhook, systemd timer, or K8s CronJob triggers engine.fire_due() and the gateway or wrapper is also polling, all of them should read the same store so a job authored anywhere is claimed once and only once.

from praisonaiagents.scheduler import ScheduleLoop, get_default_store

engine = ScheduleLoop(
    store=get_default_store(),         # share the canonical default
    on_trigger=lambda job: handle(job),
)

For deployments that need a different backend at process start, override once — every downstream consumer (agent tools, gateway tick, host bridge) picks it up.

import os
from praisonaiagents.scheduler import set_default_store
from myproject.stores import PostgresScheduleStore

set_default_store(PostgresScheduleStore(dsn=os.environ["DB_URL"]))

All three consumers resolve their store the same way, so a job written on any surface fires on every tick loop.

User Interaction Flow

A reminder set through the agent still fires even when a cloud scheduler drives the tick.

Configuration Options

Providers are protocol-only — see the auto-generated SDK reference for full signatures.

Common Patterns

Cloud webhook

Cloud Scheduler, EventBridge, or a GitHub Actions cron posts to an HTTP handler that calls fire_due().

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleLoop

agent = Agent(name="assistant", instructions="You set reminders.")
engine = ScheduleLoop(on_trigger=lambda job: agent.start(job.message))

# e.g. FastAPI / Flask route the cloud scheduler pings on a schedule
def handle_scheduler_webhook():
    engine.fire_due()
    return {"ok": True}

systemd / launchd timer

A timer unit runs a short command on schedule; the command fires one tick and exits.

# fire_once.py — invoked by a systemd timer (OnCalendar=*:0/5)
from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleLoop

agent = Agent(name="assistant", instructions="You set reminders.")
ScheduleLoop(on_trigger=lambda job: agent.start(job.message)).fire_due()

Long-running poller under any supervisor

praisonai schedule run is a zero-dependency way to host the default in-process poller under any process supervisor (systemd, pm2, tmux, foreman) — no need to run the whole BotOS gateway. It calls ScheduleLoop.run_forever(poll_seconds=…) in the foreground.

# /etc/systemd/system/praisonai-schedules.service
[Unit]
Description=PraisonAI standalone schedule poller
After=network.target

[Service]
Type=simple
ExecStart=/usr/local/bin/praisonai schedule run --poll 15
Restart=on-failure
User=praisonai

[Install]
WantedBy=multi-user.target

Kubernetes CronJob

A short-lived pod runs fire_due() and exits — no long-running scheduler pod.

# entrypoint.py — the CronJob pod's command
from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleLoop

agent = Agent(name="assistant", instructions="You set reminders.")
ScheduleLoop(on_trigger=lambda job: agent.start(job.message)).fire_due()

Best Practices

Related