pip install agent-scheduler
# Optional: for REST API support
pip install agent-scheduler[api]

# Create a recurring job
agent-scheduler add --name "daily-report" --handler report.generate \
  --cron "0 9 * * MON-FRI" --priority high --tags reporting,daily

# Create a one-time delayed job
agent-scheduler add --name "cleanup-temp" --handler cleanup.run \
  --delay 3600 --tags maintenance

# Create a job with retry policy
agent-scheduler add --name "api-poll" --handler poll.endpoint \
  --cron "*/5 * * * *" --max-retries 3 --timeout 30

# List all jobs
agent-scheduler list

# Show job details
agent-scheduler show daily-report

# Manually run a job
agent-scheduler run daily-report

# Run all due jobs
agent-scheduler run-due

# View execution history
agent-scheduler history daily-report --limit 20

# View statistics
agent-scheduler stats

# Pause/resume a job
agent-scheduler pause daily-report
agent-scheduler resume daily-report

# Delete a job
agent-scheduler delete daily-report --force

# Create a webhook for job completion events
agent-scheduler webhook add \
  --name "slack-notify" \
  --url "https://hooks.slack.com/services/XXX" \
  --events job.completed,job.failed \
  --tags monitoring \
  --secret "my-signing-secret"

# List webhooks
agent-scheduler webhook list

# View delivery history
agent-scheduler webhook deliveries

# Delete a webhook
agent-scheduler webhook delete <webhook-id> --force

# List available templates
agent-scheduler template list

# Show template details
agent-scheduler template show health-check

# Create a job from a template
agent-scheduler template use health-check \
  --name "api-health" \
  --payload '{"endpoint": "https://api.example.com/health"}'

# Create a custom template
agent-scheduler template add \
  --name "my-pipeline" \
  --handler pipeline.run \
  --description "My data pipeline" \
  --category data-pipeline \
  --cron "0 */4 * * *" \
  --tags pipeline \
  --max-retries 2 \
  --required-fields "payload.pipeline_id"

# Show the full analytics dashboard
agent-scheduler analytics

# Health report for a specific job
agent-scheduler health daily-report

# Validate a cron expression
agent-scheduler cron validate "0 9 * * MON-FRI"

# Describe a cron expression in plain English
agent-scheduler cron describe "*/15 * * * *"
# => Every 15 minutes

# Preview the next 10 runs
agent-scheduler cron preview "0 9 * * *" --count 10

# Build a cron expression from parameters
agent-scheduler cron build --frequency daily --hour 9 --minute 30
# => 30 9 * * *
agent-scheduler cron build --frequency weekly --day monday --hour 9
# => 0 9 * * 0
agent-scheduler cron build --frequency every-n-minutes --n 15
# => */15 * * * *

# Start the REST API server
agent-scheduler api --host 0.0.0.0 --port 8080

# Or start the scheduler daemon (poll loop + MCP)
agent-scheduler start

# Or start the MCP server
agent-scheduler serve --port 8080

from agent_scheduler import Scheduler, Job, Priority, RetryPolicy
from agent_scheduler.webhook import Webhook, WebhookEvent

# Create scheduler
scheduler = Scheduler()

# Add a job
job = Job(
    name="daily-report",
    handler="report.generate",
    cron="0 9 * * MON-FRI",
    priority=Priority.HIGH,
    retry_policy=RetryPolicy(max_retries=3, backoff_seconds=60),
    tags=["reporting"],
    payload={"format": "pdf", "recipients": ["team@example.com"]},
)
scheduler.add_job(job)

# Add a webhook
webhook = Webhook(
    name="slack-notify",
    url="https://hooks.slack.com/services/XXX",
    events=[WebhookEvent.JOB_COMPLETED, WebhookEvent.JOB_FAILED],
    tags=["reporting"],
    secret="my-signing-secret",
)
scheduler.webhooks.create_webhook(webhook)

# Use a template
from agent_scheduler.templates import TemplateManager
manager = TemplateManager(store=scheduler.store)
job = manager.instantiate("health-check", payload={"endpoint": "https://api.example.com/health"})
scheduler.add_job(job)

# Run due jobs (async)
import asyncio
executions = asyncio.run(scheduler.run_due_jobs())

# Get stats
stats = scheduler.get_stats()
print(f"Active: {stats.active_jobs}, Failed: {stats.failed_jobs}")

from agent_scheduler.analytics import AnalyticsEngine

engine = AnalyticsEngine(scheduler=scheduler)

# Full dashboard
dashboard = engine.dashboard()
print(f"Health: {dashboard.overall_health_grade} ({dashboard.overall_health_score}/100)")
print(f"At-risk jobs: {dashboard.at_risk_jobs}")

# Single job health
job = scheduler.get_job_by_name("daily-report")
report = engine.job_report(job)
print(f"{report.job_name}: {report.health_grade} ({report.health_score}/100)")

from agent_scheduler.cron_helper import (
    validate_cron, describe_cron, preview_runs, suggest_cron, CronBuilder
)

# Validate
result = validate_cron("0 9 * * MON-FRI")
assert result.is_valid

# Describe
print(describe_cron("*/15 * * * *"))  # => "Every 15 minutes"

# Preview next runs
runs = preview_runs("0 9 * * *", n=5)

# Build from parameters
expr = suggest_cron("daily", hour=9, minute=30)  # => "30 9 * * *"
expr = suggest_cron("weekdays", hour=9)           # => "0 9 * * 0-4"

from agent_scheduler.notifications import (
    Notification, NotificationLevel, ChannelManager,
    SlackChannel, DiscordChannel, EmailChannel, HttpChannel,
)

# Set up channels
mgr = ChannelManager()
mgr.add_channel(SlackChannel(
    webhook_url="https://hooks.slack.com/services/XXX",
    name="ops-slack",
))
mgr.add_channel(DiscordChannel(
    webhook_url="https://discord.com/api/webhooks/XXX",
), levels=[NotificationLevel.ERROR])  # Only errors to Discord

# Send a notification
notif = Notification(
    title="Job Failed",
    message="daily-report failed after 3 retries",
    level=NotificationLevel.ERROR,
    job_name="daily-report",
    event_type="job.failed",
    metadata={"retry_count": 3},
)
results = asyncio.run(mgr.send(notif))

# Or create from config dict
from agent_scheduler.notifications import create_channel_from_config
ch = create_channel_from_config({
    "type": "slack",
    "webhook_url": "https://hooks.slack.com/services/XXX",
    "channel": "#alerts",
})

from agent_scheduler import Scheduler, Job, HandlerRegistry

registry = HandlerRegistry()

# Sync handler
def my_handler(payload):
    # Do work
    return {"status": "done", "processed": len(payload)}

registry.register("my.handler", my_handler)

# Async handler
async def async_handler(payload):
    await some_async_work()
    return {"async": True}

registry.register("my.async_handler", async_handler)

scheduler = Scheduler(handler_registry=registry)