mcpaisuite-schedulermcp 1.0.5

Task scheduling for AI agents — cron, intervals, delays via MCP

schedulermcp

Task scheduling for AI agents -- cron, intervals, delays via MCP

Part of the MCP AI Suite.

Features

• Cron scheduling -- 5-field cron expressions (*, */N, N-M, N,M; weekday 0 = Monday) for recurring jobs
• Interval and one-shot jobs -- fixed-interval repeats or single delayed execution
• Watch jobs -- event-driven triggers that monitor shell commands and fire on condition changes
• Condition evaluation -- rich condition syntax with numeric comparisons, contains, and variable interpolation
• Pluggable executors -- log, webhook, or kernel (routes job goals through kernelmcp)
• Pluggable stores -- in-memory or SQLite for job persistence across restarts
• Exponential backoff retry with configurable max failures
• Job lifecycle -- create, pause, resume, cancel, delete with full history tracking

Installation

pip install mcpaisuite-schedulermcp
# Optional extras:
pip install mcpaisuite-schedulermcp[dev]       # Development tools
pip install mcpaisuite-schedulermcp[all]       # All integrations
pip install mcpaisuite-schedulermcp[kernel]    # KernelMCP executor integration
pip install mcpaisuite-schedulermcp[webhook]   # Webhook executor

Quick Start

from schedulermcp import SchedulerFactory

scheduler = SchedulerFactory.create(store="sqlite", sqlite_path="scheduler.db")
job = await scheduler.schedule(
    goal="Generate daily status report",
    job_type="cron",
    cron="0 9 * * *",
)
scheduler.start()  # Start background tick loop

MCP Server

schedulermcp serve

Configuration

Variable	Default	Description
SCHEDULERMCP_STORE	memory	Job store: memory or sqlite
SCHEDULERMCP_EXECUTOR	log	Executor: log, webhook, or kernel
SCHEDULERMCP_SQLITE_PATH	scheduler.db	SQLite database path
SCHEDULERMCP_TICK_INTERVAL	15	Seconds between scheduler ticks

API Reference

SchedulerPipeline

Manages scheduled jobs -- create, execute, track, persist.

await scheduler.schedule(goal, job_type="once", cron="", interval_seconds=0,
                         delay_seconds=0, namespace="default") -> Job
await scheduler.cancel(job_id) -> bool
await scheduler.pause(job_id) -> bool
await scheduler.resume(job_id) -> bool
await scheduler.list_jobs(namespace="", status="") -> list[Job]
await scheduler.get_history(job_id) -> list[JobResult]
await scheduler.stats(namespace="") -> SchedulerStats
scheduler.start()   # Start background loop
scheduler.stop()    # Stop background loop

SchedulerFactory

SchedulerFactory.default()               # SQLite store, log executor
SchedulerFactory.from_env()              # Build from environment variables
SchedulerFactory.create(store="sqlite", executor="kernel", kernel_pipeline=kernel, ...)

Architecture

SchedulerPipeline runs a background tick loop that checks for due jobs on each tick. Time-based jobs (cron, interval, once) are evaluated against their next_run timestamp. Watch jobs execute a shell command and evaluate a condition expression against the output. The pluggable executor system allows routing job execution through a simple logger (dev), a webhook endpoint, or the full kernelmcp pipeline for autonomous agent-driven execution.

Testing

pip install -e ".[dev]"
pytest tests/ -v

License

AGPL-3.0 — see LICENSE.

Open source for individuals and open-source projects. For commercial use in closed-source products, a commercial license is available — contact [e-mail].