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()

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
• 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