calendar_events_personal

Instructions

Implementation Reference

• tools.py:48-54 (handler)

  The @mcp.tool() decorated handler function 'calendar_events_personal' that loads personal account credentials and delegates to the calendar_events API function.

  @mcp.tool()
  def calendar_events_personal(time_min: str, time_max: str, max_results: int = 50) -> dict[str, Any]:
      """List events for the configured personal account calendars."""
      result = _load_or_error("personal")
      if isinstance(result, dict):
          return result
      return calendar_events(result, time_min, time_max, max_results)

• tools.py:49-49 (schema)

  The function signature defines the input schema: time_min (str), time_max (str), max_results (int, default 50). The return type is dict[str, Any].

  def calendar_events_personal(time_min: str, time_max: str, max_results: int = 50) -> dict[str, Any]:

• tools.py:48-48 (registration)

  The tool is registered via the @mcp.tool() decorator on line 48, which attaches it to the FastMCP server instance created on line 13.

  @mcp.tool()

• tools.py:16-21 (helper)

  The _load_or_error helper loads credentials for the given account key and returns either a Credentials object or an error dict.

  def _load_or_error(account: Account):
      """Return credentials or an error envelope if Keychain lookup fails."""
      try:
          return load_credentials(account)
      except RuntimeError as e:
          return {"ok": False, "error": str(e), "code": 401}

• api.py:98-197 (helper)

  The calendar_events function in api.py that performs the actual Google Calendar API queries. Called by calendar_events_personal with no calendar_names filter (personal gets all calendars).

  def calendar_events(
      creds: Credentials,
      time_min: str,
      time_max: str,
      max_results: int = 50,
      calendar_names: list[str] | None = None,
  ) -> dict[str, Any]:
      """List calendar events across all owned and writable calendars."""
      from google.auth.exceptions import RefreshError
      from googleapiclient.discovery import build
  
      max_results = min(max_results, 100)
  
      try:
          service = build("calendar", "v3", credentials=creds)
  
          all_calendars: list[dict[str, Any]] = []
          page_token: str | None = None
          while True:
              response = service.calendarList().list(pageToken=page_token).execute(num_retries=3)
              all_calendars.extend(response.get("items", []))
              page_token = response.get("nextPageToken")
              if not page_token:
                  break
  
          owned_calendars = [
              cal for cal in all_calendars if cal.get("accessRole") in ("owner", "writer")
          ]
  
          if calendar_names is not None:
              owned_calendars = [
                  cal
                  for cal in owned_calendars
                  if cal.get("summary", "") in calendar_names or cal.get("id", "") in calendar_names
              ]
  
          results: list[dict[str, Any]] = []
  
          for cal in owned_calendars:
              events_for_calendar = 0
              events_page_token: str | None = None
              while events_for_calendar < max_results:
                  remaining = max_results - events_for_calendar
                  events_response = (
                      service.events()
                      .list(
                          calendarId=cal["id"],
                          timeMin=time_min,
                          timeMax=time_max,
                          maxResults=min(remaining, max_results),
                          singleEvents=True,
                          orderBy="startTime",
                          pageToken=events_page_token,
                      )
                      .execute(num_retries=3)
                  )
  
                  for event in events_response.get("items", []):
                      if events_for_calendar >= max_results:
                          break
  
                      start_raw = event.get("start", {})
                      end_raw = event.get("end", {})
                      all_day = "date" in start_raw and "dateTime" not in start_raw
  
                      attendees = [
                          {"name": a.get("displayName", ""), "email": a.get("email", "")}
                          for a in event.get("attendees", [])
                      ]
  
                      results.append(
                          {
                              "id": event["id"],
                              "summary": event.get("summary", ""),
                              "start": start_raw.get("date")
                              if all_day
                              else start_raw.get("dateTime", ""),
                              "end": end_raw.get("date") if all_day else end_raw.get("dateTime", ""),
                              "location": event.get("location"),
                              "description": event.get("description"),
                              "attendees": attendees,
                              "status": event.get("status", "confirmed"),
                              "all_day": all_day,
                          }
                      )
                      events_for_calendar += 1
  
                  events_page_token = events_response.get("nextPageToken")
                  if not events_page_token:
                      break
  
          results.sort(key=lambda e: e["start"] or "")
          return {"ok": True, "data": results}
  
      except HttpError as e:
          return _parse_http_error(e)
      except RefreshError as e:
          return {"ok": False, "error": f"token refresh failed: {e}", "code": 401}
      except Exception as e:
          return {"ok": False, "error": f"upstream failure: {type(e).__name__}", "code": 503}