#!/usr/bin/env python3 """ Google Calendar MCP Server & CLI Tool Integrates Google Calendar (Family & Regensburg calendars) with OpenClaw via MCP stdio or direct CLI. Authenticates using a Google Cloud Service Account. Features: - Color-coding for family members and categories (Arzt=Rot, Mila=Gelb, Moritz=Blau, Amelie=Grün, Familie=Orange, Haushalt=Grau) - Conflict & travel-buffer detection before and during event creation - Rich location and description metadata (for Google Maps navigation & prep checklists) """ import os import sys import json import argparse from datetime import datetime, date, timedelta from typing import Dict, Any, List, Optional, Tuple, Union from zoneinfo import ZoneInfo # Environment Configuration DEFAULT_TIMEZONE_STR = os.environ.get("CALENDAR_TIMEZONE", "Europe/Berlin") DEFAULT_TZ = ZoneInfo(DEFAULT_TIMEZONE_STR) CALENDAR_ID_FAMILY = os.environ.get("CALENDAR_ID_FAMILY", "").strip() CALENDAR_ID_REGENSBURG = os.environ.get("CALENDAR_ID_REGENSBURG", "").strip() SCOPES = ["https://www.googleapis.com/auth/calendar"] # ============================================================================== # Color Coding Palette (Google Calendar Standard colorIds 1-11) # ============================================================================== COLOR_NAMES = { "11": "Rot (Tomate) - Arzt / Gesundheit / Dringend", "5": "Gelb (Banane) - Mila / Kinder / Baby", "9": "Blau (Heidelbeere) - Moritz / Arbeit", "2": "Grün (Salbei) - Amelie", "6": "Orange (Mandarine) - Familie / Ausflug / Feiern", "8": "Grau (Graphit) - Haushalt / Auto / Notizen", "1": "Lavendel", "3": "Traube / Lila", "4": "Flamingo / Rosa", "7": "Pfau / Türkis", "10": "Basilikum / Dunkelgrün", } CATEGORY_TO_COLOR = { # 🔴 Rot: Arzt / Gesundheit "arzt": "11", "kinderarzt": "11", "zahnarzt": "11", "doktor": "11", "gesundheit": "11", "impfung": "11", "impfen": "11", "u-untersuchung": "11", "vorsorge": "11", "klinik": "11", "krankenhaus": "11", "physio": "11", "apotheke": "11", "blutabnahme": "11", # 🟡 Gelb: Mila & Kinder "mila": "5", "kind": "5", "kinder": "5", "kita": "5", "kindergarten": "5", "krabbelgruppe": "5", "baby": "5", "spielplatz": "5", "babyschwimmen": "5", "kinderturnen": "5", # 🔵 Blau: Moritz "moritz": "9", "arbeit": "9", "work": "9", "büro": "9", "job": "9", # 🟢 Grün: Amelie "amelie": "2", "amy": "2", # 🟠 Orange: Familie & Feiern "familie": "6", "family": "6", "ausflug": "6", "urlaub": "6", "feier": "6", "party": "6", "geburtstag": "6", "besuch": "6", "oma": "6", "opa": "6", "essen": "6", "restaurant": "6", # ⚪ Grau: Haushalt & Wartung "haushalt": "8", "auto": "8", "tüv": "8", "werkstatt": "8", "müll": "8", "wartung": "8", } def resolve_color_id(summary: str, explicit_category_or_color: Optional[str] = None) -> Optional[str]: """Resolve colorId based on explicit argument or automatic keyword matching.""" if explicit_category_or_color: val = str(explicit_category_or_color).strip().lower() if val in COLOR_NAMES: return val if val in CATEGORY_TO_COLOR: return CATEGORY_TO_COLOR[val] # Auto-detection from summary keywords summary_lower = summary.lower() for keyword, color_id in CATEGORY_TO_COLOR.items(): if keyword in summary_lower: return color_id # U-Untersuchung regex-like check (e.g. U1..U9) import re if re.search(r"\bu[1-9]\b", summary_lower): return "11" # Arzt/Vorsorge return None def get_credentials(): """Load Google Service Account credentials from environment or file.""" from google.oauth2 import service_account sa_json_env = os.environ.get("GOOGLE_CALENDAR_SA_JSON", "").strip() if sa_json_env: if sa_json_env.startswith("{"): try: info = json.loads(sa_json_env) return service_account.Credentials.from_service_account_info(info, scopes=SCOPES) except Exception as e: raise ValueError(f"Failed to parse GOOGLE_CALENDAR_SA_JSON as JSON: {e}") elif os.path.isfile(sa_json_env): return service_account.Credentials.from_service_account_file(sa_json_env, scopes=SCOPES) gac = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "").strip() if gac and os.path.isfile(gac): return service_account.Credentials.from_service_account_file(gac, scopes=SCOPES) candidate_paths = [ "/home/openclaw/.openclaw/calendar_sa.json", "/home/node/.openclaw/calendar_sa.json", os.path.expanduser("~/.openclaw/calendar_sa.json"), os.path.join(os.path.dirname(__file__), "../../../../infrapuzzle/k8s/openclaw/calendar_sa.json.secret"), os.path.join(os.getcwd(), "k8s/openclaw/calendar_sa.json.secret"), os.path.join(os.getcwd(), "../k8s/openclaw/calendar_sa.json.secret"), ] for path in candidate_paths: if os.path.isfile(path): return service_account.Credentials.from_service_account_file(path, scopes=SCOPES) raise ValueError( "No Google Service Account credentials found. Please set GOOGLE_CALENDAR_SA_JSON " "or GOOGLE_APPLICATION_CREDENTIALS." ) def get_calendar_service(): """Build and return the Google Calendar v3 API client.""" from googleapiclient.discovery import build creds = get_credentials() return build("calendar", "v3", credentials=creds, cache_discovery=False) def resolve_calendar_id(calendar_identifier: Optional[str] = None) -> str: """Resolve calendar alias ('family' or 'regensburg') to its Google Calendar ID.""" ident = (calendar_identifier or "family").strip() ident_lower = ident.lower() if ident_lower in ("family", "familie", "familienkalender", "default", "primary"): if not CALENDAR_ID_FAMILY: raise ValueError( "CALENDAR_ID_FAMILY is not configured. Please set the environment variable " "CALENDAR_ID_FAMILY to the Google Calendar ID of the family calendar." ) return CALENDAR_ID_FAMILY if ident_lower in ("regensburg", "rgb", "regensburg-und-mehr", "regensburg_und_mehr", "events", "stadt"): if not CALENDAR_ID_REGENSBURG: raise ValueError( "Der Regensburg- / Event-Kalender ist noch nicht freigeschaltet (Berechtigungen ausstehend). " "Bitte nutze vorerst den Familienkalender ('family')." ) return CALENDAR_ID_REGENSBURG return ident def parse_datetime_str(dt_str: str) -> datetime: """Parse ISO date/datetime string, attaching Europe/Berlin if naive.""" dt_str = dt_str.strip() if dt_str.endswith("Z"): dt_str = dt_str[:-1] + "+00:00" dt = datetime.fromisoformat(dt_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=DEFAULT_TZ) return dt def normalize_recurrence(val: Any) -> Optional[List[str]]: """ Normalize recurrence input into a list of valid RFC 5545 RRULE/EXRULE strings accepted by Google Calendar API. """ if val is None or isinstance(val, bool): return None raw_rules: List[str] = [] if isinstance(val, str): val_str = val.strip() if not val_str: return None lower = val_str.lower() if lower in ("weekly", "wöchentlich"): return ["RRULE:FREQ=WEEKLY"] elif lower in ("daily", "täglich"): return ["RRULE:FREQ=DAILY"] elif lower in ("monthly", "monatlich"): return ["RRULE:FREQ=MONTHLY"] elif lower in ("yearly", "jährlich", "annual"): return ["RRULE:FREQ=YEARLY"] elif lower in ("workdays", "werktags", "werktäglich"): return ["RRULE:FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR"] elif lower in ("weekend", "wochenende"): return ["RRULE:FREQ=WEEKLY;BYDAY=SA,SU"] raw_rules = [r.strip() for r in val_str.split("\n") if r.strip()] elif isinstance(val, (list, tuple)): raw_rules = [str(r).strip() for r in val if str(r).strip()] else: return None if not raw_rules: return None cleaned_rules = [] for rule in raw_rules: rule_upper = rule.upper() if any(rule_upper.startswith(p) for p in ("RRULE:", "EXRULE:", "RDATE:", "EXDATE:")): cleaned_rules.append(rule) elif any(rule_upper.startswith(p) for p in ("FREQ=", "UNTIL=", "COUNT=", "INTERVAL=", "BYDAY=", "WKST=")): cleaned_rules.append(f"RRULE:{rule}") else: cleaned_rules.append(f"RRULE:{rule}") return cleaned_rules if cleaned_rules else None def normalize_reminders(val: Any) -> Optional[Dict[str, Any]]: """ Normalize reminder minutes into Google Calendar API reminders resource: {"useDefault": False, "overrides": [{"method": "popup", "minutes": X}, ...]} """ if val is None: return None minutes_list: List[int] = [] if isinstance(val, int): minutes_list = [val] elif isinstance(val, (list, tuple)): for item in val: try: minutes_list.append(int(item)) except (ValueError, TypeError): continue elif isinstance(val, str): val_str = val.strip() if val_str: for part in val_str.split(","): part = part.strip() try: minutes_list.append(int(part)) except (ValueError, TypeError): continue return { "useDefault": False, "overrides": [{"method": "popup", "minutes": m} for m in minutes_list] } def format_event(item: Dict[str, Any], calendar_name: str) -> Dict[str, Any]: """Format an event item into a clean representation including color labels.""" start = item.get("start", {}) end = item.get("end", {}) is_all_day = "date" in start start_str = start.get("date") if is_all_day else start.get("dateTime") end_str = end.get("date") if is_all_day else end.get("dateTime") color_id = item.get("colorId") color_label = COLOR_NAMES.get(color_id) if color_id else None recurrence = item.get("recurrence") recurring_event_id = item.get("recurringEventId") is_recurring = bool(recurrence or recurring_event_id) transparency = item.get("transparency", "opaque") show_as_busy = (transparency != "transparent") rem = item.get("reminders", {}) use_default_rem = rem.get("useDefault", True) reminder_minutes = [ov.get("minutes") for ov in rem.get("overrides", []) if "minutes" in ov] if not use_default_rem else None return { "id": item.get("id"), "calendar": calendar_name, "summary": item.get("summary", "(Ohne Titel)"), "start": start_str, "end": end_str, "all_day": is_all_day, "is_recurring": is_recurring, "recurrence": recurrence, "recurring_event_id": recurring_event_id, "show_as_busy": show_as_busy, "transparency": transparency, "reminders": { "use_default": use_default_rem, "reminder_minutes": reminder_minutes, }, "color_id": color_id, "color_name": color_label, "description": item.get("description", ""), "location": item.get("location", ""), "status": item.get("status", "confirmed"), "htmlLink": item.get("htmlLink", ""), } # ============================================================================== # Conflict & Buffer Detection # ============================================================================== def check_conflicts( calendar: str = "family", start_time: str = "", end_time: Optional[str] = None, buffer_minutes: int = 15, ) -> Dict[str, Any]: """ Check if a proposed timeslot conflicts with existing appointments or falls within a tight travel buffer (< buffer_minutes). """ if not start_time: raise ValueError("Parameter 'start_time' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) # Parse proposed time dt_start = parse_datetime_str(start_time) if end_time: dt_end = parse_datetime_str(end_time) else: dt_end = dt_start + timedelta(hours=1) # Search window: from (start - buffer) to (end + buffer) window_start = dt_start - timedelta(minutes=max(buffer_minutes, 60)) window_end = dt_end + timedelta(minutes=max(buffer_minutes, 60)) params = { "calendarId": cal_id, "timeMin": window_start.isoformat(), "timeMax": window_end.isoformat(), "singleEvents": True, "orderBy": "startTime", } res = service.events().list(**params).execute() items = res.get("items", []) direct_conflicts = [] buffer_warnings = [] for item in items: # Ignore all-day transparent events (they don't block time) if item.get("transparency") == "transparent": continue item_start_raw = item.get("start", {}).get("dateTime") item_end_raw = item.get("end", {}).get("dateTime") if not item_start_raw or not item_end_raw: continue ev_start = parse_datetime_str(item_start_raw) ev_end = parse_datetime_str(item_end_raw) # Check direct overlap: ev_start < dt_end and ev_end > dt_start if ev_start < dt_end and ev_end > dt_start: direct_conflicts.append(format_event(item, calendar)) else: # Check tight buffer before or after gap_before = (dt_start - ev_end).total_seconds() / 60.0 gap_after = (ev_start - dt_end).total_seconds() / 60.0 if 0 <= gap_before < buffer_minutes: buffer_warnings.append({ "type": "tight_buffer_before", "minutes_gap": int(gap_before), "event": format_event(item, calendar), }) elif 0 <= gap_after < buffer_minutes: buffer_warnings.append({ "type": "tight_buffer_after", "minutes_gap": int(gap_after), "event": format_event(item, calendar), }) has_conflict = len(direct_conflicts) > 0 has_buffer_warning = len(buffer_warnings) > 0 return { "has_conflict": has_conflict, "has_buffer_warning": has_buffer_warning, "direct_conflicts_count": len(direct_conflicts), "direct_conflicts": direct_conflicts, "buffer_warnings_count": len(buffer_warnings), "buffer_warnings": buffer_warnings, "buffer_minutes_checked": buffer_minutes, "proposed_start": dt_start.isoformat(), "proposed_end": dt_end.isoformat(), } # ============================================================================== # Calendar Operations # ============================================================================== def list_events( calendar: str = "family", time_min: Optional[str] = None, time_max: Optional[str] = None, query: Optional[str] = None, category: Optional[str] = None, color_id: Optional[str] = None, max_results: int = 25, ) -> Dict[str, Any]: """List events from the specified calendar with optional query, category, and color filters.""" service = get_calendar_service() cal_id = resolve_calendar_id(calendar) now = datetime.now(DEFAULT_TZ) if not time_min: start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0) time_min_str = start_of_day.isoformat() else: time_min_str = parse_datetime_str(time_min).isoformat() if not time_max: time_max_dt = now + timedelta(days=30) time_max_str = time_max_dt.isoformat() else: time_max_str = parse_datetime_str(time_max).isoformat() params: Dict[str, Any] = { "calendarId": cal_id, "timeMin": time_min_str, "timeMax": time_max_str, "singleEvents": True, "orderBy": "startTime", "maxResults": max_results, } if query: params["q"] = query res = service.events().list(**params).execute() items = res.get("items", []) # Filter by category / colorId if requested target_color_id = color_id or (resolve_color_id("", explicit_category_or_color=category) if category else None) if target_color_id: items = [it for it in items if it.get("colorId") == target_color_id] events = [format_event(item, calendar) for item in items] return { "calendar": calendar, "calendar_id": cal_id, "time_min": time_min_str, "time_max": time_max_str, "filter_category": category, "filter_color_id": target_color_id, "count": len(events), "events": events, } def create_event( calendar: str = "family", summary: str = "", start_time: str = "", end_time: Optional[str] = None, description: Optional[str] = None, location: Optional[str] = None, all_day: bool = False, category: Optional[str] = None, color_id: Optional[str] = None, recurrence: Optional[Any] = None, show_as_busy: Optional[bool] = None, transparency: Optional[str] = None, reminder_minutes: Optional[Any] = None, inclusive_end: bool = True, check_conflicts_flag: bool = True, buffer_minutes: int = 15, ) -> Dict[str, Any]: """ Create a new event in the specified calendar with automatic color-coding, conflict detection, recurrence (RRULE), transparency, and custom reminders. """ if not summary: raise ValueError("Parameter 'summary' is required.") if not start_time: raise ValueError("Parameter 'start_time' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) # 1. Resolve Color chosen_color_id = resolve_color_id(summary, explicit_category_or_color=color_id or category) # 2. Check for conflicts if timed event conflict_info = None if check_conflicts_flag and not all_day: conflict_info = check_conflicts( calendar=calendar, start_time=start_time, end_time=end_time, buffer_minutes=buffer_minutes, ) # 3. Build Google Event Resource body: Dict[str, Any] = { "summary": summary, } if description: body["description"] = description if location: body["location"] = location if chosen_color_id: body["colorId"] = chosen_color_id # Recurrence rules (RFC 5545 RRULE) norm_rec = normalize_recurrence(recurrence) if norm_rec: body["recurrence"] = norm_rec if conflict_info and (conflict_info["has_conflict"] or conflict_info["has_buffer_warning"]): conflict_info["recurring_note"] = "Conflict check was performed on the initial occurrence only." # Event Transparency (Free vs. Busy) if transparency: body["transparency"] = transparency.lower().strip() elif show_as_busy is False: body["transparency"] = "transparent" elif show_as_busy is True: body["transparency"] = "opaque" # Custom Reminders norm_rem = normalize_reminders(reminder_minutes) if norm_rem is not None: body["reminders"] = norm_rem if all_day: start_date = start_time.split("T")[0] dt_start = date.fromisoformat(start_date) if end_time: end_date_raw = end_time.split("T")[0] dt_end_raw = date.fromisoformat(end_date_raw) if inclusive_end and dt_end_raw >= dt_start: end_date = (dt_end_raw + timedelta(days=1)).isoformat() else: end_date = dt_end_raw.isoformat() else: end_date = (dt_start + timedelta(days=1)).isoformat() body["start"] = {"date": start_date} body["end"] = {"date": end_date} else: dt_start = parse_datetime_str(start_time) if end_time: dt_end = parse_datetime_str(end_time) else: dt_end = dt_start + timedelta(hours=1) body["start"] = { "dateTime": dt_start.isoformat(), "timeZone": DEFAULT_TIMEZONE_STR, } body["end"] = { "dateTime": dt_end.isoformat(), "timeZone": DEFAULT_TIMEZONE_STR, } created = service.events().insert(calendarId=cal_id, body=body).execute() result = { "status": "created", "event": format_event(created, calendar), } # Attach conflict warning if overlaps were detected if conflict_info and (conflict_info["has_conflict"] or conflict_info["has_buffer_warning"]): result["conflict_warning"] = conflict_info return result def update_event( calendar: str = "family", event_id: str = "", summary: Optional[str] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, description: Optional[str] = None, location: Optional[str] = None, category: Optional[str] = None, color_id: Optional[str] = None, recurrence: Optional[Any] = None, show_as_busy: Optional[bool] = None, transparency: Optional[str] = None, reminder_minutes: Optional[Any] = None, inclusive_end: bool = True, update_series: bool = False, ) -> Dict[str, Any]: """Update an existing event or series.""" if not event_id: raise ValueError("Parameter 'event_id' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) target_id = event_id if update_series: try: item_check = service.events().get(calendarId=cal_id, eventId=event_id).execute() if item_check.get("recurringEventId"): target_id = item_check["recurringEventId"] except Exception: pass existing = service.events().get(calendarId=cal_id, eventId=target_id).execute() if summary is not None: existing["summary"] = summary if description is not None: existing["description"] = description if location is not None: existing["location"] = location new_color = resolve_color_id(summary or existing.get("summary", ""), explicit_category_or_color=color_id or category) if new_color: existing["colorId"] = new_color if recurrence is not None: if recurrence == [] or recurrence == "" or recurrence is False: existing.pop("recurrence", None) else: norm_rec = normalize_recurrence(recurrence) if norm_rec: existing["recurrence"] = norm_rec if transparency is not None: existing["transparency"] = transparency.lower().strip() elif show_as_busy is not None: existing["transparency"] = "opaque" if show_as_busy else "transparent" if reminder_minutes is not None: norm_rem = normalize_reminders(reminder_minutes) if norm_rem is not None: existing["reminders"] = norm_rem if start_time is not None: if "date" in existing.get("start", {}): start_date_str = start_time.split("T")[0] existing["start"]["date"] = start_date_str dt_start = date.fromisoformat(start_date_str) if end_time: end_date_raw = end_time.split("T")[0] dt_end_raw = date.fromisoformat(end_date_raw) if inclusive_end and dt_end_raw >= dt_start: existing["end"]["date"] = (dt_end_raw + timedelta(days=1)).isoformat() else: existing["end"]["date"] = end_date_raw else: dt_start = parse_datetime_str(start_time) existing["start"] = { "dateTime": dt_start.isoformat(), "timeZone": DEFAULT_TIMEZONE_STR, } if end_time: dt_end = parse_datetime_str(end_time) existing["end"] = { "dateTime": dt_end.isoformat(), "timeZone": DEFAULT_TIMEZONE_STR, } updated = service.events().update(calendarId=cal_id, eventId=target_id, body=existing).execute() return { "status": "updated", "event": format_event(updated, calendar), "updated_series": (target_id != event_id or update_series), } def delete_event( calendar: str = "family", event_id: str = "", delete_series: bool = False, ) -> Dict[str, Any]: """Delete an event or recurring series from the specified calendar.""" if not event_id: raise ValueError("Parameter 'event_id' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) target_id = event_id if delete_series: try: item = service.events().get(calendarId=cal_id, eventId=event_id).execute() if item.get("recurringEventId"): target_id = item["recurringEventId"] except Exception: pass service.events().delete(calendarId=cal_id, eventId=target_id).execute() return { "status": "deleted", "calendar": calendar, "event_id": target_id, "deleted_series": (target_id != event_id or delete_series), } def move_event( event_id: str, source_calendar: str = "family", destination_calendar: str = "regensburg", ) -> Dict[str, Any]: """Move an event atomically between calendars (e.g. from regensburg to family or vice versa).""" if not event_id: raise ValueError("Parameter 'event_id' is required.") service = get_calendar_service() src_id = resolve_calendar_id(source_calendar) dst_id = resolve_calendar_id(destination_calendar) moved = service.events().move( calendarId=src_id, eventId=event_id, destination=dst_id, ).execute() return { "status": "moved", "source_calendar": source_calendar, "destination_calendar": destination_calendar, "event": format_event(moved, destination_calendar), } def find_free_slots( calendar: str = "family", time_min: Optional[str] = None, time_max: Optional[str] = None, duration_minutes: int = 60, day_start_hour: int = 8, day_end_hour: int = 20, ) -> Dict[str, Any]: """ Find available free time windows between day_start_hour and day_end_hour across specified calendar(s) using freebusy.query. """ service = get_calendar_service() now = datetime.now(DEFAULT_TZ) if not time_min: dt_start = now else: dt_start = parse_datetime_str(time_min) if not time_max: dt_end = dt_start + timedelta(days=7) else: dt_end = parse_datetime_str(time_max) # Determine which calendars to query cal_ids = [] cal_lower = calendar.lower().strip() if cal_lower in ("both", "alle", "all"): if CALENDAR_ID_FAMILY: cal_ids.append(CALENDAR_ID_FAMILY) if CALENDAR_ID_REGENSBURG: cal_ids.append(CALENDAR_ID_REGENSBURG) else: cal_ids.append(resolve_calendar_id(calendar)) body = { "timeMin": dt_start.isoformat(), "timeMax": dt_end.isoformat(), "timeZone": DEFAULT_TIMEZONE_STR, "items": [{"id": cid} for cid in cal_ids], } fb_res = service.freebusy().query(body=body).execute() cals_dict = fb_res.get("calendars", {}) busy_intervals = [] for cid, cdata in cals_dict.items(): for b in cdata.get("busy", []): b_start = parse_datetime_str(b["start"]) b_end = parse_datetime_str(b["end"]) busy_intervals.append((b_start, b_end)) busy_intervals.sort(key=lambda x: x[0]) merged_busy = [] for b_start, b_end in busy_intervals: if not merged_busy: merged_busy.append([b_start, b_end]) else: prev_start, prev_end = merged_busy[-1] if b_start <= prev_end: merged_busy[-1][1] = max(prev_end, b_end) else: merged_busy.append([b_start, b_end]) free_slots = [] current_day = dt_start.date() end_day = dt_end.date() while current_day <= end_day: window_start = datetime(current_day.year, current_day.month, current_day.day, day_start_hour, 0, 0, tzinfo=DEFAULT_TZ) window_end = datetime(current_day.year, current_day.month, current_day.day, day_end_hour, 0, 0, tzinfo=DEFAULT_TZ) eff_start = max(window_start, dt_start) eff_end = min(window_end, dt_end) if eff_start < eff_end: slot_pointer = eff_start for b_start, b_end in merged_busy: if b_end <= slot_pointer: continue if b_start >= eff_end: break if b_start > slot_pointer: gap_minutes = (b_start - slot_pointer).total_seconds() / 60.0 if gap_minutes >= duration_minutes: free_slots.append({ "start": slot_pointer.isoformat(), "end": b_start.isoformat(), "duration_minutes": int(gap_minutes), "date": current_day.isoformat(), }) slot_pointer = max(slot_pointer, b_end) if slot_pointer < eff_end: gap_minutes = (eff_end - slot_pointer).total_seconds() / 60.0 if gap_minutes >= duration_minutes: free_slots.append({ "start": slot_pointer.isoformat(), "end": eff_end.isoformat(), "duration_minutes": int(gap_minutes), "date": current_day.isoformat(), }) current_day += timedelta(days=1) return { "calendar": calendar, "search_min": dt_start.isoformat(), "search_max": dt_end.isoformat(), "requested_duration_minutes": duration_minutes, "day_hours": f"{day_start_hour:02d}:00 - {day_end_hour:02d}:00", "slots_found": len(free_slots), "free_slots": free_slots, } def get_event(calendar: str = "family", event_id: str = "") -> Dict[str, Any]: """Get event details by ID.""" if not event_id: raise ValueError("Parameter 'event_id' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) item = service.events().get(calendarId=cal_id, eventId=event_id).execute() return format_event(item, calendar) def test_connection() -> Dict[str, Any]: """Test Google Calendar API connectivity and list accessible calendars.""" service = get_calendar_service() creds = get_credentials() sa_email = getattr(creds, "service_account_email", "unknown") cal_list = service.calendarList().list().execute() calendars = [ { "id": item.get("id"), "summary": item.get("summary"), "accessRole": item.get("accessRole"), "primary": item.get("primary", False), } for item in cal_list.get("items", []) ] family_status = "not_configured" if CALENDAR_ID_FAMILY: try: f_cal = service.calendars().get(calendarId=CALENDAR_ID_FAMILY).execute() family_status = f"ok ({f_cal.get('summary')})" except Exception as e: family_status = f"error: {e}" regensburg_status = "not_configured" if CALENDAR_ID_REGENSBURG: try: r_cal = service.calendars().get(calendarId=CALENDAR_ID_REGENSBURG).execute() regensburg_status = f"ok ({r_cal.get('summary')})" except Exception as e: regensburg_status = f"error: {e}" return { "status": "ok", "service_account_email": sa_email, "configured_family_id": CALENDAR_ID_FAMILY or "(empty)", "family_calendar_check": family_status, "configured_regensburg_id": CALENDAR_ID_REGENSBURG or "(empty)", "regensburg_calendar_check": regensburg_status, "accessible_calendars": calendars, "color_palette": COLOR_NAMES, } # ============================================================================== # MCP Protocol Implementation (stdio JSON-RPC) # ============================================================================== TOOLS_MANIFEST = [ { "name": "calendar_list_events", "description": "List upcoming appointments and events from the Google Calendar. Returns event title, start/end time, color category, location, recurrence, and description.", "inputSchema": { "type": "object", "properties": { "calendar": { "type": "string", "description": "Which calendar to query: 'family' (default) for family appointments, or 'regensburg' for city/local events.", "default": "family", }, "time_min": { "type": "string", "description": "Start of time range in ISO 8601 format (e.g. '2026-09-20' or '2026-09-20T08:00:00'). Defaults to start of today.", }, "time_max": { "type": "string", "description": "End of time range in ISO 8601 format (e.g. '2026-09-27' or '2026-09-27T23:59:59'). Defaults to 30 days ahead.", }, "query": { "type": "string", "description": "Optional search term to filter event titles and descriptions.", }, "category": { "type": "string", "description": "Optional category filter: 'arzt', 'mila', 'moritz', 'amelie', 'familie', 'haushalt'.", }, "color_id": { "type": "string", "description": "Optional colorId filter ('11' for Arzt, '5' for Mila, '9' for Moritz, etc.).", }, "max_results": { "type": "integer", "description": "Maximum number of events to return (default 25).", "default": 25, }, }, }, }, { "name": "calendar_check_conflicts", "description": "Check whether a proposed appointment time conflicts with existing events in the calendar or has tight travel buffers (< buffer_minutes). Use this before booking new events to warn the user about overlaps.", "inputSchema": { "type": "object", "required": ["start_time"], "properties": { "calendar": { "type": "string", "description": "Which calendar to check: 'family' (default) or 'regensburg'.", "default": "family", }, "start_time": { "type": "string", "description": "Proposed start time in ISO format (e.g. '2026-09-20T14:30:00').", }, "end_time": { "type": "string", "description": "Proposed end time in ISO format (defaults to 1 hour after start_time).", }, "buffer_minutes": { "type": "integer", "description": "Buffer minutes to check before and after the event for travel time (default 15).", "default": 15, }, }, }, }, { "name": "calendar_create_event", "description": "Create a new appointment in Google Calendar. Supports automatic color-coding (Arzt=Rot, Mila=Gelb, Moritz=Blau, Amelie=Grün, Familie=Orange, Haushalt=Grau), recurring series (RRULE), transparency (busy vs free), phone notification reminders, address location, checklists, and conflict warnings.", "inputSchema": { "type": "object", "required": ["summary", "start_time"], "properties": { "calendar": { "type": "string", "description": "Which calendar to add to: 'family' (default) or 'regensburg'.", "default": "family", }, "summary": { "type": "string", "description": "Title or summary of the event (e.g. 'U6 Mila Kinderarzt', 'Moritz Zahnarzt', 'Familienausflug').", }, "start_time": { "type": "string", "description": "Start time in ISO format (e.g. '2026-09-20T14:30:00' or '2026-09-20' if all-day).", }, "end_time": { "type": "string", "description": "Optional end time in ISO format. If omitted for timed events, defaults to 1 hour after start_time.", }, "recurrence": { "description": "Optional recurrence rule(s) for repeating events (e.g. ['RRULE:FREQ=WEEKLY;BYDAY=SA'], 'RRULE:FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR,SA;UNTIL=20261231T230000Z', or shortcuts like 'weekly', 'daily', 'monthly', 'workdays').", }, "show_as_busy": { "type": "boolean", "description": "Whether this event marks time as busy. Set False for informational city events, festivals, or notes so they don't block family calendar slots or cause false conflict warnings (default True).", "default": True, }, "transparency": { "type": "string", "description": "Explicit Google transparency: 'opaque' (busy) or 'transparent' (free/available).", }, "reminder_minutes": { "description": "Optional notification reminder minutes before event (e.g. [15, 60] or 1440 for 1 day before). Sends popup alert to phone.", }, "category": { "type": "string", "description": "Optional category/person for color-coding: 'arzt', 'mila', 'moritz', 'amelie', 'familie', 'haushalt'. If omitted, auto-detected from summary.", }, "color_id": { "type": "string", "description": "Explicit Google Calendar colorId (11=Rot/Arzt, 5=Gelb/Mila, 9=Blau/Moritz, 2=Grün/Amelie, 6=Orange/Familie, 8=Grau/Haushalt).", }, "location": { "type": "string", "description": "Optional address or venue (e.g. 'Prüfeninger Str. 10, Regensburg' or 'Westbad Regensburg'). Triggers navigation reminders on mobile.", }, "description": { "type": "string", "description": "Optional notes or checklist (e.g. '📋 Mitnehmen: U-Heft, Impfpass, Wickelsachen').", }, "all_day": { "type": "boolean", "description": "Whether the event lasts all day (default False).", "default": False, }, "inclusive_end": { "type": "boolean", "description": "For all-day events with end_time date, whether the end date is inclusive (default True).", "default": True, }, "check_conflicts": { "type": "boolean", "description": "Whether to check for scheduling overlaps and attach warnings (default True).", "default": True, }, }, }, }, { "name": "calendar_update_event", "description": "Update an existing event or recurring series in the Google Calendar (title, times, recurrence, transparency, reminders, category/color, description, location).", "inputSchema": { "type": "object", "required": ["event_id"], "properties": { "calendar": { "type": "string", "description": "Which calendar the event belongs to ('family' or 'regensburg'). Default 'family'.", "default": "family", }, "event_id": { "type": "string", "description": "The unique ID of the event to update.", }, "summary": { "type": "string", "description": "New title for the event.", }, "start_time": { "type": "string", "description": "New start time in ISO format.", }, "end_time": { "type": "string", "description": "New end time in ISO format.", }, "recurrence": { "description": "Updated recurrence rule(s), or [] / empty string to remove recurrence.", }, "show_as_busy": { "type": "boolean", "description": "Whether this event marks time as busy (True) or free (False).", }, "transparency": { "type": "string", "description": "Explicit transparency: 'opaque' (busy) or 'transparent' (available).", }, "reminder_minutes": { "description": "Updated reminder minutes list before event (e.g. [15, 60]).", }, "update_series": { "type": "boolean", "description": "If True and event is an occurrence of a series, updates the entire master series (default False).", "default": False, }, "category": { "type": "string", "description": "Category for color-coding ('arzt', 'mila', 'moritz', 'amelie', 'familie', 'haushalt').", }, "color_id": { "type": "string", "description": "Explicit colorId.", }, "description": { "type": "string", "description": "New description/notes/checklist.", }, "location": { "type": "string", "description": "New location/address.", }, }, }, }, { "name": "calendar_delete_event", "description": "Delete an event or recurring series from the Google Calendar.", "inputSchema": { "type": "object", "required": ["event_id"], "properties": { "calendar": { "type": "string", "description": "Which calendar the event belongs to ('family' or 'regensburg'). Default 'family'.", "default": "family", }, "event_id": { "type": "string", "description": "The unique ID of the event to delete.", }, "delete_series": { "type": "boolean", "description": "If True and the event is part of a recurring series, deletes the entire recurring series instead of just this single occurrence (default False).", "default": False, }, }, }, }, { "name": "calendar_move_event", "description": "Move an event atomically between calendars (e.g. from 'regensburg' to 'family' or vice versa).", "inputSchema": { "type": "object", "required": ["event_id"], "properties": { "event_id": { "type": "string", "description": "The unique ID of the event to move.", }, "source_calendar": { "type": "string", "description": "Source calendar ('family' or 'regensburg'). Default 'family'.", "default": "family", }, "destination_calendar": { "type": "string", "description": "Destination calendar ('family' or 'regensburg'). Default 'regensburg'.", "default": "regensburg", }, }, }, }, { "name": "calendar_find_free_slots", "description": "Find available free time windows across calendars (family, regensburg, or both) using Google Calendar freebusy query.", "inputSchema": { "type": "object", "properties": { "calendar": { "type": "string", "description": "Which calendar(s) to check: 'family' (default), 'regensburg', or 'both'/'all'.", "default": "family", }, "time_min": { "type": "string", "description": "Start of search timeframe in ISO format (defaults to now).", }, "time_max": { "type": "string", "description": "End of search timeframe in ISO format (defaults to 7 days ahead).", }, "duration_minutes": { "type": "integer", "description": "Minimum length of the free slot required in minutes (default 60).", "default": 60, }, "day_start_hour": { "type": "integer", "description": "Earliest hour of the day to consider (default 8 for 08:00).", "default": 8, }, "day_end_hour": { "type": "integer", "description": "Latest hour of the day to consider (default 20 for 20:00).", "default": 20, }, }, }, }, { "name": "calendar_get_event", "description": "Retrieve full details for a specific event by ID.", "inputSchema": { "type": "object", "required": ["event_id"], "properties": { "calendar": { "type": "string", "description": "Which calendar the event belongs to ('family' or 'regensburg'). Default 'family'.", "default": "family", }, "event_id": { "type": "string", "description": "The unique ID of the event.", }, }, }, }, ] def run_mcp_server(): """Run the stdio MCP JSON-RPC server loop.""" for line in sys.stdin: line = line.strip() if not line: continue try: req = json.loads(line) except Exception: continue method = req.get("method") msg_id = req.get("id") params = req.get("params", {}) if method == "initialize": resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": { "name": "google-calendar-mcp", "version": "1.2.0", }, }, } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() elif method == "notifications/initialized": continue elif method == "tools/list": resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "tools": TOOLS_MANIFEST, }, } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() elif method == "tools/call": tool_name = params.get("name") args = params.get("arguments", {}) try: if tool_name == "calendar_list_events": data = list_events( calendar=args.get("calendar", "family"), time_min=args.get("time_min"), time_max=args.get("time_max"), query=args.get("query"), category=args.get("category"), color_id=args.get("color_id"), max_results=args.get("max_results", 25), ) elif tool_name == "calendar_check_conflicts": data = check_conflicts( calendar=args.get("calendar", "family"), start_time=args.get("start_time", ""), end_time=args.get("end_time"), buffer_minutes=args.get("buffer_minutes", 15), ) elif tool_name == "calendar_create_event": data = create_event( calendar=args.get("calendar", "family"), summary=args.get("summary", ""), start_time=args.get("start_time", ""), end_time=args.get("end_time"), description=args.get("description"), location=args.get("location"), all_day=args.get("all_day", False), category=args.get("category"), color_id=args.get("color_id"), recurrence=args.get("recurrence"), show_as_busy=args.get("show_as_busy"), transparency=args.get("transparency"), reminder_minutes=args.get("reminder_minutes"), inclusive_end=args.get("inclusive_end", True), check_conflicts_flag=args.get("check_conflicts", True), ) elif tool_name == "calendar_update_event": data = update_event( calendar=args.get("calendar", "family"), event_id=args.get("event_id", ""), summary=args.get("summary"), start_time=args.get("start_time"), end_time=args.get("end_time"), description=args.get("description"), location=args.get("location"), category=args.get("category"), color_id=args.get("color_id"), recurrence=args.get("recurrence"), show_as_busy=args.get("show_as_busy"), transparency=args.get("transparency"), reminder_minutes=args.get("reminder_minutes"), inclusive_end=args.get("inclusive_end", True), update_series=args.get("update_series", False), ) elif tool_name == "calendar_delete_event": data = delete_event( calendar=args.get("calendar", "family"), event_id=args.get("event_id", ""), delete_series=args.get("delete_series", False), ) elif tool_name == "calendar_move_event": data = move_event( event_id=args.get("event_id", ""), source_calendar=args.get("source_calendar", "family"), destination_calendar=args.get("destination_calendar", "regensburg"), ) elif tool_name == "calendar_find_free_slots": data = find_free_slots( calendar=args.get("calendar", "family"), time_min=args.get("time_min"), time_max=args.get("time_max"), duration_minutes=args.get("duration_minutes", 60), day_start_hour=args.get("day_start_hour", 8), day_end_hour=args.get("day_end_hour", 20), ) elif tool_name == "calendar_get_event": data = get_event( calendar=args.get("calendar", "family"), event_id=args.get("event_id", ""), ) else: raise ValueError(f"Unknown tool: {tool_name}") text_result = json.dumps(data, indent=2, ensure_ascii=False) resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "content": [{"type": "text", "text": text_result}], }, } except Exception as e: resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "isError": True, "content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}], }, } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() else: if msg_id is not None: resp = { "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() # ============================================================================== # CLI Entry Point # ============================================================================== def run_cli(): parser = argparse.ArgumentParser(description="Google Calendar CLI & MCP Tool") subparsers = parser.add_subparsers(dest="command") # Test subparsers.add_parser("test", help="Test connectivity and check accessible calendars") # List sp_list = subparsers.add_parser("list", help="List events") sp_list.add_argument("--calendar", "-c", default="family", help="'family', 'regensburg', or Calendar ID") sp_list.add_argument("--min", dest="time_min", help="ISO start time") sp_list.add_argument("--max", dest="time_max", help="ISO end time") sp_list.add_argument("--query", "-q", help="Search text query") sp_list.add_argument("--category", help="Category filter") sp_list.add_argument("--color", dest="color_id", help="Color ID filter") sp_list.add_argument("--limit", "-n", type=int, default=25, help="Max results") # Conflicts sp_conf = subparsers.add_parser("conflicts", help="Check conflicts") sp_conf.add_argument("start_time", help="Start time") sp_conf.add_argument("--end", dest="end_time", help="End time") sp_conf.add_argument("--calendar", "-c", default="family") sp_conf.add_argument("--buffer", type=int, default=15, help="Buffer minutes") # Free slots sp_slots = subparsers.add_parser("free-slots", help="Find free time windows") sp_slots.add_argument("--calendar", "-c", default="family", help="'family', 'regensburg', or 'both'") sp_slots.add_argument("--min", dest="time_min", help="ISO start time") sp_slots.add_argument("--max", dest="time_max", help="ISO end time") sp_slots.add_argument("--duration", type=int, default=60, help="Required duration in minutes") # Create sp_create = subparsers.add_parser("create", help="Create event") sp_create.add_argument("summary", help="Event summary/title") sp_create.add_argument("start_time", help="Start time (ISO or YYYY-MM-DD)") sp_create.add_argument("--end", dest="end_time", help="End time") sp_create.add_argument("--calendar", "-c", default="family", help="'family' or 'regensburg'") sp_create.add_argument("--category", help="Category: arzt, mila, moritz, amelie, familie, haushalt") sp_create.add_argument("--recurrence", "-r", help="RRULE recurrence (e.g. 'RRULE:FREQ=WEEKLY;BYDAY=SA' or 'weekly')") sp_create.add_argument("--free", "--transparent", dest="transparent", action="store_true", help="Mark event as free/transparent") sp_create.add_argument("--reminder", dest="reminder_minutes", type=int, action="append", help="Reminder in minutes before event") sp_create.add_argument("--description", "-d", help="Description") sp_create.add_argument("--location", "-l", help="Location") sp_create.add_argument("--all-day", action="store_true", help="All day event") # Move sp_move = subparsers.add_parser("move", help="Move event between calendars") sp_move.add_argument("event_id", help="Event ID") sp_move.add_argument("--from", dest="source_calendar", default="family", help="Source calendar") sp_move.add_argument("--to", dest="destination_calendar", default="regensburg", help="Destination calendar") # Delete sp_del = subparsers.add_parser("delete", help="Delete event") sp_del.add_argument("event_id", help="Event ID") sp_del.add_argument("--calendar", "-c", default="family", help="'family' or 'regensburg'") sp_del.add_argument("--series", action="store_true", help="Delete entire recurring series") args = parser.parse_args() if args.command == "test": print(json.dumps(test_connection(), indent=2, ensure_ascii=False)) elif args.command == "conflicts": print(json.dumps(check_conflicts( calendar=args.calendar, start_time=args.start_time, end_time=args.end_time, buffer_minutes=args.buffer, ), indent=2, ensure_ascii=False)) elif args.command == "free-slots": print(json.dumps(find_free_slots( calendar=args.calendar, time_min=args.time_min, time_max=args.time_max, duration_minutes=args.duration, ), indent=2, ensure_ascii=False)) elif args.command == "list": print(json.dumps(list_events( calendar=args.calendar, time_min=args.time_min, time_max=args.time_max, query=args.query, category=args.category, color_id=args.color_id, max_results=args.limit, ), indent=2, ensure_ascii=False)) elif args.command == "create": print(json.dumps(create_event( calendar=args.calendar, summary=args.summary, start_time=args.start_time, end_time=args.end_time, category=args.category, recurrence=args.recurrence, show_as_busy=not args.transparent, reminder_minutes=args.reminder_minutes, description=args.description, location=args.location, all_day=args.all_day, ), indent=2, ensure_ascii=False)) elif args.command == "move": print(json.dumps(move_event( event_id=args.event_id, source_calendar=args.source_calendar, destination_calendar=args.destination_calendar, ), indent=2, ensure_ascii=False)) elif args.command == "delete": print(json.dumps(delete_event( calendar=args.calendar, event_id=args.event_id, delete_series=args.series, ), indent=2, ensure_ascii=False)) else: parser.print_help() if __name__ == "__main__": if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"): run_mcp_server() else: run_cli()