moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(calendar): implement visual color-coding, conflict-guard, and prep checklists

Browse Source
This commit is contained in:
Moritz Graf
2026-09-15 22:28:26 +02:00
parent dc874415b4
commit 7e82424ca2
2 changed files with 384 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/google-calendar/scripts/google_calendar.py
+331 -29
View File
@@ -3,6 +3,10 @@
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
@@ -10,7 +14,7 @@ import sys
import json
import argparse
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional
from typing import Dict, Any, List, Optional, Tuple
from zoneinfo import ZoneInfo
# Environment Configuration
@@ -22,6 +26,104 @@ 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."""
@@ -29,7 +131,6 @@ def get_credentials():
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)
@@ -39,12 +140,10 @@ def get_credentials():
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",
@@ -72,16 +171,12 @@ def get_calendar_service():
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.
"""
"""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:
# 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."
@@ -96,7 +191,6 @@ def resolve_calendar_id(calendar_identifier: Optional[str] = None) -> str:
)
return CALENDAR_ID_REGENSBURG
# Assume direct calendar ID
return ident
@@ -112,7 +206,7 @@ def parse_datetime_str(dt_str: str) -> datetime:
def format_event(item: Dict[str, Any], calendar_name: str) -> Dict[str, Any]:
"""Format an event item into a clean representation."""
"""Format an event item into a clean representation including color labels."""
start = item.get("start", {})
end = item.get("end", {})
@@ -120,6 +214,9 @@ def format_event(item: Dict[str, Any], calendar_name: str) -> Dict[str, Any]:
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,
@@ -127,6 +224,8 @@ def format_event(item: Dict[str, Any], calendar_name: str) -> Dict[str, Any]:
"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"),
@@ -134,6 +233,100 @@ def format_event(item: Dict[str, Any], calendar_name: str) -> Dict[str, Any]:
}
# ==============================================================================
# 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
# ==============================================================================
@@ -149,7 +342,6 @@ def list_events(
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)
@@ -196,8 +388,15 @@ def create_event(
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."""
"""
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:
@@ -206,6 +405,20 @@ def create_event(
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,
}
@@ -213,14 +426,14 @@ def create_event(
body["description"] = description
if location:
body["location"] = location
if chosen_color_id:
body["colorId"] = chosen_color_id
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()
@@ -243,11 +456,17 @@ def create_event(
}
created = service.events().insert(calendarId=cal_id, body=body).execute()
return {
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",
@@ -257,6 +476,8 @@ def update_event(
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:
@@ -265,7 +486,6 @@ def update_event(
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:
@@ -275,6 +495,10 @@ def update_event(
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]
@@ -369,6 +593,7 @@ def test_connection() -> Dict[str, Any]:
"configured_regensburg_id": CALENDAR_ID_REGENSBURG or "(empty)",
"regensburg_calendar_check": regensburg_status,
"accessible_calendars": calendars,
"color_palette": COLOR_NAMES,
}
@@ -379,13 +604,13 @@ def test_connection() -> Dict[str, Any]:
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.",
"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 and activities.",
"description": "Which calendar to query: 'family' (default) for family appointments, or 'regensburg' for city/local events.",
"default": "family",
},
"time_min": {
@@ -408,21 +633,49 @@ TOOLS_MANIFEST = [
},
},
},
{
"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 event or appointment in the Google Calendar.",
"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) for family schedule, or 'regensburg' for local events.",
"description": "Which calendar to add to: 'family' (default) or 'regensburg'.",
"default": "family",
},
"summary": {
"type": "string",
"description": "Title or summary of the event (e.g. 'Kinderarzt U6 Termin', 'Bürgerfest Regensburg').",
"description": "Title or summary of the event (e.g. 'U6 Mila Kinderarzt', 'Moritz Zahnarzt', 'Familienausflug').",
},
"start_time": {
"type": "string",
@@ -432,25 +685,38 @@ TOOLS_MANIFEST = [
"type": "string",
"description": "Optional end time in ISO format. If omitted for timed events, defaults to 1 hour after start_time.",
},
"description": {
"category": {
"type": "string",
"description": "Optional notes or details for the event.",
"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 location or address.",
"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.",
"description": "Update an existing event in the Google Calendar (title, times, category/color, description, location).",
"inputSchema": {
"type": "object",
"required": ["event_id"],
@@ -476,13 +742,21 @@ TOOLS_MANIFEST = [
"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.",
"description": "New description/notes/checklist.",
},
"location": {
"type": "string",
"description": "New location.",
"description": "New location/address.",
},
},
},
@@ -553,7 +827,7 @@ def run_mcp_server():
"capabilities": {"tools": {}},
"serverInfo": {
"name": "google-calendar-mcp",
"version": "1.0.0",
"version": "1.1.0",
},
},
}
@@ -586,6 +860,13 @@ def run_mcp_server():
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"),
@@ -595,6 +876,9 @@ def run_mcp_server():
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(
@@ -605,6 +889,8 @@ def run_mcp_server():
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(
@@ -669,12 +955,20 @@ def run_cli():
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")
@@ -688,6 +982,13 @@ def run_cli():
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,
@@ -702,6 +1003,7 @@ def run_cli():
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,