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()
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.",
)
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
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,
)
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 |
