AI Agent for Store Operations: Employee Scheduling, Task Management, and Announcements

The Store Operations Challenge

Retail store managers spend hours each week on scheduling, task delegation, and team communication. An AI operations agent automates schedule generation based on employee availability and labor rules, assigns tasks with priority tracking, broadcasts announcements, and flags compliance issues before they become problems.

Employee and Availability Data Model

from agents import Agent, Runner, function_tool
from datetime import datetime, timedelta
from typing import Optional

EMPLOYEES = {
    "EMP-001": {
        "name": "Maria Santos",
        "role": "Sales Associate",
        "hourly_rate": 18.50,
        "max_hours_week": 40,
        "availability": {
            "monday": ("09:00", "17:00"),
            "tuesday": ("09:00", "17:00"),
            "wednesday": None,  # Day off
            "thursday": ("12:00", "20:00"),
            "friday": ("09:00", "17:00"),
            "saturday": ("10:00", "18:00"),
            "sunday": None,
        },
        "skills": ["register", "floor", "fitting_room"],
        "hours_this_week": 24,
    },
    "EMP-002": {
        "name": "James Park",
        "role": "Stock Associate",
        "hourly_rate": 17.00,
        "max_hours_week": 32,
        "availability": {
            "monday": ("06:00", "14:00"),
            "tuesday": ("06:00", "14:00"),
            "wednesday": ("06:00", "14:00"),
            "thursday": ("06:00", "14:00"),
            "friday": None,
            "saturday": ("08:00", "16:00"),
            "sunday": ("08:00", "14:00"),
        },
        "skills": ["receiving", "stocking", "inventory"],
        "hours_this_week": 16,
    },
    "EMP-003": {
        "name": "Priya Nair",
        "role": "Keyholder",
        "hourly_rate": 21.00,
        "max_hours_week": 40,
        "availability": {
            "monday": ("08:00", "16:00"),
            "tuesday": None,
            "wednesday": ("08:00", "16:00"),
            "thursday": ("08:00", "16:00"),
            "friday": ("08:00", "16:00"),
            "saturday": ("09:00", "17:00"),
            "sunday": ("10:00", "16:00"),
        },
        "skills": ["register", "floor", "opening", "closing", "training"],
        "hours_this_week": 32,
    },
}

LABOR_RULES = {
    "min_break_after_hours": 5,  # Break required after 5 hours
    "break_duration_min": 30,
    "min_hours_between_shifts": 10,
    "max_consecutive_days": 6,
    "overtime_threshold": 40,
    "minor_max_hours": 20,  # Under 18
}

Schedule Generation

@function_tool
def generate_schedule(day: str, required_roles: str) -> str:
    """Generate a shift schedule for a given day based on required roles."""
    day_lower = day.lower()
    roles_needed = [r.strip() for r in required_roles.split(",")]

    available = []
    for emp_id, emp in EMPLOYEES.items():
        avail = emp["availability"].get(day_lower)
        if avail is None:
            continue
        remaining_hours = emp["max_hours_week"] - emp["hours_this_week"]
        if remaining_hours <= 0:
            continue
        available.append({
            "id": emp_id,
            "name": emp["name"],
            "role": emp["role"],
            "shift": f"{avail[0]}-{avail[1]}",
            "skills": emp["skills"],
            "remaining_hours": remaining_hours,
        })

    if not available:
        return f"No employees available on {day}."

    schedule_lines = [f"Schedule for {day.capitalize()}:"]
    assigned = set()
    for role in roles_needed:
        matched = None
        for emp in available:
            if emp["id"] in assigned:
                continue
            if role.lower() in emp["role"].lower() or \
               role.lower() in [s.lower() for s in emp["skills"]]:
                matched = emp
                break
        if matched:
            assigned.add(matched["id"])
            schedule_lines.append(
                f"  {matched['shift']}: {matched['name']} "
                f"({matched['role']}) - {role}"
            )
        else:
            schedule_lines.append(
                f"  UNSTAFFED: {role} - no available employee"
            )

    return "\n".join(schedule_lines)

@function_tool
def check_compliance(employee_id: str) -> str:
    """Check labor compliance for an employee."""
    emp = EMPLOYEES.get(employee_id)
    if not emp:
        return "Employee not found."

    issues = []
    hours = emp["hours_this_week"]
    max_h = emp["max_hours_week"]

    if hours >= LABOR_RULES["overtime_threshold"]:
        issues.append(
            f"OVERTIME: {emp['name']} has {hours}h this week "
            f"(threshold: {LABOR_RULES['overtime_threshold']}h)"
        )
    elif hours >= max_h - 4:
        issues.append(
            f"APPROACHING LIMIT: {emp['name']} at {hours}/{max_h}h "
            f"this week"
        )

    if not issues:
        return (
            f"{emp['name']}: No compliance issues. "
            f"{hours}/{max_h} hours scheduled this week."
        )
    return f"Compliance alerts for {emp['name']}:\n" + \
           "\n".join(f"  - {i}" for i in issues)

Task Management

TASKS = []

@function_tool
def create_task(title: str, assigned_to: str, priority: str,
                due_by: str, description: str = "") -> str:
    """Create and assign a task to an employee."""
    emp = EMPLOYEES.get(assigned_to)
    if not emp:
        return "Employee not found."

    task = {
        "task_id": f"TASK-{len(TASKS) + 1:04d}",
        "title": title,
        "assigned_to": assigned_to,
        "assigned_name": emp["name"],
        "priority": priority,
        "due_by": due_by,
        "description": description,
        "status": "pending",
        "created_at": datetime.now().isoformat(),
    }
    TASKS.append(task)
    return (
        f"Task {task['task_id']} created: '{title}' assigned to "
        f"{emp['name']} (priority: {priority}, due: {due_by})"
    )

@function_tool
def get_task_status(filter_by: str = "all") -> str:
    """Get task list, optionally filtered by status or employee ID."""
    filtered = TASKS
    if filter_by != "all":
        filtered = [
            t for t in TASKS
            if t["status"] == filter_by or t["assigned_to"] == filter_by
        ]

    if not filtered:
        return "No tasks found."

    lines = ["Task Board:"]
    for t in filtered:
        lines.append(
            f"  {t['task_id']}: {t['title']} | "
            f"{t['assigned_name']} | {t['priority']} | "
            f"{t['status']} | Due: {t['due_by']}"
        )
    return "\n".join(lines)

@function_tool
def update_task_status(task_id: str, new_status: str) -> str:
    """Update a task's status (pending, in_progress, completed)."""
    task = next((t for t in TASKS if t["task_id"] == task_id), None)
    if not task:
        return "Task not found."
    old_status = task["status"]
    task["status"] = new_status
    return (
        f"Task {task_id} updated: {old_status} -> {new_status}"
    )

Announcements

ANNOUNCEMENTS = []

@function_tool
def broadcast_announcement(title: str, message: str,
                           priority: str = "normal") -> str:
    """Broadcast an announcement to all store employees."""
    announcement = {
        "id": f"ANN-{len(ANNOUNCEMENTS) + 1:04d}",
        "title": title,
        "message": message,
        "priority": priority,
        "timestamp": datetime.now().isoformat(),
        "acknowledged_by": [],
    }
    ANNOUNCEMENTS.append(announcement)
    recipient_count = len(EMPLOYEES)
    return (
        f"Announcement {announcement['id']} broadcast to "
        f"{recipient_count} employees: '{title}' "
        f"(priority: {priority})"
    )

Assembling the Operations Agent

ops_agent = Agent(
    name="Store Operations Manager",
    instructions="""You assist retail store managers with daily operations.

    Capabilities:
    - Generate employee schedules based on availability and roles needed
    - Check labor compliance before finalizing schedules
    - Create and track tasks with priority levels
    - Broadcast announcements to the team
    - Flag staffing gaps and overtime risks

    Always check compliance before confirming any schedule.
    Prioritize tasks by urgency and employee workload.
    Flag any unstaffed shifts immediately.""",
    tools=[generate_schedule, check_compliance, create_task,
           get_task_status, update_task_status,
           broadcast_announcement],
)

FAQ

How do I integrate with existing workforce management systems?

Most WFM platforms like Kronos (UKG), ADP, or Deputy expose REST APIs. Build adapter tools that wrap these APIs: one for pulling employee availability, one for submitting generated schedules, and one for syncing time-clock data. The agent generates candidate schedules using the tools above, and a separate integration layer pushes approved schedules into the WFM system.

How should the agent handle shift swap requests?

Create a swap request tool that takes the requesting employee, the shift to swap, and an optional preferred replacement. The agent checks availability and compliance for both employees, proposes a swap if valid, and requires manager approval before finalizing. Track swap history to identify patterns — frequent swaps on specific days may indicate a scheduling pattern issue.

What compliance rules should I encode for different jurisdictions?

Labor laws vary significantly. At minimum, encode maximum weekly hours, mandatory break intervals, minimum rest between shifts, overtime thresholds, and minor-specific restrictions. For predictive scheduling jurisdictions like Oregon, San Francisco, and New York City, also encode advance notice requirements (typically 14 days) and premium pay for last-minute changes. Store these rules as configuration per location rather than hardcoding them.

---

#StoreOperations #EmployeeScheduling #TaskManagement #RetailManagement #LaborCompliance #AgenticAI #LearnAI #AIEngineering