handbook/docker/packages/google_calendar/google_calendar/server.py

"""FastMCP tool surface for the Google Calendar mock."""

import functools
import inspect

from fastmcp import FastMCP
from pydantic import EmailStr

from .models import (
    AttendeeResponseStatus,
    AvailabilityDurationMinutes,
    BirthdayProperties,
    CalendarAccountId,
    CalendarAccountsState,
    CalendarId,
    CalendarPerson,
    CalendarState,
    CalendarSummary,
    CalendarTimeZone,
    EventAttendee,
    EventClearField,
    EventDateTime,
    EventId,
    EventReminders,
    EventSource,
    EventSummary,
    EventTransparency,
    EventType,
    ExtendedProperties,
    FocusTimeProperties,
    GoogleCalendarState,
    ListEventsMaxResults,
    OutOfOfficeProperties,
    RecurrenceRuleString,
    Rfc3339OffsetDateTimeString,
    SearchEventsMaxResults,
    SearchEventsOrderBy,
    SearchEventsResponse,
    WorkingLocationProperties,
)
from .state import list_accounts as list_calendar_accounts
from .state import set_active_account, state_from_json, state_to_json, write_snapshots
from .tools import availability, calendars, events, search

mcp = FastMCP("google_calendar")


def _snapshot_on_write(fn):
    """Write configured state snapshots after successful write tools.

    The wrapper is async so the registered tool is a coroutine: FastMCP then
    validates and runs it on the event loop instead of a worker threadpool,
    avoiding the pydantic-core concurrency panic. See ``async_tool_guard``.
    """

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        result = fn(*args, **kwargs)
        if inspect.isawaitable(result):
            result = await result
        write_snapshots()
        return result

    return wrapper


def _with_account(fn):
    """Decorator: extract account_id from tool calls and select that account."""

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        account_id = kwargs.pop("account_id", None)
        if account_id is not None:
            set_active_account(account_id)
        result = fn(*args, **kwargs)
        if inspect.isawaitable(result):
            result = await result
        return result

    return wrapper


@mcp.tool()
async def export_state() -> GoogleCalendarState:
    """Export the full google_calendar state as JSON.

    Round-trips with import_state.
    """
    state = state_to_json()
    if "accounts" in state:
        return CalendarAccountsState.model_validate(state)
    return CalendarState.model_validate(state)


@mcp.tool()
@_snapshot_on_write
def import_state(state: GoogleCalendarState) -> dict:
    """Replace the google_calendar state with the provided JSON.

    For synthetic-data injection and test setup. Round-trips with export_state.
    """
    state_from_json(state)
    return {"status": "success"}


@mcp.tool()
async def list_accounts() -> dict:
    """List available isolated Google Calendar accounts.

    Multi-account worlds store each account as a fully separate calendar state.
    Use the returned ``account_id`` with other tools to select the account to
    read or mutate. Single-account worlds expose their only configured account.
    """
    accounts = list_calendar_accounts()
    return {"status": "success", "accounts": accounts, "count": len(accounts)}


@mcp.tool()
@_with_account
@_snapshot_on_write
def create_event(
    summary: EventSummary,
    start: EventDateTime,
    end: EventDateTime,
    description: str | None = None,
    location: str | None = None,
    calendar_id: CalendarId = "primary",
    recurrence: list[RecurrenceRuleString] | None = None,
    reminders: EventReminders | None = None,
    attendees: list[EventAttendee] | None = None,
    creator: CalendarPerson | None = None,
    organizer: CalendarPerson | None = None,
    extendedProperties: ExtendedProperties | None = None,
    source: EventSource | None = None,
    transparency: EventTransparency | None = None,
    eventType: EventType = EventType.DEFAULT,
    outOfOfficeProperties: OutOfOfficeProperties | None = None,
    focusTimeProperties: FocusTimeProperties | None = None,
    workingLocationProperties: WorkingLocationProperties | None = None,
    birthdayProperties: BirthdayProperties | None = None,
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Creates a new event in Google Calendar.

    Args:
        summary: Event title
        start: Start time object with either 'dateTime' (ISO format, e.g. '2025-12-10T09:00:00-05:00')
               for timed events, or 'date' (YYYY-MM-DD, e.g. '2025-12-10') for all-day events.
               Optionally include 'timeZone' (e.g. 'America/New_York').
        end: End time object with either 'dateTime' or 'date' (same format as start).
             For all-day events, 'date' is exclusive (e.g. end '2025-12-11' means the event ends on 2025-12-10).
        description: Event description
        location: Event location
        calendar_id: Calendar to create the event in (default 'primary')
        recurrence: RRULE strings for recurring events (e.g., ['RRULE:FREQ=WEEKLY;BYDAY=TU;COUNT=10'])
        reminders: Reminder overrides, e.g. {"useDefault": false, "overrides": [{"method": "popup", "minutes": 15}]}
        attendees: List of attendees, e.g. [{"email": "alice@co.com", "displayName": "Alice"}]. responseStatus defaults to 'needsAction'.
        transparency: Whether the event blocks availability. 'opaque' is busy; 'transparent' is free.
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        The created event object with its ID
    """
    return events.create_event(
        summary=summary,
        start=start,
        end=end,
        description=description,
        location=location,
        calendar_id=calendar_id,
        recurrence=recurrence,
        reminders=reminders,
        attendees=attendees,
        creator=creator,
        organizer=organizer,
        extendedProperties=extendedProperties,
        source=source,
        transparency=transparency,
        eventType=eventType,
        outOfOfficeProperties=outOfOfficeProperties,
        focusTimeProperties=focusTimeProperties,
        workingLocationProperties=workingLocationProperties,
        birthdayProperties=birthdayProperties,
    )


@mcp.tool()
@_with_account
def get_event(
    eventId: EventId,
    calendar_id: CalendarId = "primary",
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Retrieves details of a specific event.

    Args:
        eventId: ID of the event to retrieve
        calendar_id: Calendar to look in (default 'primary')
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        The event object if found
    """
    return events.get_event(eventId=eventId, calendar_id=calendar_id)


@mcp.tool()
@_with_account
@_snapshot_on_write
def update_event(
    eventId: EventId,
    summary: EventSummary | None = None,
    start: EventDateTime | None = None,
    end: EventDateTime | None = None,
    description: str | None = None,
    location: str | None = None,
    calendar_id: CalendarId = "primary",
    recurrence: list[RecurrenceRuleString] | None = None,
    reminders: EventReminders | None = None,
    attendees: list[EventAttendee] | None = None,
    creator: CalendarPerson | None = None,
    organizer: CalendarPerson | None = None,
    extendedProperties: ExtendedProperties | None = None,
    source: EventSource | None = None,
    transparency: EventTransparency | None = None,
    eventType: EventType | None = None,
    outOfOfficeProperties: OutOfOfficeProperties | None = None,
    focusTimeProperties: FocusTimeProperties | None = None,
    workingLocationProperties: WorkingLocationProperties | None = None,
    birthdayProperties: BirthdayProperties | None = None,
    clear_fields: list[EventClearField] | None = None,
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Updates an existing event.

    Args:
        eventId: ID of the event to update
        summary: New event title
        start: New start time object with either 'dateTime' (ISO format) for timed events,
               or 'date' (YYYY-MM-DD) for all-day events. Optionally include 'timeZone'.
        end: New end time object with either 'dateTime' or 'date' (same format as start).
        description: New event description
        location: New event location
        calendar_id: Calendar containing the event (default 'primary')
        recurrence: RRULE strings for recurring events (pass empty list to remove recurrence)
        reminders: Reminder overrides (pass {"useDefault": true} to reset to defaults)
        attendees: Replace attendee list (pass empty list to remove all attendees)
        transparency: Whether the event blocks availability. 'opaque' is busy; 'transparent' is free.
        eventType: Event type to set. Switching types clears properties that only belong to the old type.
        clear_fields: Optional event fields to remove, e.g. ['source', 'transparency'].
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        The updated event object
    """
    return events.update_event(
        eventId=eventId,
        summary=summary,
        start=start,
        end=end,
        description=description,
        location=location,
        calendar_id=calendar_id,
        recurrence=recurrence,
        reminders=reminders,
        attendees=attendees,
        creator=creator,
        organizer=organizer,
        extendedProperties=extendedProperties,
        source=source,
        transparency=transparency,
        eventType=eventType,
        outOfOfficeProperties=outOfOfficeProperties,
        focusTimeProperties=focusTimeProperties,
        workingLocationProperties=workingLocationProperties,
        birthdayProperties=birthdayProperties,
        clear_fields=clear_fields,
    )


@mcp.tool()
@_with_account
@_snapshot_on_write
def delete_event(
    eventId: EventId,
    calendar_id: CalendarId = "primary",
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Deletes an event from the calendar.

    Args:
        eventId: ID of the event to delete
        calendar_id: Calendar containing the event (default 'primary')
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        Confirmation of deletion
    """
    return events.delete_event(eventId=eventId, calendar_id=calendar_id)


@mcp.tool()
@_with_account
@_snapshot_on_write
def respond_to_event(
    eventId: EventId,
    email: EmailStr,
    response: AttendeeResponseStatus,
    calendar_id: CalendarId = "primary",
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Update an attendee's RSVP response on an event.

    Args:
        eventId: ID of the event
        email: Email address of the attendee responding
        response: Response status — 'accepted', 'declined', 'tentative', or 'needsAction'
        calendar_id: Calendar containing the event (default 'primary')
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        The updated event with attendee responses
    """
    return events.respond_to_event(eventId=eventId, email=email, response=response, calendar_id=calendar_id)


@mcp.tool()
@_with_account
@_snapshot_on_write
def create_calendar(
    summary: CalendarSummary,
    description: str | None = None,
    timeZone: CalendarTimeZone | None = None,
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Creates a new calendar (e.g., 'Personal', 'Team Meetings', 'On-Call').

    Args:
        summary: Calendar name
        description: Calendar description
        timeZone: IANA time zone for the calendar. Defaults to the primary calendar time zone.
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        The created calendar object
    """
    return calendars.create_calendar(summary=summary, description=description, timeZone=timeZone)


@mcp.tool()
@_with_account
def list_calendars(account_id: CalendarAccountId | None = None) -> dict:
    """
    Lists all calendars. Always includes the 'primary' calendar plus any
    additional calendars that have been created.

    Returns:
        List of calendars with their IDs, names, and event counts
    """
    return calendars.list_calendars()


@mcp.tool()
@_with_account
def list_events(
    timeMin: Rfc3339OffsetDateTimeString,
    timeMax: Rfc3339OffsetDateTimeString,
    maxResults: ListEventsMaxResults | None = None,
    orderBy: SearchEventsOrderBy | None = None,
    calendar_id: CalendarId | None = None,
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Lists events within a specified time range.

    Args:
        timeMin: Lower bound (exclusive) for an event's end time (ISO format).
                 Events that end after this time are included.
        timeMax: Upper bound (exclusive) for an event's start time (ISO format).
                 Events that start before this time are included.
        maxResults: Maximum number of events to return
        orderBy: Sort order ('startTime' or 'updated')
        calendar_id: Calendar to list events from. If omitted, lists events from all calendars.
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        List of events that overlap the given time range
    """
    return search.list_events(
        timeMin=timeMin,
        timeMax=timeMax,
        maxResults=maxResults,
        orderBy=orderBy,
        calendar_id=calendar_id,
    )


@mcp.tool()
@_with_account
def search_events(
    query: str = "",
    timeMin: Rfc3339OffsetDateTimeString | None = None,
    timeMax: Rfc3339OffsetDateTimeString | None = None,
    calendar_id: CalendarId | None = None,
    attendee_email: EmailStr | None = None,
    response_status: AttendeeResponseStatus | None = None,
    organizer_email: EmailStr | None = None,
    creator_email: EmailStr | None = None,
    maxResults: SearchEventsMaxResults | None = None,
    orderBy: SearchEventsOrderBy | None = SearchEventsOrderBy.START_TIME,
    account_id: CalendarAccountId | None = None,
) -> SearchEventsResponse:
    """
    Search calendar events by keyword, phrase, attendee, location, or calendar.

    Args:
        query: Search query. Bare words are ANDed across summary, description,
               location, attendees, and calendar names. Quoted phrases require
               exact adjacency. May be empty when attendee_email is provided.
        timeMin: Optional lower bound (exclusive) for an event's end time.
        timeMax: Optional upper bound (exclusive) for an event's start time.
        calendar_id: Calendar to search. If omitted, searches all calendars.
        attendee_email: Optional attendee email or display-name substring.
        response_status: Optional attendee RSVP filter ('accepted', 'declined',
                         'tentative', or 'needsAction'). When attendee_email is
                         provided, applies to that attendee; otherwise matches
                         any attendee with that status.
        organizer_email: Optional organizer email substring.
        creator_email: Optional creator email substring.
        maxResults: Maximum number of matching events to return. 0 returns no results.
        orderBy: Sort order ('startTime', 'updated', or None).
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        Matching events with calendar_id annotated on each result.
    """
    return search.search_events(
        query=query,
        timeMin=timeMin,
        timeMax=timeMax,
        calendar_id=calendar_id,
        attendee_email=attendee_email,
        response_status=response_status,
        organizer_email=organizer_email,
        creator_email=creator_email,
        maxResults=maxResults,
        orderBy=orderBy,
    )


@mcp.tool()
@_with_account
def check_availability(
    timeMin: Rfc3339OffsetDateTimeString,
    timeMax: Rfc3339OffsetDateTimeString,
    duration_minutes: AvailabilityDurationMinutes = 30,
    calendar_id: CalendarId | None = None,
    attendee_emails: list[EmailStr] | None = None,
    account_id: CalendarAccountId | None = None,
) -> dict:
    """
    Check availability within a time range. Returns busy periods and free slots
    of at least the requested duration.

    When attendee_emails is provided, only events where at least one of those
    people is an attendee (and hasn't declined) are considered busy. This lets
    you find times that work for specific people.

    Args:
        timeMin: Start of the range to check (ISO format)
        timeMax: End of the range to check (ISO format)
        duration_minutes: Minimum free slot duration in minutes (default 30)
        calendar_id: Calendar to check. If omitted, checks all calendars.
        attendee_emails: Only consider events involving these attendees (optional)
        account_id: Calendar account to use in multi-account worlds. Defaults to the active account.

    Returns:
        Busy periods and available free slots within the range
    """
    return availability.check_availability(
        timeMin=timeMin,
        timeMax=timeMax,
        duration_minutes=duration_minutes,
        calendar_id=calendar_id,
        attendee_emails=attendee_emails,
    )