#!/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 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 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 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, "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, max_results: int = 25, ) -> Dict[str, Any]: """List events from the specified calendar.""" 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 = { "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", []) 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, "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, check_conflicts_flag: bool = True, buffer_minutes: int = 15, ) -> Dict[str, Any]: """ Create a new event in the specified calendar with automatic color-coding and conflict detection. """ 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 if all_day: start_date = start_time.split("T")[0] if end_time: end_date = end_time.split("T")[0] else: dt_start = date.fromisoformat(start_date) 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, ) -> Dict[str, Any]: """Update an existing event.""" if not event_id: raise ValueError("Parameter 'event_id' is required.") service = get_calendar_service() cal_id = resolve_calendar_id(calendar) existing = service.events().get(calendarId=cal_id, eventId=event_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 start_time is not None: if "date" in existing.get("start", {}): existing["start"]["date"] = start_time.split("T")[0] if end_time: existing["end"]["date"] = end_time.split("T")[0] 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=event_id, body=existing).execute() return { "status": "updated", "event": format_event(updated, calendar), } def delete_event(calendar: str = "family", event_id: str = "") -> Dict[str, Any]: """Delete an event 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) service.events().delete(calendarId=cal_id, eventId=event_id).execute() return { "status": "deleted", "calendar": calendar, "event_id": event_id, } 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, 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.", }, "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), address location, prep 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.", }, "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, }, "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 in the Google Calendar (title, times, 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.", }, "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 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.", }, }, }, }, { "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.1.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"), 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"), 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"), ) elif tool_name == "calendar_delete_event": data = delete_event( calendar=args.get("calendar", "family"), event_id=args.get("event_id", ""), ) 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("--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") # 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("--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") # 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'") 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 == "list": print(json.dumps(list_events( calendar=args.calendar, time_min=args.time_min, time_max=args.time_max, query=args.query, 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, description=args.description, location=args.location, all_day=args.all_day, ), indent=2, ensure_ascii=False)) elif args.command == "delete": print(json.dumps(delete_event( calendar=args.calendar, event_id=args.event_id, ), 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()