Scheduler Configuration

Overview

The Scheduler feature enables you to configure Digital Employees to execute tasks automatically at specified times using cron-based scheduling. This is ideal for recurring tasks, automated reports, periodic checks, and time-based workflows.

Currently we utilize AIP Scheduled Run feature to manage these workflows.

Note: Schedules are only available for remote runs (deployed Digital Employees). Local runs do not support scheduled execution.

Key Concepts

What is a Scheduler?

A scheduler allows your Digital Employee to run tasks automatically based on time-based triggers. Each schedule consists of:

• Schedule Configuration: Defines when the task should run (using cron syntax)

• Input: The message or instruction to execute when the schedule triggers

Benefits

• Automation: Execute tasks without manual intervention

• Consistency: Ensure tasks run at predictable times

• Efficiency: Free up human resources for higher-value work

• Reliability: Never miss scheduled tasks or reminders

Configuration

Basic Components

Schedules are configured using two main classes:

1. ScheduleConfig: Defines the timing using cron-like parameters

2. ScheduleItemConfig: Combines the schedule with the input to execute

Schedule Configuration Parameters

The ScheduleConfig class uses cron-style parameters:

from glaip_sdk.models.schedule import ScheduleConfig

schedule = ScheduleConfig(
    minute="0",           # 0-59 or "*" for every minute
    hour="8",             # 0-23 or "*" for every hour
    day_of_month="*",     # 1-31 or "*" for every day
    month="*",            # 1-12 or "*" for every month
    day_of_week="0-4",    # 0-6 (0=Monday, 6=Sunday) or "*" for every day
)

Reference to GLAIP Python SDK.

Cron Syntax Guide

Field	Values	Special Characters	Examples
minute	0-59	* (every), - (range), , (list)	0, */15, 0,30
hour	0-23	* (every), - (range), , (list)	8, 9-17, 8,12,18
day_of_month	1-31	* (every), - (range), , (list)	1, 1-15, 1,15
month	1-12	* (every), - (range), , (list)	*, 1-6, 1,7
day_of_week	0-6	* (every), - (range), , (list)	0-4, 0,6, *

Note: Day of week starts with Monday (0) and ends with Sunday (6).

Complete Example

Step 1: Import Required Components

from glaip_sdk.models.schedule import ScheduleConfig
from digital_employee_core import (
    DigitalEmployee,
    DigitalEmployeeIdentity,
    DigitalEmployeeJob,
)
from digital_employee_core.schedule import ScheduleItemConfig

Step 2: Create Digital Employee Identity

# Define the job
job = DigitalEmployeeJob(
    title="Friendly Greeter",
    description="A friendly digital employee that greets people at different times",
    instruction="You are a friendly greeter. Always respond with a cheerful greeting!",
)

# Create identity
identity = DigitalEmployeeIdentity(
    name="Sunny",
    email="[e-mail]",
    job=job,
)

Step 3: Configure Schedule

# Morning greeting - weekdays at 8 AM
morning_schedule = ScheduleConfig(
    minute="0",
    hour="8",
    day_of_month="*",
    month="*",
    day_of_week="0-4",  # Monday to Friday
)

Step 4: Create Schedule Item

# Create schedule item with input
morning_schedule_item = ScheduleItemConfig(
    schedule_config=morning_schedule,
    input="Morning greeting"
)

Step 5: Create and Deploy Digital Employee

# Create Digital Employee with schedule
digital_employee = DigitalEmployee(
    identity=identity,
    schedules=[morning_schedule_item],
)

# Deploy the Digital Employee (this also creates the schedule)
digital_employee.deploy()

Step 6: Verify Schedules (Optional)

# Get schedules from the Digital Employee
schedules = digital_employee.get_schedule()
print(f"Number of schedules: {len(schedules)}")

# Display schedule details
for i, schedule_item in enumerate(schedules, 1):
    print(f"Schedule {i}:")
    print(f"  Input: {schedule_item.input}")
    print(f"  Cron: {schedule_item.schedule_config.to_cron_string()}")

Step 7: Monitor Schedule Runs (Optional)

from glaip_sdk import Client

# Get the deployed agent
client = Client()
agent = client.get_agent_by_id(digital_employee.agent.id)

# List all schedules
agent_schedules = agent.schedule.list()
print(f"Found {len(agent_schedules)} schedule(s)")

# Check runs for a specific schedule
for schedule in agent_schedules:
    runs = agent.schedule.list_runs(schedule.id)
    print(f"\nSchedule: {schedule.input}")
    print(f"Runs: {len(runs)}")
    
    # Display recent runs
    for run in runs[-5:]:  # Last 5 runs
        print(f"  - Run ID: {run.id}")
        print(f"    Status: {run.status}")
        if run.status == "success":
            result = run.get_result()
            print(f"    Result: {result}")

Common Use Cases

Daily Morning Report

Send a daily report every weekday at 9 AM:

daily_report_schedule = ScheduleConfig(
    minute="0",
    hour="9",
    day_of_month="*",
    month="*",
    day_of_week="0-4",  # Monday to Friday
)

report_item = ScheduleItemConfig(
    schedule_config=daily_report_schedule,
    input="Generate and send the daily morning report"
)

Hourly Data Check

Check data every hour during business hours:

hourly_check_schedule = ScheduleConfig(
    minute="0",
    hour="9-17",  # 9 AM to 5 PM
    day_of_month="*",
    month="*",
    day_of_week="0-4",  # Monday to Friday
)

check_item = ScheduleItemConfig(
    schedule_config=hourly_check_schedule,
    input="Check system status and alert if issues found"
)

Weekly Summary

Generate a weekly summary every Friday at 5 PM:

weekly_summary_schedule = ScheduleConfig(
    minute="0",
    hour="17",
    day_of_month="*",
    month="*",
    day_of_week="4",  # Friday
)

summary_item = ScheduleItemConfig(
    schedule_config=weekly_summary_schedule,
    input="Generate weekly summary report"
)

Monthly Reminder

Send a reminder on the first day of each month:

monthly_reminder_schedule = ScheduleConfig(
    minute="0",
    hour="9",
    day_of_month="1",  # First day of month
    month="*",
    day_of_week="*",
)

reminder_item = ScheduleItemConfig(
    schedule_config=monthly_reminder_schedule,
    input="Send monthly reminder to review pending tasks"
)

Every 15 Minutes

Run a task every 15 minutes:

frequent_check_schedule = ScheduleConfig(
    minute="*/15",  # Every 15 minutes
    hour="*",
    day_of_month="*",
    month="*",
    day_of_week="*",
)

frequent_item = ScheduleItemConfig(
    schedule_config=frequent_check_schedule,
    input="Check for urgent notifications"
)

Best Practices

1. Use Descriptive Inputs

Provide clear, actionable instructions in the schedule input:

# Good: Specific and actionable
ScheduleItemConfig(
    schedule_config=schedule,
    input="Review unread emails from VIP customers and respond to urgent ones"
)

# Avoid: Vague or unclear
ScheduleItemConfig(
    schedule_config=schedule,
    input="Check emails"
)

2. Consider Time Zones

Be aware of the time zone used by your deployment:

# Document the time zone in comments
# Schedule runs at 9 AM UTC
morning_schedule = ScheduleConfig(
    minute="0",
    hour="9",
    day_of_month="*",
    month="*",
    day_of_week="0-4",
)

3. Avoid Overlapping Schedules

Ensure schedules don't conflict or create excessive load:

# Good: Staggered schedules
schedule_1 = ScheduleConfig(minute="0", hour="9", ...)  # 9:00 AM
schedule_2 = ScheduleConfig(minute="30", hour="9", ...)  # 9:30 AM

# Avoid: Multiple schedules at the same time
schedule_1 = ScheduleConfig(minute="0", hour="9", ...)  # 9:00 AM
schedule_2 = ScheduleConfig(minute="0", hour="9", ...)  # 9:00 AM (conflict)

4. Test Schedule Timing

Verify your cron expressions produce the expected schedule:

schedule = ScheduleConfig(
    minute="0",
    hour="9",
    day_of_month="*",
    month="*",
    day_of_week="0-4",
)

# Check the cron string
cron_string = schedule.to_cron_string()
print(f"Cron expression: {cron_string}")
# Output: "0 9 * * 0-4"

5. Monitor Schedule Execution

Regularly check schedule runs to ensure they're executing as expected:

# Check recent failed runs
failed_runs = agent.schedule.list_runs(schedule_id, status="failed")

if failed_runs:
    print(f"Warning: {len(failed_runs)} failed runs detected")

6. Use Multiple Schedules Strategically

Group related schedules in a single Digital Employee:

# Good: Related schedules for a single purpose
digital_employee = DigitalEmployee(
    identity=identity,
    schedules=[
        morning_report_item,    # Daily morning report
        afternoon_check_item,   # Afternoon status check
        evening_summary_item,   # Evening summary
    ],
)

# Avoid: Unrelated schedules in one employee
# Consider creating separate Digital Employees for different purposes

Advanced Configuration

Dynamic Schedule Inputs

Use detailed inputs to provide context:

schedule_item = ScheduleItemConfig(
    schedule_config=schedule,
    input="""
    Generate a daily report including:
    1. Summary of completed tasks
    2. Pending items requiring attention
    3. Any system alerts or issues
    4. Send the report to the team channel
    """
)

Combining with Other Features

Schedules work seamlessly with other Digital Employee features:

from digital_employee_core.connectors.mcps import google_mail_mcp

# Digital Employee with schedules and MCP tools
digital_employee = DigitalEmployee(
    identity=identity,
    schedules=[email_check_schedule_item],
    mcps=[google_mail_mcp],  # Can use email tools in scheduled tasks
    configurations=configurations,
)

Troubleshooting

Schedule Not Triggering

Problem: Schedule was created but tasks are not running.

Solution:

• Verify the Digital Employee is deployed: digital_employee.deploy()

• Check the cron expression is valid: schedule.to_cron_string()

• Confirm the schedule exists: agent.schedule.list()

• Check for failed runs: agent.schedule.list_runs(schedule_id)

Incorrect Timing

Problem: Schedule runs at unexpected times.

Solution:

• Verify time zone settings

• Double-check cron parameters (especially day_of_week where 0=Monday)

• Test with to_cron_string() to see the actual cron expression

• Review the next_run_time field in the schedule object

Schedule Runs Failed

Problem: Schedule triggers but execution fails.

Solution:

• Check run details: run.get_result() for error messages

• Verify the input instruction is clear and actionable

• Ensure required tools/MCPs are configured and accessible

• Review Digital Employee logs for detailed error information

Multiple Schedules Conflict

Problem: Multiple schedules running simultaneously cause issues.

Solution:

• Stagger schedule times by adjusting minute/hour values

• Consider if schedules can be combined into a single task

• Monitor system resources and adjust frequency if needed

API Reference

ScheduleConfig

from glaip_sdk.models.schedule import ScheduleConfig

schedule = ScheduleConfig(
    minute: str,        # "0-59", "*", "*/15", "0,30"
    hour: str,          # "0-23", "*", "9-17", "8,12,18"
    day_of_month: str,  # "1-31", "*", "1,15"
    month: str,         # "1-12", "*", "1-6"
    day_of_week: str,   # "0-6", "*", "0-4", "0,6"
)

# Convert to cron string
cron_string = schedule.to_cron_string()

ScheduleItemConfig

from digital_employee_core.schedule import ScheduleItemConfig

schedule_item = ScheduleItemConfig(
    schedule_config: ScheduleConfig,  # The timing configuration
    input: str,                       # The instruction to execute
)

DigitalEmployee with Schedules

from digital_employee_core import DigitalEmployee

digital_employee = DigitalEmployee(
    identity: DigitalEmployeeIdentity,
    schedules: list[ScheduleItemConfig] = [],  # Optional list of schedules
    mcps: list = [],                           # Optional MCP connectors
    configurations: list = [],                 # Optional configurations
)

# Deploy to activate schedules
digital_employee.deploy()

# Get configured schedules
schedules = digital_employee.get_schedule()

Managing Schedules via Agent

from glaip_sdk import Client

client = Client()
agent = client.get_agent_by_id(agent_id)

# List all schedules
schedules = agent.schedule.list()

# Get runs for a schedule
runs = agent.schedule.list_runs(schedule_id)

# Access run details
for run in runs:
    print(f"Status: {run.status}")
    print(f"ID: {run.id}")
    if run.status == "success":
        result = run.get_result()

Examples in Repository

For complete working examples, see:

• https://github.com/GDP-ADMIN/CATAPA-SDK/blob/main/python/digital-employee-core/examples/scheduler_usage.py - Comprehensive scheduler demonstration