CronJob

Overview

The CronJob class wraps any callable (including Swarms agents) and turns it into a scheduled job that runs at specified intervals. It provides scheduling, failure handling, execution tracking, and optional callbacks for output customization. One CronJob binds one agent to one interval. For several agents on different cadences, use run_many.

Failure model. A task that raises is logged, recorded as a failed execution, and retried at the next interval, the way cron behaves. It does not take the schedule down, and a failing task is not retried any faster than its interval. Set max_consecutive_errors to stop a job that is failing every time; when that budget is exhausted the job stops and run() raises, so a dead schedule is never mistaken for a healthy one.

Constructor

Create a CronJob instance to schedule recurring agent executions.

from swarms import Agent, CronJob

agent = Agent(
    agent_name="Financial-Analyst",
    system_prompt="You are a financial analyst...",
    model_name="gpt-5.4",
    max_loops=1,
)

# Create scheduled job
cron = CronJob(
    agent=agent,
    interval="10minutes",
    job_id="financial_analysis_job"
)

Parameters

Union[Any, Callable]

default:"None"

The Swarms Agent instance or callable to be scheduled. Required: None raises CronJobConfigError. An object with a callable run attribute is called as agent.run(task=task, **kwargs); anything else is called as agent(task, **kwargs)

str

default:"None"

The interval string (e.g., “5seconds”, “10minutes”, “1hour”). Must be greater than zero. A malformed or empty value raises CronJobConfigError at construction. Leaving it None is accepted at construction, but run() then raises CronJobExecutionError

str

default:"None"

Optional unique identifier for the job. If not provided, one will be generated.

int

default:"None"

Stop the job after this many back-to-back failed executions. None (the default) never gives up and retries forever, which is what cron does. When the budget is exhausted the job stops and run() raises CronJobExecutionError.

Callable[[Any, str, dict], Any]

default:"None"

Optional callback function to customize output processing.Signature: callback(output: Any, task: str, metadata: dict) -> Any

output: The original output from the agent
task: The task that was executed
metadata: Dictionary containing job_id, timestamp, execution_count, etc.
Returns: The customized output

Attributes

str

Unique identifier for the job

bool

Flag indicating if the job is currently running

int

Number of successful executions. Failed executions are counted in error_count instead

float

Timestamp when the job was started

int

Total number of failed executions over the job’s life

int

Failures since the last success. Reset to 0 on any successful tick

Optional[Exception]

The most recent failure, or None if the job has never failed

Methods

run

Schedule and run the job with a specified task.

cron.run(
    task="Analyze Q4 earnings for AAPL",
    img="/path/to/chart.png"  # Optional image parameter
)

str

required

The task string to be executed by the agent

Any

Additional parameters to pass to the agent’s run method (e.g., img, imgs, correct_answer, streaming_callback)

schedule.Job

The scheduled job, returned once the schedule stops

Raises:

CronJobExecutionError: If scheduling failed (including when no interval was configured), or if the job gave up after exhausting max_consecutive_errors. The message names the failure count and the last error.

Behavior:

Schedules the task according to the configured interval
Starts the background execution thread
Blocks the calling thread until stop() is called, KeyboardInterrupt is received, or the job exhausts its error budget
On KeyboardInterrupt it stops the job itself and returns normally
A task that raises is logged and retried at the next interval; it does not stop the schedule

Calling the instance, cron("task", **kwargs), is the same as cron.run("task", **kwargs).

batched_run

Schedule several tasks on the same interval, then block like run().

tasks = [
    "Analyze AAPL stock",
    "Analyze GOOGL stock",
    "Analyze MSFT stock"
]

jobs = cron.batched_run(tasks)

List[str]

required

List of task strings to execute

Any

Additional parameters to pass to the agent’s run method

List[schedule.Job]

One scheduled job per task, in input order, returned once the schedule stops. Agent outputs are not collected; use a callback to capture them

Raises: the same exceptions as run(). Behavior: every task in tasks is registered on the job’s interval before blocking, so all of them run on each tick. This is one agent doing several things on one cadence.

Earlier versions scheduled only the first task: batched_run called run() per task, and run() blocked, so the loop never reached the second task. That is fixed — all tasks are scheduled, and the list of scheduled jobs is returned.

For several agents on different cadences, use run_many instead.

run_many

Run several agents together, each on its own interval. A class method. A CronJob binds one agent to one interval, so a fleet on mixed cadences needs one job per agent. run_many builds them, starts them all, and optionally blocks. Each job keeps its own scheduler thread, so the agents are isolated: one failing does not delay or stop the others, and each carries its own error budget.

from swarms import CronJob

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

List[Dict[str, Any]]

required

One mapping per agent.Required keys: agent, interval, taskOptional keys: job_id, callback, max_consecutive_errors, and kwargs (a dict forwarded to that agent’s run)

bool

default:"True"

Hold the calling thread until KeyboardInterrupt or until every job has stopped, then stop them all. Pass False to start the fleet and return immediately, leaving the caller responsible for stop_many.

List[CronJob]

The started jobs, in the order given, so they can be inspected via get_execution_stats() or stopped individually

Raises:

CronJobConfigError: If schedules is empty, or an entry is missing (or has an empty) agent, interval or task. The message names the index and the missing key. Entries are checked as they are started, so jobs for earlier entries are already running when a later entry fails.
CronJobExecutionError: If, once blocking ends, any job had stopped because it exhausted max_consecutive_errors

Non-blocking use:

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

# ... your own main loop ...

for job in jobs:
    print(job.get_execution_stats())

CronJob.stop_many(jobs)

stop_many

Stop every job in a list, continuing past any that fail to stop. A static method.

CronJob.stop_many(jobs)

List[CronJob]

required

The jobs to stop, typically the return value of run_many(..., block=False)

Behavior: one job refusing to stop is logged and does not strand the rest.

start

Start the background scheduler thread.

cron.start()

start() does not schedule a task. It only starts the thread that runs tasks already scheduled. Use run(), batched_run() or run_many() to schedule work; they call start() for you.

Raises:

CronJobExecutionError: If the job fails to start

Behavior:

Creates a daemon thread for job execution
Sets is_running to True
Records start_time
If already running, logs a warning

stop

Stop the scheduled job.

cron.stop()

Raises:

CronJobExecutionError: If the job fails to stop properly

Behavior:

Sets is_running to False
Waits up to 5 seconds for thread to terminate
Clears the schedule
Logs warning if thread doesn’t terminate gracefully

Safe to call from another thread while run() is blocking, and safe to call on a job that is already stopped. A clean stop() makes run() return normally.

set_callback

Set or update the callback function for output customization.

def custom_callback(output, task, metadata):
    # Process output
    processed = output.upper()
    
    # Log execution info
    print(f"Execution #{metadata['execution_count']}")
    print(f"Task: {task}")
    print(f"Timestamp: {metadata['timestamp']}")
    
    return processed

cron.set_callback(custom_callback)

Callable[[Any, str, dict], Any]

required

Callback function with signature: callback(output, task, metadata) -> Any

get_execution_stats

Get execution statistics for the cron job.

stats = cron.get_execution_stats()
print(stats)
# {
#   "job_id": "financial_analysis_job",
#   "is_running": True,
#   "execution_count": 42,          # successes
#   "start_time": 1234567890.123,
#   "uptime": 3600.5,
#   "interval": "10minutes",
#   "error_count": 3,               # total failures
#   "consecutive_errors": 0,        # since the last success
#   "last_error": "upstream API timed out",
#   "stopped_due_to_error": False   # True only if it gave up
# }

dict

Dictionary containing:

job_id: Job identifier
is_running: Current running status
execution_count: Number of successful executions
start_time: Start timestamp
uptime: Time elapsed since start (seconds)
interval: Configured interval string
error_count: Total failed executions
consecutive_errors: Failures since the last success
last_error: Most recent failure as a string, or None
stopped_due_to_error: True only when the job gave up after exhausting max_consecutive_errors

Interval Formats

The interval parameter accepts strings in the format <number><unit>:

Supported Units

Seconds: "5seconds", "30second"
Minutes: "10minutes", "1minute"
Hours: "2hours", "1hour"

Examples

# Every 5 seconds
CronJob(agent=agent, interval="5seconds")

# Every 30 seconds
CronJob(agent=agent, interval="30seconds")

# Every 10 minutes
CronJob(agent=agent, interval="10minutes")

# Every 2 hours
CronJob(agent=agent, interval="2hours")

Rejected at construction

These raise CronJobConfigError immediately rather than failing later:

Value	Why
`"0second"`, `"0minutes"`	A zero interval schedules a job that never fires
`""`, `" "`	Empty is a bad interval, not an absent one
`"-1second"`	Negative
`"1day"`	Unsupported unit; use seconds, minutes or hours
`"1 second"`	No space between the number and the unit
`"second"`	Missing the number

Complete Examples

Basic Scheduled Analysis

from swarms import Agent, CronJob

# Create financial analyst agent
agent = Agent(
    agent_name="Market-Analyst",
    system_prompt="""You are an expert market analyst.
    Provide concise daily market updates focusing on:
    1. Major index movements
    2. Notable sector performance
    3. Key market drivers
    """,
    model_name="gpt-5.4",
    max_loops=1,
    verbose=True
)

# Schedule to run every hour
cron = CronJob(
    agent=agent,
    interval="1hour",
    job_id="hourly_market_update"
)

# Run the scheduled job
try:
    cron.run(task="Provide a market update for the current hour")
except KeyboardInterrupt:
    print("Stopping scheduled job...")
    cron.stop()

Multi-Stock Analysis with Batching

from swarms import Agent, CronJob

stock_agent = Agent(
    agent_name="Stock-Analyzer",
    system_prompt="Analyze stock performance and provide investment insights.",
    model_name="gpt-5.4",
    max_loops=1
)

def print_analysis(output, task, metadata):
    print(f"\n{task}:\n{output}")
    return output

# Every task runs on each 30-minute tick, and the callback prints each result
cron = CronJob(
    agent=stock_agent,
    interval="30minutes",
    job_id="stock_analysis",
    callback=print_analysis,
)

stocks = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]
tasks = [f"Analyze {symbol} stock performance" for symbol in stocks]

# Blocks until Ctrl-C or stop()
cron.batched_run(tasks)

Custom Callback for Output Processing

from swarms import Agent, CronJob
import json
from datetime import datetime

def save_analysis_callback(output, task, metadata):
    """Save analysis results to file with metadata."""
    result = {
        "job_id": metadata["job_id"],
        "execution_number": metadata["execution_count"],
        "timestamp": datetime.fromtimestamp(metadata["timestamp"]).isoformat(),
        "task": task,
        "analysis": output,
        "uptime_seconds": metadata["timestamp"] - metadata["start_time"]
    }
    
    # Save to file
    filename = f"analysis_{metadata['execution_count']}.json"
    with open(filename, 'w') as f:
        json.dump(result, f, indent=2)
    
    print(f"Saved analysis #{metadata['execution_count']} to {filename}")
    return output

# Create agent
agent = Agent(
    agent_name="Crypto-Analyst",
    system_prompt="Analyze cryptocurrency market trends.",
    model_name="gpt-5.4",
    max_loops=1
)

# Create cron with callback
cron = CronJob(
    agent=agent,
    interval="15minutes",
    job_id="crypto_analysis",
    callback=save_analysis_callback
)

try:
    cron.run(task="Analyze Bitcoin and Ethereum market trends")
except KeyboardInterrupt:
    cron.stop()

Image Analysis with Scheduling

from swarms import Agent, CronJob

# Create vision-capable agent
vision_agent = Agent(
    agent_name="Chart-Analyzer",
    system_prompt="Analyze financial charts and provide technical insights.",
    model_name="claude-sonnet-4-6",  # Vision-capable model
    max_loops=1
)

# Schedule chart analysis every 2 hours
cron = CronJob(
    agent=vision_agent,
    interval="2hours",
    job_id="chart_analysis"
)

try:
    cron.run(
        task="Analyze this chart for support/resistance levels and trend direction",
        img="/path/to/chart.png"
    )
except KeyboardInterrupt:
    cron.stop()

Monitoring Execution Stats

run_many(..., block=False) schedules the task and returns at once, which leaves your thread free to poll the stats.

from swarms import Agent, CronJob
import time

agent = Agent(
    agent_name="Monitor-Agent",
    system_prompt="Monitor system metrics.",
    model_name="gpt-5.4",
    max_loops=1
)

jobs = CronJob.run_many(
    [{"agent": agent, "interval": "10seconds", "task": "Report current system metrics.", "job_id": "monitor"}],
    block=False,
)
cron = jobs[0]

try:
    while cron.is_running:
        time.sleep(30)
        stats = cron.get_execution_stats()
        print(f"\nExecution Stats:")
        print(f"  Running: {stats['is_running']}")
        print(f"  Executions: {stats['execution_count']}")
        print(f"  Failures in a row: {stats['consecutive_errors']}")
        print(f"  Uptime: {stats['uptime']:.2f}s")
except KeyboardInterrupt:
    print("Stopping monitor...")
finally:
    CronJob.stop_many(jobs)

Exception Handling

The CronJob module defines several custom exceptions. They are not exported from the top-level swarms package, so import them from swarms.structs.cron_job:

from swarms.structs.cron_job import (
    CronJobConfigError,
    CronJobError,
    CronJobExecutionError,
    CronJobScheduleError,
)

CronJobError

Base exception class for all CronJob errors.

CronJobConfigError

Raised at construction for a missing agent or a malformed, empty or zero interval, and by run_many for an empty or incomplete schedule list.

try:
    cron = CronJob(agent=None, interval="10minutes")
except CronJobConfigError as e:
    print(f"Configuration error: {e}")

CronJobScheduleError

Defined for scheduling errors, but CronJob does not currently raise it. An invalid interval raises CronJobConfigError:

try:
    cron = CronJob(agent=agent, interval="invalid_format")
except CronJobConfigError as e:
    print(f"Invalid interval: {e}")

CronJobExecutionError

Raised by run() and batched_run() when scheduling fails (for example, no interval was set) or when the job gives up after max_consecutive_errors failures in a row, and by start() and stop() when the thread cannot be started or stopped. A single failed execution does not raise; it is recorded in the error counters.

cron = CronJob(agent=agent, interval="1minute", max_consecutive_errors=5)

try:
    cron.run(task="Fetch the latest reading")
except CronJobExecutionError as e:
    print(f"Job gave up: {e}")

Best Practices

Use descriptive job IDs: Make job identifiers meaningful for tracking
Set appropriate intervals: Choose intervals based on task complexity and resource availability
Implement callbacks: Use callbacks for logging, saving results, or sending notifications
Monitor execution stats: Regularly check stats to ensure jobs are running as expected
Handle interrupts: run() catches KeyboardInterrupt, stops the job and returns, so Ctrl-C shuts a job down cleanly
Consider agent limits: Ensure your agent’s max_loops is appropriate for scheduled tasks
Set an error budget for flaky dependencies: max_consecutive_errors stops a job that is failing every time. Leave it as None when a task should retry indefinitely
Watch consecutive_errors, not just error_count: occasional failures on a long-running job are normal; a rising consecutive count is the signal something is actually broken
Use run_many for mixed cadences: one job per agent keeps them isolated, so a failing agent cannot hold up the others
Log failures: Enable verbose mode and implement proper error logging
Test intervals: Start with longer intervals and optimize based on performance
Resource management: Be mindful of API rate limits and costs with frequent scheduling
Graceful shutdown: Always call stop() when terminating scheduled jobs

Thread Safety

CronJob uses threading for background execution:

Daemon threads are used to prevent blocking program exit
Thread-safe scheduling with the schedule library
Proper cleanup on stop() with timeout handling

Callback Metadata

The callback function receives a metadata dictionary with:

{
    "job_id": str,              # Job identifier
    "timestamp": float,         # Unix timestamp of execution
    "execution_count": int,     # Successful executions, including this one
    "task": str,                # The task that was run
    "kwargs": dict,             # Additional kwargs passed
    "start_time": float,        # Job start timestamp
    "is_running": bool          # Current running status
}

The callback runs only after a successful execution. A failed execution skips it, and a callback that raises is logged and the original output is used. Use this metadata for:

Logging execution history
Conditional processing based on execution count
Time-based analysis
Debugging and monitoring

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

Constructor

Parameters

Attributes

Methods

run

batched_run

run_many

stop_many

start

stop

set_callback

get_execution_stats

Interval Formats

Supported Units

Examples

Rejected at construction

Complete Examples

Basic Scheduled Analysis

Multi-Stock Analysis with Batching

Custom Callback for Output Processing

Image Analysis with Scheduling

Monitoring Execution Stats

Exception Handling

CronJobError

CronJobConfigError

CronJobScheduleError

CronJobExecutionError

Best Practices

Thread Safety

Callback Metadata

​Overview

​Constructor

​Parameters

​Attributes

​Methods

​run

​batched_run

​run_many

​stop_many

​start

​stop

​set_callback

​get_execution_stats

​Interval Formats

​Supported Units

​Examples

​Rejected at construction

​Complete Examples

​Basic Scheduled Analysis

​Multi-Stock Analysis with Batching

​Custom Callback for Output Processing

​Image Analysis with Scheduling

​Monitoring Execution Stats

​Exception Handling

​CronJobError

​CronJobConfigError

​CronJobScheduleError

​CronJobExecutionError

​Best Practices

​Thread Safety

​Callback Metadata

Overview

Constructor

Parameters

Attributes

Methods

run

batched_run

run_many

stop_many

start

stop

set_callback

get_execution_stats

Interval Formats

Supported Units

Examples

Rejected at construction

Complete Examples

Basic Scheduled Analysis

Multi-Stock Analysis with Batching

Custom Callback for Output Processing

Image Analysis with Scheduling

Monitoring Execution Stats

Exception Handling

CronJobError

CronJobConfigError

CronJobScheduleError

CronJobExecutionError

Best Practices

Thread Safety

Callback Metadata