#!/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. """ import os import sys import json import argparse from datetime import datetime, date, timedelta from typing import Dict, Any, List, Optional 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"] 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: # Check if it is a JSON string or a file path 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) # Fallback: Check GOOGLE_APPLICATION_CREDENTIALS 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) # Fallback: Check local secret files in repo 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. If a full calendar ID (containing '@') is passed, returns it as-is. """ ident = (calendar_identifier or "family").strip() ident_lower = ident.lower() if ident_lower in ("family", "familie", "familienkalender", "default", "primary"): if not CALENDAR_ID_FAMILY: # If not configured in env, check if it's in a shared calendar list 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( "CALENDAR_ID_REGENSBURG is not configured. Please set the environment variable " "CALENDAR_ID_REGENSBURG to the Google Calendar ID of the Regensburg calendar." ) return CALENDAR_ID_REGENSBURG # Assume direct calendar ID 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.""" 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") 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, "description": item.get("description", ""), "location": item.get("location", ""), "status": item.get("status", "confirmed"), "htmlLink": item.get("htmlLink", ""), } # ============================================================================== # 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) # Defaults: from start of current day to 30 days ahead 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, ) -> Dict[str, Any]: """Create a new event in the specified calendar.""" 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) body: Dict[str, Any] = { "summary": summary, } if description: body["description"] = description if location: body["location"] = location if all_day: # Expected format YYYY-MM-DD start_date = start_time.split("T")[0] if end_time: end_date = end_time.split("T")[0] else: # Single-day all-day event in Google Calendar ends on the NEXT day 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() return { "status": "created", "event": format_event(created, calendar), } 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, ) -> 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) # Fetch existing 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 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, } # ============================================================================== # MCP Protocol Implementation (stdio JSON-RPC) # ============================================================================== TOOLS_MANIFEST = [ { "name": "calendar_list_events", "description": "List upcoming appointments and events from the Google Calendar. Supports filtering by calendar ('family' or 'regensburg'), date range, and search query.", "inputSchema": { "type": "object", "properties": { "calendar": { "type": "string", "description": "Which calendar to query: 'family' (default) for family appointments, or 'regensburg' for city/local events and activities.", "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_create_event", "description": "Create a new event or appointment in the Google Calendar.", "inputSchema": { "type": "object", "required": ["summary", "start_time"], "properties": { "calendar": { "type": "string", "description": "Which calendar to add to: 'family' (default) for family schedule, or 'regensburg' for local events.", "default": "family", }, "summary": { "type": "string", "description": "Title or summary of the event (e.g. 'Kinderarzt U6 Termin', 'Bürgerfest Regensburg').", }, "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.", }, "description": { "type": "string", "description": "Optional notes or details for the event.", }, "location": { "type": "string", "description": "Optional location or address.", }, "all_day": { "type": "boolean", "description": "Whether the event lasts all day (default False).", "default": False, }, }, }, }, { "name": "calendar_update_event", "description": "Update an existing event in 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 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.", }, "description": { "type": "string", "description": "New description/notes.", }, "location": { "type": "string", "description": "New location.", }, }, }, }, { "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.0.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_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), ) 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"), ) 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") # 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("--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 == "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, 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()