Multiple Agents on Different Schedules

Real monitoring is rarely one cadence. You want a price check every thirty seconds, an anomaly scan every ten minutes, and a digest once an hour. Three rhythms, one process. A CronJob binds one agent to one interval, so a fleet needs one job per agent. CronJob.run_many builds them, starts them together, and blocks once.

Overview

Feature	Description
One job per agent	Each agent gets its own interval and its own scheduler thread
Isolated	A failing agent does not delay or stop its siblings
Per-job error budgets	`max_consecutive_errors` is set per agent, not per fleet
Blocking or not	`block=True` runs it as your main loop; `block=False` hands control back

run_many([...])
     │
     ├── Price-Checker    every 30s  ──> own thread, own error counters
     ├── Anomaly-Scanner  every 10m  ──> own thread, own error counters
     └── Hourly-Digest    every 1h   ──> own thread, own error counters
              │
              └── blocks once, here, until Ctrl-C or all jobs stop

Three agents, three cadences

from swarms import Agent, CronJob

price_agent = Agent(
    agent_name="Price-Checker",
    system_prompt="Report the current price and flag any move over 2%. Two lines maximum.",
    model_name="gpt-5.4",
    max_loops=1,
)

anomaly_agent = Agent(
    agent_name="Anomaly-Scanner",
    system_prompt="Scan for unusual patterns. Report only genuine anomalies, not noise.",
    model_name="gpt-5.4",
    max_loops=1,
)

digest_agent = Agent(
    agent_name="Hourly-Digest",
    system_prompt="Write a short digest of the last hour. Lead with what changed.",
    model_name="gpt-5.4",
    max_loops=1,
)

CronJob.run_many([
    {"agent": price_agent,   "interval": "30seconds", "task": "Check the BTC price."},
    {"agent": anomaly_agent, "interval": "10minutes", "task": "Scan recent data for anomalies."},
    {"agent": digest_agent,  "interval": "1hour",     "task": "Summarise the last hour."},
])

The schedule spec

Each entry is a mapping.

Key	Required	Purpose
`agent`	yes	The `Agent` or callable to schedule
`interval`	yes	`"30seconds"`, `"10minutes"`, `"1hour"`
`task`	yes	The task string handed to the agent on every tick
`job_id`	no	A readable identifier; generated if omitted
`callback`	no	`callback(output, task, metadata)` post-processor
`max_consecutive_errors`	no	Error budget for this agent only
`kwargs`	no	Dict forwarded to this agent’s `run`, e.g. `{"img": "chart.png"}`

A missing required key raises CronJobConfigError naming the index and the key, before anything starts.

Isolation is the point

Each job runs on its own scheduler thread, so the agents cannot interfere with each other. Give the flaky one a budget and leave the others alone:

CronJob.run_many([
    {"agent": price_agent,   "interval": "30seconds", "task": "Check the BTC price."},
    {
        "agent": anomaly_agent,
        "interval": "10minutes",
        "task": "Scan recent data for anomalies.",
        # Talks to a flaky upstream. Five failures in a row and this one stops.
        # The other two carry on regardless.
        "max_consecutive_errors": 5,
    },
    {"agent": digest_agent,  "interval": "1hour", "task": "Summarise the last hour."},
])

This isolation depends on the failure model: a task that raises is logged and retried on the next tick rather than killing its scheduler thread. Without that, one agent raising once would silently take itself offline while the fleet appeared healthy.

If any job exhausts its budget, run_many raises CronJobExecutionError once blocking ends, naming which jobs gave up and why.

Not blocking

When the schedule is not the main thing your process does — a web server, a bot, a notebook — pass block=False. You get the jobs back and own the lifecycle.

jobs = CronJob.run_many(schedules, block=False)

# ... your own main loop ...

for job in jobs:
    stats = job.get_execution_stats()
    print(f"{stats['job_id']:<20} ok={stats['execution_count']} failed={stats['error_count']}")

CronJob.stop_many(jobs)

stop_many continues past any job that fails to stop, so one stuck job cannot strand the rest.

Verifying cadence

Cadences are independent and accurate. Running three agents at 1s, 2s and 3s for six seconds produces roughly 6, 3 and 2 executions:

import time
from swarms.structs.cron_job import CronJob

class Echo:
    def __init__(self, name): self.name, self.runs = name, 0
    def run(self, task=None, **kwargs):
        self.runs += 1
        return f"{self.name}:{self.runs}"

fast, mid, slow = Echo("fast"), Echo("mid"), Echo("slow")

jobs = CronJob.run_many([
    {"agent": fast, "interval": "1second",  "task": "poll"},
    {"agent": mid,  "interval": "2seconds", "task": "check"},
    {"agent": slow, "interval": "3seconds", "task": "summarise"},
], block=False)

time.sleep(6.2)
CronJob.stop_many(jobs)

print(fast.runs, mid.runs, slow.runs)   # 6 3 2

This runs without API keys, since Echo stands in for a real agent. Anything exposing run(task=...) works.

Next steps

Failure Handling

Error budgets and live monitoring in depth

CronJob Quickstart

Start with a single agent

CronJob Reference

Full run_many and stop_many documentation

Runnable Examples

The example files in the repository

Index

Basic Examples

Model Providers

Single Agent

Multi-Agent Examples

Auto Agent Builder

Scheduled Agents

Applications

Research

Technical Analysis

Use Cases

Finance

Voice Agents

Integrations

Deployment

CLI

Multiple Agents on Different Schedules

Overview

Three agents, three cadences

The schedule spec

Isolation is the point

Not blocking

Verifying cadence

Next steps

Failure Handling

CronJob Quickstart

CronJob Reference

Runnable Examples

​Overview

​Three agents, three cadences

​The schedule spec

​Isolation is the point

​Not blocking

​Verifying cadence

​Next steps

Failure Handling

CronJob Quickstart

CronJob Reference

Runnable Examples

Overview

Three agents, three cadences

The schedule spec

Isolation is the point

Not blocking

Verifying cadence

Next steps