"""Google Calendar + Meet in-process MCP for Jarvis. Gives Claude real calendar hands: create events, spin up a Google Meet with a shareable link, and read what's coming up — all against Ahmed's own Google account via a one-time desktop OAuth consent (no API key, no per-call billing). Like ``desktop_control``, this runs IN-PROCESS via the Claude Agent SDK's in-process MCP server (key ``"gcal"``; Claude sees ``mcp__gcal__``). The Google client libraries are imported lazily inside the auth helper so merely importing this module never hard-fails when they're missing. Auth: a desktop OAuth client (``GOOGLE_OAUTH_JSON``, default ``google_oauth.json`` at the project root). The FIRST calendar action opens a browser once for Ahmed to approve; the resulting token is cached to ``GOOGLE_TOKEN_JSON`` (default ``google_token.json``) and refreshed silently forever after. Set ``MCP_GCAL=0`` to disable. Times are RFC3339-ish local wall-clock (``2026-07-08T15:00:00``); the calendar timezone defaults to ``CAL_TZ`` (Asia/Riyadh) so "3pm tomorrow" means 3pm here. """ from __future__ import annotations import os import uuid from datetime import datetime, timedelta from pathlib import Path _ROOT = Path(__file__).resolve().parent.parent # Full read/write on the user's calendars — create, list, update, delete. _SCOPES = ["https://www.googleapis.com/auth/calendar"] _TZ = os.environ.get("CAL_TZ", "Asia/Riyadh") _SVC = None # cached Calendar API service (built on first use) def enabled() -> bool: """Whether the calendar MCP should be exposed to Claude.""" if os.environ.get("MCP_GCAL", "1") == "0": return False oauth = _ROOT / os.environ.get("GOOGLE_OAUTH_JSON", "google_oauth.json") return oauth.exists() def _service(): """Return an authenticated Calendar API service, running the one-time browser consent if no cached token exists. Blocking — callers invoke it via ``asyncio.to_thread`` so the voice loop never stalls.""" global _SVC if _SVC is not None: return _SVC from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from googleapiclient.discovery import build oauth_path = _ROOT / os.environ.get("GOOGLE_OAUTH_JSON", "google_oauth.json") token_path = _ROOT / os.environ.get("GOOGLE_TOKEN_JSON", "google_token.json") creds = None if token_path.exists(): creds = Credentials.from_authorized_user_file(str(token_path), _SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: print(" [gcal] first-time Google sign-in — a browser window will " "open; approve Jarvis to see your calendar.") flow = InstalledAppFlow.from_client_secrets_file( str(oauth_path), _SCOPES) creds = flow.run_local_server(port=0) token_path.write_text(creds.to_json()) _SVC = build("calendar", "v3", credentials=creds, cache_discovery=False) return _SVC def _text(msg: str) -> dict: """Wrap a plain string as an MCP text content result.""" return {"content": [{"type": "text", "text": msg}]} def _end_from(start: str, end: str | None, minutes: int) -> str: """Resolve the end time: explicit `end` wins, else start + `minutes`.""" if end: return end dt = datetime.fromisoformat(start) return (dt + timedelta(minutes=minutes)).isoformat() def build_server(): """Build and return the in-process ``gcal`` MCP server.""" import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server @tool("calendar_create_event", "Create a calendar event on Ahmed's Google Calendar. Times are local " "wall-clock in ISO form like '2026-07-08T15:00:00'. Give either `end` " "or `duration_minutes` (default 30). Attendees (comma-separated " "emails) get an invite emailed automatically.", {"title": str, "start": str, "end": str, "duration_minutes": int, "attendees": str, "description": str, "location": str}) async def calendar_create_event(args: dict) -> dict: try: title = str(args.get("title", "")).strip() or "(untitled)" start = str(args.get("start", "")).strip() if not start: return _text("create_event failed: no start time given.") end = _end_from(start, (args.get("end") or "").strip() or None, int(args.get("duration_minutes") or 30)) attendees = [e.strip() for e in str(args.get("attendees", "")).split(",") if e.strip()] body = { "summary": title, "start": {"dateTime": start, "timeZone": _TZ}, "end": {"dateTime": end, "timeZone": _TZ}, } if attendees: body["attendees"] = [{"email": e} for e in attendees] if args.get("description"): body["description"] = str(args["description"]) if args.get("location"): body["location"] = str(args["location"]) def _do(): svc = _service() return svc.events().insert( calendarId="primary", body=body, sendUpdates="all" if attendees else "none").execute() ev = await asyncio.to_thread(_do) who = f" — invited {', '.join(attendees)}" if attendees else "" return _text(f"Created '{title}' at {start} ({_TZ}){who}. " f"{ev.get('htmlLink', '')}") except Exception as e: # noqa: BLE001 return _text(f"create_event failed: {e}") @tool("calendar_create_meeting", "Create a calendar event WITH a Google Meet video link and return the " "link. Same fields as create_event; attendees (comma-separated " "emails) are emailed the invite + Meet link automatically.", {"title": str, "start": str, "end": str, "duration_minutes": int, "attendees": str, "description": str}) async def calendar_create_meeting(args: dict) -> dict: try: title = str(args.get("title", "")).strip() or "Meeting" start = str(args.get("start", "")).strip() if not start: return _text("create_meeting failed: no start time given.") end = _end_from(start, (args.get("end") or "").strip() or None, int(args.get("duration_minutes") or 30)) attendees = [e.strip() for e in str(args.get("attendees", "")).split(",") if e.strip()] body = { "summary": title, "start": {"dateTime": start, "timeZone": _TZ}, "end": {"dateTime": end, "timeZone": _TZ}, "conferenceData": {"createRequest": { "requestId": uuid.uuid4().hex, "conferenceSolutionKey": {"type": "hangoutsMeet"}, }}, } if attendees: body["attendees"] = [{"email": e} for e in attendees] if args.get("description"): body["description"] = str(args["description"]) def _do(): svc = _service() return svc.events().insert( calendarId="primary", body=body, conferenceDataVersion=1, sendUpdates="all" if attendees else "none").execute() ev = await asyncio.to_thread(_do) link = ev.get("hangoutLink", "(no link returned)") who = f", invited {', '.join(attendees)}" if attendees else "" return _text(f"Meeting '{title}' at {start} ({_TZ}){who}. " f"Meet link: {link}") except Exception as e: # noqa: BLE001 return _text(f"create_meeting failed: {e}") @tool("calendar_list", "List upcoming events on Ahmed's calendar over the next `days` days " "(default 7, max ~30).", {"days": int}) async def calendar_list(args: dict) -> dict: try: days = max(1, min(int(args.get("days") or 7), 30)) def _do(): svc = _service() # timeMin must be RFC3339 with a 'Z'/offset — use UTC now. from datetime import timezone now = datetime.now(timezone.utc) tmin = now.isoformat() tmax = (now + timedelta(days=days)).isoformat() res = svc.events().list( calendarId="primary", timeMin=tmin, timeMax=tmax, singleEvents=True, orderBy="startTime", maxResults=50).execute() return res.get("items", []) items = await asyncio.to_thread(_do) if not items: return _text(f"No events in the next {days} days.") lines = [] for ev in items: when = ev.get("start", {}).get( "dateTime") or ev.get("start", {}).get("date", "?") lines.append(f"{when} — {ev.get('summary', '(untitled)')}") return _text("\n".join(lines)) except Exception as e: # noqa: BLE001 return _text(f"calendar_list failed: {e}") return create_sdk_mcp_server( name="gcal", version="1.0.0", tools=[calendar_create_event, calendar_create_meeting, calendar_list], )