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 = 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=[_GCAL_SCOPE],
    )

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=[_GCAL_SCOPE],
                    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	Not yet used (filtering is done server-side via timeMin). 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,  # noqa: ARG002 — reserved for future use
    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:
        Not yet used (filtering is done server-side via ``timeMin``).
    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", [])

    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
                )
            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