How to Build an AI Agent for Scheduling and Time Management

from datetime import datetime, timezone

def read_busy(cal, attendees: list[str], start, end) -> dict:
    """Return each attendee's busy blocks, normalised to UTC."""
    resp = cal.freebusy().query(body={
        "timeMin": start.isoformat(), "timeMax": end.isoformat(),
        "items": [{"id": a} for a in attendees]}).execute()
    out = {}
    for a, cal_data in resp["calendars"].items():
        out[a] = [(_utc(b["start"]), _utc(b["end"]))    # always store in UTC
                  for b in cal_data["busy"]]
    return out

def _utc(iso: str) -> datetime:
    return datetime.fromisoformat(iso).astimezone(timezone.utc)
# Microsoft Graph: POST /me/calendar/getSchedule returns the same idea.
# Everything downstream compares in UTC; local time is only for display.

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class Request(BaseModel):
    attendees: list[str]; duration_min: int
    window_start: str; window_end: str          # ISO dates
    timezone: str; constraints: str | None = None

def parse_request(text: str) -> Request:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Extract the scheduling request. Today is 2026-07-01.\n{text}"],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Request))
    return res.parsed                            # a typed Request object

from datetime import timedelta
from zoneinfo import ZoneInfo

def free_slots(busy: dict, window, duration_min: int, zones: dict) -> list[dict]:
    step = timedelta(minutes=15); need = timedelta(minutes=duration_min)
    slot, out = window[0], []                    # window is (start, end) in UTC
    while slot + need <= window[1]:
        end = slot + need
        clash = any(b0 < end and slot < b1       # overlap test, all in UTC
                    for blocks in busy.values() for b0, b1 in blocks)
        if not clash:
            out.append({"utc": slot, "local": {              # render per attendee
                a: slot.astimezone(ZoneInfo(z)).strftime("%a %H:%M")
                for a, z in zones.items()}})
            slot = end
        else:
            slot += step
    return out
# Intersection happens in UTC; the only place local time appears is the label.

from google import genai
client = genai.Client()
REPLY = ["accept", "decline", "reschedule", "other"]

def propose(slots: list, attendees: str, tone: str = "warm and brief") -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Write a {tone} message offering these times to {attendees}. "
                  f"Ask them to pick one.\n{slots}"]).text

def classify_reply(text: str) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Classify this reply into one of {REPLY}. Label only.\n{text}"]
        ).text.strip()

def book(cal, slot: dict, attendees: list[str], req_id: str) -> dict:
    key = idem_key(req_id, slot["utc"], attendees)   # attendees + slot + request
    if store.booked(key):                            # idempotent: never twice
        return store.event(key)
    external = [a for a in attendees if not internal(a)]
    if external and not approved(req_id):            # invite leaves the domain
        approval_queue.add(req_id, slot, external)   # wait for a person
        return {"status": "awaiting_approval"}
    event = cal.events().insert(calendarId="primary", body={
        "start": {"dateTime": slot["utc"].isoformat(), "timeZone": "UTC"},
        "end":   {"dateTime": slot["end"].isoformat(), "timeZone": "UTC"},
        "attendees": [{"email": a} for a in attendees]}).execute()
    store.save(key, event); audit.log("booked", req_id, event["id"])
    return event

from datetime import time

FOCUS = (time(9, 0), time(11, 0))      # daily deep-work block, local time
MAX_MEETINGS = 5                       # cap per day

def allowed(slot_local, meetings_today: int) -> bool:
    t = slot_local.time()
    if FOCUS[0] <= t < FOCUS[1]:                 # never book over deep work
        return False
    if meetings_today >= MAX_MEETINGS:           # cap the day
        return False
    return True

def filter_slots(slots: list, zones: dict, counts: dict) -> list:
    owner, z = next(iter(zones.items()))         # the calendar being protected
    return [s for s in slots
            if allowed(s["utc"].astimezone(ZoneInfo(z)), counts[owner])]
# Focus time is checked in the owner's local zone, where 9 to 11 means 9 to 11.

def conflicts(events: list[dict]) -> list[tuple]:
    """Return overlapping pairs. All times compared in UTC."""
    ev = sorted(events, key=lambda e: e["start"])     # start, end are UTC
    clashes = []
    for i in range(len(ev) - 1):
        a, b = ev[i], ev[i + 1]
        if b["start"] < a["end"]:                      # they overlap
            clashes.append((a, b))
    return clashes

def resolve(a: dict, b: dict):
    loser = a if a["status"] == "tentative" else b     # accepted beats tentative
    if a["status"] == b["status"]:
        return review_queue.add((a, b), "double_booked")  # a person decides
    cal.events().delete(calendarId="primary",
                        eventId=loser["id"]).execute()
    audit.log("conflict_resolved", loser["id"])

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

scheduling_pipeline = SequentialAgent(name="scheduling_pipeline",
    sub_agents=[parser, availability, proposer])  # each an LlmAgent with the tools
runner = Runner(agent=scheduling_pipeline, app_name="scheduling",
                session_service=InMemorySessionService())

def guard(req: dict, slot: dict, attendees: list) -> str | None:
    if store.booked(idem_key(req["id"], slot["utc"], attendees)):
        return "already_booked"                  # idempotent: never double-book
    if any(not internal(a) for a in attendees) and not approved(req["id"]):
        return "needs_approval"                  # external invite waits for a person
    return None
# Run on each request / reply via webhook or cron. UTC throughout; local only to show.