Schedule Tools

from praisonaiagents import Agent
from praisonaiagents.tools import schedule_add, schedule_list, schedule_remove

agent = Agent(
    name="Reporter",
    instructions="Set up recurring reports for the user.",
    tools=[schedule_add, schedule_list, schedule_remove],
)
agent.start("Every week send me a weekly-digest summary")

from praisonaiagents import Agent
from praisonaiagents.tools import schedule_add, schedule_list, schedule_remove

agent = Agent(
    name="news-bot",
    instructions="""You help users stay informed.
    When asked, create schedules for news summaries.
    Use schedule_add with cron expressions for precise timing.""",
    tools=[schedule_add, schedule_list, schedule_remove],
)

# Agent will create: schedule_add("morning-news", "cron:0 7 * * *", "Summarize AI news")
agent.start("Send me an AI news summary every morning at 7am")

# Agent will create: schedule_add("meeting-prep", "in 20 minutes", "Prepare for standup")
agent.start("Remind me in 20 minutes to prepare for standup")

from praisonaiagents.tools import schedule_add

# Continuable (default) — reply resumes the conversation with the brief in context
schedule_add(
    name="morning-brief",
    schedule="cron:0 7 * * *",
    message="Summarise the morning news",
    deliver="telegram:123456",
)

# Fire-and-forget — reply stays a brand-new turn
schedule_add(
    name="backup-alert",
    schedule="cron:0 3 * * *",
    message="Backup ok",
    deliver="slack:C012345",
    continuable=False,
)

# Agent will call schedule_list() and schedule_remove("old-task")
agent.start("Show me my schedules and remove 'old-task'")

agent = Agent(
    name="scheduler",
    tools=["schedule_add", "schedule_list", "schedule_remove"],
)

from praisonaiagents.scheduler import get_default_store, set_default_store

# read the current default
store = get_default_store()

# swap it for a custom backend at process start
class MyStore: ...
set_default_store(MyStore())

from praisonaiagents.scheduler import ScheduleStoreProtocol

class MyDatabaseStore:
    """Any object with these methods works."""
    def add(self, job): ...
    def get(self, job_id) -> Optional[Any]: ...
    def list(self, agent_id=None) -> list: ...
    def update(self, job): ...
    def remove(self, job_id) -> bool: ...
    def get_by_name(self, name) -> Optional[Any]: ...
    def remove_by_name(self, name) -> bool: ...

assert isinstance(MyDatabaseStore(), ScheduleStoreProtocol)  # ✅

from praisonaiagents.tools.schedule_tools import set_store

my_store = MyDatabaseStore()
set_store(my_store)  # All schedule tools now use this store

from praisonaiagents.scheduler import SchedulerProviderProtocol, ScheduleLoop

# The built-in in-process loop is the default provider (also exported as
# InProcessScheduleProvider). Any object with start(on_due, store=...) + stop()
# satisfies the protocol — e.g. a cloud-scheduler webhook, systemd timer, or
# APScheduler cron trigger that calls engine.fire_due() on its own schedule.
engine = ScheduleLoop(on_trigger=handle_job)

# Serverless — fire when the external system pings us, no always-on thread:
def on_webhook():
    engine.fire_due()

from praisonaiagents.scheduler import ScheduleRunner, get_default_store

store = get_default_store()
runner = ScheduleRunner(store=store)

# Get jobs that are due right now
due_jobs = runner.get_due_jobs()

for job in due_jobs:
    print(f"Due: {job.name} — {job.message}")
    runner.mark_run(job)  # Updates last_run_at

from praisonaiagents.scheduler import get_default_store

store = get_default_store()

# Get last 10 executions for a job
history = store.get_history("job-abc123", limit=10)

for run in history:
    print(f"{run.job_name}: {run.status} ({run.duration:.1f}s)")
    if run.error:
        print(f"  Error: {run.error}")

from praisonaiagents.scheduler import ScheduleLoop

def handle_job(job):
    print(f"🔔 {job.name}: {job.message}")

loop = ScheduleLoop(on_trigger=handle_job, tick_seconds=30)
loop.start()

pre_run: "gh issue list --state open --search 'created:>10m ago' --json title"

pre_run: "test -s $HOME/.mail/inbox/unread"

pre_run: "test $(date +%u) -lt 6 && test $(date +%H) -ge 9 && test $(date +%H) -lt 18"

from praisonaiagents.scheduler.protocols import GateResult, JobConditionProtocol
from praisonai.scheduler.executor import ScheduledAgentExecutor

class DatabaseGate:
    def should_run(self, job) -> GateResult:
        rows = pending_rows()
        if rows == 0:
            return GateResult(run=False, reason="no pending rows")
        return GateResult(run=True, context=f"{rows} rows waiting")

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda agent_id: gateway.get_agent(agent_id),
    condition_resolver=lambda job: DatabaseGate(),
)

# bot.yaml — schedules work out of the box
agent:
  name: assistant
  tools:
    - schedule_add
    - schedule_list
    - schedule_remove

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}

praisonai bot start --config bot.yaml
# Agent can now self-schedule and BotOS executes + delivers results