Calendar Connector¶

The CalendarConnector provides access to production schedules, shift patterns, technician availability, and planned downtime windows. It supports three pluggable backends behind a unified interface.

Backend	Source	Read	Write
iCal	`.ics` files or URLs	✅	❌ (read-only)
Google Calendar	Google Calendar API v3	✅	✅
Outlook	Microsoft Graph API	✅	✅

Prerequisites¶

iCalGoogle CalendarOutlook

An .ics file (local path or HTTP/HTTPS URL)
No external service credentials needed

A Google Cloud project with Calendar API enabled
A service account key file or OAuth2 client secrets
Shared calendars must grant the service account read/write access

An Azure AD app registration with Calendars.Read and Calendars.ReadWrite application permissions
Admin consent granted for the tenant
Tenant ID, Client ID, and Client Secret

Installation¶

# iCal backend (lightweight)
pip install machina-ai[ical]

# Google Calendar backend
pip install machina-ai[google-calendar]

# Outlook / Microsoft 365 backend
pip install machina-ai[outlook-calendar]

# All three backends
pip install machina-ai[calendar]

Quick Start¶

iCalGoogle CalendarOutlook

from machina.connectors import Calendar
from machina.domain.calendar import EventType

cal = Calendar(
    backend="ical",
    source="/data/production_schedule.ics",
    calendar_type_map={
        "production": EventType.PRODUCTION,
        "shutdown": EventType.DOWNTIME,
    },
)

await cal.connect()
events = await cal.read_events()

from machina.connectors import Calendar
from machina.domain.calendar import EventType

cal = Calendar(
    backend="google",
    service_account_file="/secrets/sa-key.json",
    calendar_type_map={
        "prod-calendar-id@group.calendar.google.com": EventType.PRODUCTION,
        "shifts-calendar-id@group.calendar.google.com": EventType.SHIFT,
    },
)

await cal.connect()
events = await cal.read_events(calendar_id="prod-calendar-id@group.calendar.google.com")

from machina.connectors import Calendar
from machina.domain.calendar import EventType

cal = Calendar(
    backend="outlook",
    tenant_id="your-tenant-id",
    client_id="your-client-id",
    client_secret="your-client-secret",
    user_id="plant-ops@company.com",
    calendar_type_map={
        "AAMkAG...": EventType.PRODUCTION,  # Outlook calendar ID
    },
)

await cal.connect()
events = await cal.read_events(calendar_id="AAMkAG...")

YAML Configuration¶

connectors:
  calendar:
    type: calendar
    settings:
      backend: ical
      source: /data/schedules/production.ics
      calendar_type_map:
        production: production
        shutdown: downtime
        shift: shift

Capabilities¶

Capability	iCal	Google	Outlook	Description
`read_calendar_events`	✅	✅	✅	List and filter calendar events
`create_calendar_event`	❌	✅	✅	Create a new event
`delete_calendar_event`	❌	✅	✅	Delete an existing event

Calendar Type Mapping¶

The calendar_type_map parameter maps calendar identifiers to EventType values for automatic classification:

iCal: Maps summary keywords — if the keyword appears in an event's title, it's classified with that type
Google/Outlook: Maps calendar IDs — all events from a calendar inherit its type

from machina.domain.calendar import EventType

# iCal: keyword-based mapping
cal = Calendar(
    backend="ical",
    source="schedules.ics",
    calendar_type_map={
        "production": EventType.PRODUCTION,
        "shutdown": EventType.DOWNTIME,
        "shift": EventType.SHIFT,
    },
)

# Google: calendar-ID-based mapping
cal = Calendar(
    backend="google",
    service_account_file="sa.json",
    calendar_type_map={
        "prod-cal@group.calendar.google.com": EventType.PRODUCTION,
        "shifts-cal@group.calendar.google.com": EventType.SHIFT,
    },
)

Available event types: production, maintenance, downtime, shift, meeting, other.

Convenience Methods¶

The connector provides helper methods that filter events by type:

from datetime import datetime, timezone

start = datetime(2026, 4, 1, tzinfo=timezone.utc)
end = datetime(2026, 4, 30, tzinfo=timezone.utc)

# Production windows
schedule = await cal.get_production_schedule(start=start, end=end)

# Planned downtime / shutdowns
downtime = await cal.get_planned_downtime(start=start, end=end)

# Technician shift patterns
shifts = await cal.get_technician_availability(start=start, end=end)

Domain Entities¶

The connector returns CalendarEvent instances and also provides related domain models:

CalendarEvent — A single calendar event with id, title, start, end, event_type, attendees, and metadata
PlannedDowntime — A typed downtime window with asset_id, area, reason, and approved fields
ShiftPattern — A recurring shift definition with start_time, end_time, days_of_week, technicians, and skills

Recurring Events¶

The iCal backend automatically expands recurring events (RRULE) within the requested time range. Google and Outlook APIs handle expansion server-side when singleEvents=True.

Maximum 500 occurrences are returned per recurring event to prevent memory issues.

Limitations¶

iCal backend is read-only — creating or deleting events raises ConnectorError
iCal URL fetching uses Python's urllib with a 30-second timeout
Google Calendar service account requires explicit calendar sharing
Outlook requires Azure AD application permissions with admin consent
Recurring event expansion is capped at 500 occurrences per master event
Time zones: all events are normalised to UTC-aware datetimes

API Reference¶

CalendarConnector ¶

CalendarConnector(*, backend: str = 'ical', calendar_type_map: dict[str, EventType] | None = None, **kwargs: Any)

Connector for calendar and scheduling data.

Reads (and optionally creates/deletes) calendar events from Google Calendar, Microsoft Outlook, or iCal sources. Events are returned as :class:~machina.domain.calendar.CalendarEvent domain entities.

Convenience methods filter events by :class:EventType so the agent can quickly retrieve production schedules, planned downtime, and technician availability.

Parameters:

Name	Type	Description	Default
`backend`	`str`	Calendar backend — `"google"`, `"outlook"`, or `"ical"`. Defaults to `"ical"`.	`'ical'`
`calendar_type_map`	`dict[str, EventType] \| None`	Mapping of calendar IDs (or summary keywords for iCal) to :class:`EventType` values for automatic classification.	`None`
`**kwargs`	`Any`	Backend-specific parameters forwarded to the selected backend class.	`{}`

Example

from machina.connectors.calendar import CalendarConnector

# iCal — read a local .ics file
cal = CalendarConnector(
    backend="ical",
    source="/data/production_schedule.ics",
    calendar_type_map={"shutdown": EventType.DOWNTIME},
)
await cal.connect()
events = await cal.read_events(start=..., end=...)

# Google Calendar
cal = CalendarConnector(
    backend="google",
    service_account_file="/secrets/sa-key.json",
    calendar_type_map={"prod-cal-id": EventType.PRODUCTION},
)

connect `async` ¶

connect() -> None

Initialise the backend and establish a connection.

disconnect `async` ¶

disconnect() -> None

Gracefully close the backend connection.

health_check `async` ¶

health_check() -> ConnectorHealth

Check whether the backend is connected and operational.

read_events `async` ¶

read_events(*, calendar_id: str = '', start: datetime | None = None, end: datetime | None = None, event_type: EventType | None = None, max_results: int = 250) -> list[CalendarEvent]

Read calendar events, optionally filtered by type and time range.

Parameters:

Name	Type	Description	Default
`calendar_id`	`str`	Calendar to read from (backend-specific).	`''`
`start`	`datetime \| None`	Only include events starting at or after this time.	`None`
`end`	`datetime \| None`	Only include events ending at or before this time.	`None`
`event_type`	`EventType \| None`	Filter to a specific :class:`EventType`.	`None`
`max_results`	`int`	Maximum events to return.	`250`

Returns:

Type	Description
`list[CalendarEvent]`	List of :class:`CalendarEvent` instances.

create_event `async` ¶

create_event(event: CalendarEvent, calendar_id: str = '') -> CalendarEvent

Create a calendar event.

Parameters:

Name	Type	Description	Default
`event`	`CalendarEvent`	The event to create.	required
`calendar_id`	`str`	Target calendar (backend-specific).	`''`

Returns:

Type	Description
`CalendarEvent`	The created event (with backend-assigned ID).

Raises:

Type	Description
`ConnectorError`	If the backend is read-only (iCal).

delete_event `async` ¶

delete_event(event_id: str, calendar_id: str = '') -> None

Delete a calendar event.

Parameters:

Name	Type	Description	Default
`event_id`	`str`	Event identifier.	required
`calendar_id`	`str`	Calendar containing the event (backend-specific).	`''`

Raises:

Type	Description
`ConnectorError`	If the backend is read-only (iCal).

get_production_schedule `async` ¶

get_production_schedule(start: datetime | None = None, end: datetime | None = None, calendar_id: str = '') -> list[CalendarEvent]

Return production-schedule events within the time range.

Parameters:

Name	Type	Description	Default
`start`	`datetime \| None`	Lower bound filter.	`None`
`end`	`datetime \| None`	Upper bound filter.	`None`
`calendar_id`	`str`	Calendar ID to read from.	`''`

Returns:

Type	Description
`list[CalendarEvent]`	Events classified as :attr:`EventType.PRODUCTION`.

get_planned_downtime `async` ¶

get_planned_downtime(start: datetime | None = None, end: datetime | None = None, calendar_id: str = '') -> list[CalendarEvent]

Return planned-downtime events within the time range.

Parameters:

Name	Type	Description	Default
`start`	`datetime \| None`	Lower bound filter.	`None`
`end`	`datetime \| None`	Upper bound filter.	`None`
`calendar_id`	`str`	Calendar ID to read from.	`''`

Returns:

Type	Description
`list[CalendarEvent]`	Events classified as :attr:`EventType.DOWNTIME`.

get_technician_availability `async` ¶

get_technician_availability(start: datetime | None = None, end: datetime | None = None, calendar_id: str = '') -> list[CalendarEvent]

Return shift / technician-availability events.

Parameters:

Name	Type	Description	Default
`start`	`datetime \| None`	Lower bound filter.	`None`
`end`	`datetime \| None`	Upper bound filter.	`None`
`calendar_id`	`str`	Calendar ID to read from.	`''`

Returns:

Type	Description
`list[CalendarEvent]`	Events classified as :attr:`EventType.SHIFT`.

Calendar Connector¶

Prerequisites¶

Installation¶

Quick Start¶

YAML Configuration¶

Capabilities¶

Calendar Type Mapping¶

Convenience Methods¶

Domain Entities¶

Recurring Events¶

Limitations¶

API Reference¶

CalendarConnector ¶

connect async ¶

disconnect async ¶

health_check async ¶

read_events async ¶

create_event async ¶

delete_event async ¶

get_production_schedule async ¶

get_planned_downtime async ¶

get_technician_availability async ¶

connect `async` ¶

disconnect `async` ¶

health_check `async` ¶

read_events `async` ¶

create_event `async` ¶

delete_event `async` ¶

get_production_schedule `async` ¶

get_planned_downtime `async` ¶

get_technician_availability `async` ¶