1547 lines
59 KiB
Python
Executable File
1547 lines
59 KiB
Python
Executable File
#!/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()
|