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 and retried on the next tick, the way cron behaves. It does not take the schedule down. 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.structs.cron_job import CronJob
from swarms import Agent

# Create an agent
agent = Agent(
    agent_name="Financial-Analyst",
    system_prompt="You are a financial analyst...",
    model_name="gpt-4"
)

# 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

str

default:"None"

The interval string (e.g., “5seconds”, “10minutes”, “1hour”)

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 times the job has been executed

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)

Raises:

CronJobConfigError: If agent or interval is not configured
CronJobExecutionError: If scheduling failed, 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
A task that raises is logged and retried on the next tick; it does not stop the schedule

batched_run

Run multiple tasks sequentially with the same schedule.

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

results = 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[Any]

List of results from each task execution

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.structs.cron_job 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 agent, interval or task. The message names the index and the missing key.
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

Manually start the scheduled job.

cron.start()

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

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 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
from swarms.structs.cron_job import 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-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
from swarms.structs.cron_job import CronJob

# Create stock analysis agent
stock_agent = Agent(
    agent_name="Stock-Analyzer",
    system_prompt="Analyze stock performance and provide investment insights.",
    model_name="gpt-4",
    max_loops=2
)

# Schedule every 30 minutes
cron = CronJob(
    agent=stock_agent,
    interval="30minutes",
    job_id="stock_analysis"
)

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

try:
    results = cron.batched_run(tasks)
    for symbol, result in zip(stocks, results):
        print(f"\n{symbol} Analysis:")
        print(result)
except KeyboardInterrupt:
    print("Analysis stopped.")
    cron.stop()

Custom Callback for Output Processing

from swarms import Agent
from swarms.structs.cron_job import 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.get("start_time", 0)
    }
    
    # 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-4"
)

# 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
from swarms.structs.cron_job import 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

from swarms import Agent
from swarms.structs.cron_job import CronJob
import time

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

cron = CronJob(
    agent=agent,
    interval="10seconds",
    job_id="monitor"
)

# Start the job in background
cron.start()

try:
    # Monitor stats every 30 seconds
    while True:
        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"  Uptime: {stats['uptime']:.2f}s")
except KeyboardInterrupt:
    print("Stopping monitor...")
    cron.stop()

Exception Handling

The CronJob class defines several custom exceptions:

CronJobError

Base exception class for all CronJob errors.

CronJobConfigError

Raised for configuration errors.

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

CronJobScheduleError

Raised for scheduling related errors.

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

CronJobExecutionError

Raised for execution related errors.

try:
    cron.run(task="")
except CronJobExecutionError as e:
    print(f"Execution failed: {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: Always wrap run() in try-except to handle KeyboardInterrupt
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,     # Number of times executed
    "task": str,                # The task that was run
    "kwargs": dict,             # Additional kwargs passed
    "start_time": float,        # Job start timestamp
    "is_running": bool          # Current running status
}

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