gcalendar

gcalendar

Google Calendar connector — event sync via the Calendar REST API v3.

Uses OAuth 2.0 tokens stored locally (see :mod:openjarvis.connectors.oauth). All network calls are isolated in module-level functions (_gcal_api_*) to make them trivially mockable in tests.

Classes

GCalendarConnector

GCalendarConnector(credentials_path: str = '')

Bases: BaseConnector

Connector that syncs events from Google Calendar via the REST API v3.

Authentication is handled through Google OAuth 2.0. Tokens are stored locally in a JSON credentials file.

PARAMETER	DESCRIPTION
credentials_path	Path to the JSON file where OAuth tokens are stored. Defaults to ~/.openjarvis/connectors/gcalendar.json. TYPE: str DEFAULT: ''

Source code in src/openjarvis/connectors/gcalendar.py

def __init__(self, credentials_path: str = "") -> None:
    self._credentials_path = resolve_google_credentials(
        credentials_path or _DEFAULT_CREDENTIALS_PATH
    )
    self._items_synced: int = 0
    self._items_total: int = 0
    self._last_sync: Optional[datetime] = None
    self._last_cursor: Optional[str] = None

Functions

is_connected

is_connected() -> bool

Return True if a credentials file with a valid access token exists.

Source code in src/openjarvis/connectors/gcalendar.py

def is_connected(self) -> bool:
    """Return ``True`` if a credentials file with a valid access token exists."""
    tokens = load_tokens(self._credentials_path)
    if tokens is None:
        return False
    # Must have an actual access_token, not just a client_id
    return bool(tokens.get("access_token") or tokens.get("token"))

disconnect

disconnect() -> None

Delete the stored credentials file.

Source code in src/openjarvis/connectors/gcalendar.py

def disconnect(self) -> None:
    """Delete the stored credentials file."""
    delete_tokens(self._credentials_path)

auth_url

auth_url() -> str

Return a Google OAuth consent URL requesting calendar.readonly scope.

Source code in src/openjarvis/connectors/gcalendar.py

def auth_url(self) -> str:
    """Return a Google OAuth consent URL requesting ``calendar.readonly`` scope."""
    tokens = load_tokens(self._credentials_path)
    client_id = ""
    if tokens:
        client_id = tokens.get("client_id", "")
    if not client_id:
        return "https://console.cloud.google.com/apis/credentials"
    return build_google_auth_url(
        client_id=client_id,
        scopes=GOOGLE_ALL_SCOPES,
    )

handle_callback

handle_callback(code: str) -> None

Handle the OAuth callback.

If code looks like a client_id:client_secret pair (containing .apps.googleusercontent.com), store the credentials and trigger the full browser-based OAuth flow. Otherwise treat it as a raw token / auth code.

Source code in src/openjarvis/connectors/gcalendar.py

def handle_callback(self, code: str) -> None:
    """Handle the OAuth callback.

    If *code* looks like a ``client_id:client_secret`` pair (containing
    ``.apps.googleusercontent.com``), store the credentials and trigger
    the full browser-based OAuth flow.  Otherwise treat it as a raw
    token / auth code.
    """
    code = code.strip()
    # If user pastes client_id:client_secret, store and run OAuth flow
    if ":" in code and ".apps.googleusercontent.com" in code:
        client_id, client_secret = code.split(":", 1)
        save_tokens(
            self._credentials_path,
            {
                "client_id": client_id.strip(),
                "client_secret": client_secret.strip(),
            },
        )
        import threading

        def _run() -> None:
            try:
                run_oauth_flow(
                    client_id=client_id.strip(),
                    client_secret=client_secret.strip(),
                    scopes=GOOGLE_ALL_SCOPES,
                    credentials_path=self._credentials_path,
                )
            except Exception:  # noqa: BLE001
                pass

        threading.Thread(target=_run, daemon=True).start()
    else:
        # Raw token or auth code
        save_tokens(self._credentials_path, {"token": code})

sync

sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Yield :class:Document objects for Google Calendar events.

Fetches all calendars from the calendarList endpoint, then paginates through each calendar's events.

PARAMETER	DESCRIPTION
since	Only return events starting after this datetime. Defaults to 24 hours ago when None. TYPE: Optional[datetime] DEFAULT: None
cursor	nextPageToken from a previous sync to resume pagination. TYPE: Optional[str] DEFAULT: None

Source code in src/openjarvis/connectors/gcalendar.py

def sync(
    self,
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,
) -> Iterator[Document]:
    """Yield :class:`Document` objects for Google Calendar events.

    Fetches all calendars from the calendarList endpoint, then paginates
    through each calendar's events.

    Parameters
    ----------
    since:
        Only return events starting after this datetime.  Defaults to
        24 hours ago when ``None``.
    cursor:
        ``nextPageToken`` from a previous sync to resume pagination.
    """
    tokens = load_tokens(self._credentials_path)
    if not tokens:
        return

    token: str = tokens.get("access_token", tokens.get("token", ""))
    if not token:
        return

    # Fetch list of calendars
    calendars_resp = _gcal_api_calendars_list(token)
    calendars: List[Dict[str, Any]] = calendars_resp.get("items", [])

    # Default to 24 hours ago so we don't dump the entire calendar history
    if since is None:
        since = datetime.now() - timedelta(days=1)
    time_min = since.strftime("%Y-%m-%dT%H:%M:%SZ")

    synced = 0

    for calendar in calendars:
        calendar_id: str = calendar.get("id", "")
        if not calendar_id:
            continue

        page_token: Optional[str] = cursor

        while True:
            try:
                events_resp = _gcal_api_events_list(
                    token,
                    calendar_id,
                    page_token=page_token,
                    time_min=time_min,
                )
            except httpx.HTTPStatusError:
                break
            events: List[Dict[str, Any]] = events_resp.get("items", [])

            for event in events:
                evt_id: str = event.get("id", "")
                if not evt_id:
                    continue

                summary: str = event.get("summary", "")
                organizer: Dict[str, Any] = event.get("organizer", {})
                organizer_email: str = organizer.get("email", "")
                attendees: List[Dict[str, Any]] = event.get("attendees", [])
                participant_emails: List[str] = [
                    a.get("email", "") for a in attendees if a.get("email")
                ]
                timestamp = _parse_event_timestamp(event)
                html_link: Optional[str] = event.get("htmlLink")

                content = _format_event(event)

                doc = Document(
                    doc_id=f"gcalendar:{evt_id}",
                    source="gcalendar",
                    doc_type="event",
                    content=content,
                    title=summary,
                    author=organizer_email,
                    participants=participant_emails,
                    timestamp=timestamp,
                    url=html_link,
                    metadata={
                        "calendar_id": calendar_id,
                        "event_id": evt_id,
                    },
                )
                synced += 1
                yield doc

            next_page: Optional[str] = events_resp.get("nextPageToken")
            if not next_page:
                self._last_cursor = None
                break
            page_token = next_page
            self._last_cursor = next_page

    self._items_synced = synced
    self._last_sync = datetime.now()

sync_status

sync_status() -> SyncStatus

Return sync progress from the most recent :meth:sync call.

Source code in src/openjarvis/connectors/gcalendar.py

def sync_status(self) -> SyncStatus:
    """Return sync progress from the most recent :meth:`sync` call."""
    return SyncStatus(
        state="idle",
        items_synced=self._items_synced,
        last_sync=self._last_sync,
        cursor=self._last_cursor,
    )

mcp_tools

mcp_tools() -> List[ToolSpec]

Expose three MCP tool specs for real-time Google Calendar queries.

Source code in src/openjarvis/connectors/gcalendar.py

def mcp_tools(self) -> List[ToolSpec]:
    """Expose three MCP tool specs for real-time Google Calendar queries."""
    return [
        ToolSpec(
            name="calendar_get_events_today",
            description=(
                "Retrieve all Google Calendar events scheduled for today. "
                "Returns a list of events with title, time, location, "
                "and attendees."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "calendar_id": {
                        "type": "string",
                        "description": (
                            "Calendar ID to query. Defaults to 'primary'."
                        ),
                        "default": "primary",
                    },
                },
                "required": [],
            },
            category="productivity",
        ),
        ToolSpec(
            name="calendar_search_events",
            description=(
                "Search Google Calendar events by keyword. "
                "Matches against event titles, descriptions, and locations."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search term to match against event fields",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of events to return",
                        "default": 20,
                    },
                    "calendar_id": {
                        "type": "string",
                        "description": (
                            "Calendar ID to search. Defaults to 'primary'."
                        ),
                        "default": "primary",
                    },
                },
                "required": ["query"],
            },
            category="productivity",
        ),
        ToolSpec(
            name="calendar_next_meeting",
            description=(
                "Find the next upcoming meeting on the user's Google Calendar. "
                "Returns title, start time, location, and attendees."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "calendar_id": {
                        "type": "string",
                        "description": (
                            "Calendar ID to query. Defaults to 'primary'."
                        ),
                        "default": "primary",
                    },
                },
                "required": [],
            },
            category="productivity",
        ),
    ]

Functions