Stefan Lohmaier
924366ac6c
- travel_minutes: shows as gray block before event on iPhone - location geocoding via Nominatim → X-APPLE-STRUCTURED-LOCATION with geo coords - reminder_minutes: VALARM trigger before event - Also externalized all credentials to config.json Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
305 lines
16 KiB
Python
305 lines
16 KiB
Python
"""MCP Calendar + Tasks Server — reads and writes calendars and tasks via CalDAV."""
|
|
|
|
import os, sys, re, contextlib, uuid
|
|
from datetime import datetime, timedelta
|
|
from typing import Annotated
|
|
|
|
import httpx, vobject
|
|
from pydantic import Field
|
|
from mcp.server.fastmcp import FastMCP
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Mount
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
from common import get_current_user, OAUTH_ROUTES, BearerAuthMiddleware
|
|
|
|
from common import load_config as _lc
|
|
_cfg = _lc()
|
|
RADICALE = _cfg['radicale_url']
|
|
CREDS = {u: (d['username'], d['password']) for u, d in _cfg['radicale_users'].items()}
|
|
CAL_PATHS = _cfg['calendar_paths']
|
|
|
|
mcp = FastMCP("Calendar", stateless_http=True,
|
|
transport_security={"enable_dns_rebinding_protection": False})
|
|
|
|
|
|
def _auth(user):
|
|
c = CREDS.get(user)
|
|
return httpx.BasicAuth(c[0], c[1]) if c else None
|
|
|
|
def _propfind(url, auth, depth=1):
|
|
body = '<?xml version="1.0"?><d:propfind xmlns:d="DAV:" xmlns:cs="urn:ietf:params:xml:ns:caldav"><d:prop><d:resourcetype/><d:displayname/><cs:supported-calendar-component-set/></d:prop></d:propfind>'
|
|
return httpx.request("PROPFIND", url, content=body, auth=auth, headers={"Depth": str(depth), "Content-Type": "application/xml"}, timeout=30).text
|
|
|
|
def _discover(user, comp):
|
|
auth = _auth(user)
|
|
cols = []
|
|
for base in CAL_PATHS.get(user, []):
|
|
xml = _propfind(RADICALE + base, auth)
|
|
for m in re.finditer(r'<d:href>([^<]+)</d:href>', xml):
|
|
href = m.group(1)
|
|
if href.rstrip("/") == base.rstrip("/"): continue
|
|
block = xml[m.start():xml.find("</d:response>", m.end())]
|
|
if comp not in block: continue
|
|
nm = re.search(r'<d:displayname>([^<]*)</d:displayname>', block)
|
|
cols.append({"name": nm.group(1) if nm else href.split("/")[-2], "href": href})
|
|
return cols
|
|
|
|
def _report(href, auth, start, end):
|
|
body = f'<?xml version="1.0"?><c:calendar-query xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav"><d:prop><d:getetag/><c:calendar-data/></d:prop><c:filter><c:comp-filter name="VCALENDAR"><c:comp-filter name="VEVENT"><c:time-range start="{start}" end="{end}"/></c:comp-filter></c:comp-filter></c:filter></c:calendar-query>'
|
|
return httpx.request("REPORT", RADICALE + href, content=body, auth=auth, headers={"Depth": "1", "Content-Type": "application/xml"}, timeout=30).text
|
|
|
|
def _report_tasks(href, auth, inc=False):
|
|
filt = '<c:comp-filter name="VTODO"/>' if inc else '<c:comp-filter name="VTODO"><c:prop-filter name="STATUS"><c:text-match negate-condition="yes">COMPLETED</c:text-match></c:prop-filter></c:comp-filter>'
|
|
body = f'<?xml version="1.0"?><c:calendar-query xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav"><d:prop><d:getetag/><c:calendar-data/></d:prop><c:filter><c:comp-filter name="VCALENDAR">{filt}</c:comp-filter></c:filter></c:calendar-query>'
|
|
return httpx.request("REPORT", RADICALE + href, content=body, auth=auth, headers={"Depth": "1", "Content-Type": "application/xml"}, timeout=30).text
|
|
|
|
def _parse(xml, comp="VEVENT"):
|
|
objs = []
|
|
for m in re.finditer(r'<c:calendar-data[^>]*>(.*?)</c:calendar-data>', xml, re.DOTALL):
|
|
raw = m.group(1).replace("<","<").replace(">",">").replace("&","&")
|
|
try:
|
|
for c in vobject.readOne(raw).components():
|
|
if c.name == comp: objs.append(c)
|
|
except: pass
|
|
return objs
|
|
|
|
def _fmt_ev(ev, cal=""):
|
|
s = getattr(ev, 'dtstart', None)
|
|
e = getattr(ev, 'dtend', None)
|
|
summary = str(ev.summary.value) if hasattr(ev, 'summary') else '(kein Titel)'
|
|
start = s.value.isoformat() if s and hasattr(s.value, 'isoformat') else str(s.value) if s else '?'
|
|
end_s = e.value.isoformat() if e and hasattr(e.value, 'isoformat') else ''
|
|
loc = str(ev.location.value) if hasattr(ev, 'location') and ev.location.value else ''
|
|
desc = str(ev.description.value)[:300] if hasattr(ev, 'description') and ev.description.value else ''
|
|
uid = str(ev.uid.value) if hasattr(ev, 'uid') else ''
|
|
status = str(ev.status.value) if hasattr(ev, 'status') else ''
|
|
attendees = []
|
|
if hasattr(ev, 'attendee_list'):
|
|
for att in ev.attendee_list:
|
|
cn = att.params.get("CN", [""])[0] if hasattr(att, 'params') and att.params else ""
|
|
ps = att.params.get("PARTSTAT", [""])[0] if hasattr(att, 'params') and att.params else ""
|
|
attendees.append(f"{cn} ({ps})" if cn else str(att.value))
|
|
rrule = str(ev.rrule.value) if hasattr(ev, 'rrule') else ''
|
|
parts = [f"[{cal}] {summary}" if cal else summary]
|
|
parts.append(f" Wann: {start}" + (f" bis {end_s}" if end_s else ""))
|
|
if loc: parts.append(f" Ort: {loc}")
|
|
if desc: parts.append(f" Beschreibung: {desc}")
|
|
if status: parts.append(f" Status: {status}")
|
|
if attendees: parts.append(f" Teilnehmer: {', '.join(attendees[:8])}")
|
|
if rrule: parts.append(f" Wiederholung: {rrule}")
|
|
parts.append(f" UID: {uid}")
|
|
return "\n".join(parts)
|
|
|
|
def _fmt_task(t, lst=""):
|
|
summary = str(t.summary.value) if hasattr(t, 'summary') else '(kein Titel)'
|
|
status = str(t.status.value) if hasattr(t, 'status') else 'NEEDS-ACTION'
|
|
due = t.due.value.isoformat() if hasattr(t, 'due') and hasattr(t.due.value, 'isoformat') else str(t.due.value) if hasattr(t, 'due') else ''
|
|
pri = int(t.priority.value) if hasattr(t, 'priority') else 0
|
|
uid = str(t.uid.value) if hasattr(t, 'uid') else ''
|
|
desc = str(t.description.value)[:200] if hasattr(t, 'description') and t.description.value else ''
|
|
chk = '[x]' if status == 'COMPLETED' else '[ ]'
|
|
parts = [f"[{lst}] {chk} {summary}" if lst else f"{chk} {summary}"]
|
|
if due: parts.append(f" Faellig: {due}")
|
|
if pri: parts.append(f" Prioritaet: {pri} (1=hoechste, 9=niedrigste)")
|
|
if desc: parts.append(f" Notiz: {desc}")
|
|
parts.append(f" UID: {uid}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
@mcp.tool()
|
|
def list_calendars() -> str:
|
|
"""List all calendars available to the user (personal + shared). Call this first to see calendar names."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
cals = _discover(user, "VEVENT")
|
|
return "\n".join(f"{c['name']}" for c in cals) if cals else "Keine Kalender gefunden"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_events(
|
|
calendar: Annotated[str, Field(description="Calendar name from list_calendars, e.g. 'Stefan', 'Arbeit', 'Familie'. Use partial match.")],
|
|
date_from: Annotated[str, Field(description="Start date as YYYY-MM-DD, e.g. '2026-06-12'")],
|
|
date_to: Annotated[str, Field(description="End date as YYYY-MM-DD, e.g. '2026-06-19'")],
|
|
) -> str:
|
|
"""Get all events from a calendar within a date range. Returns title, time, location, attendees, and UID."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
cals = [c for c in _discover(user, "VEVENT") if calendar.lower() in c["name"].lower()]
|
|
if not cals: return f"Kalender '{calendar}' nicht gefunden. Nutze list_calendars."
|
|
s = datetime.strptime(date_from, "%Y-%m-%d").strftime("%Y%m%dT000000Z")
|
|
e = datetime.strptime(date_to, "%Y-%m-%d").strftime("%Y%m%dT235959Z")
|
|
results = []
|
|
for cal in cals:
|
|
for ev in _parse(_report(cal["href"], _auth(user), s, e)):
|
|
results.append(_fmt_ev(ev, cal["name"]))
|
|
return "\n\n".join(results) if results else "Keine Termine in diesem Zeitraum"
|
|
|
|
|
|
@mcp.tool()
|
|
def search_events(
|
|
query: Annotated[str, Field(description="Search term — matches title, description, location. Example: 'Zahnarzt', 'Meeting', 'Berlin'")],
|
|
date_from: Annotated[str, Field(description="Start date YYYY-MM-DD. Default: 90 days ago")] = "",
|
|
date_to: Annotated[str, Field(description="End date YYYY-MM-DD. Default: 1 year ahead")] = "",
|
|
) -> str:
|
|
"""Search events by keyword across all calendars. Returns matching events with details."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
if not date_from: date_from = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d")
|
|
if not date_to: date_to = (datetime.now() + timedelta(days=365)).strftime("%Y-%m-%d")
|
|
s = datetime.strptime(date_from, "%Y-%m-%d").strftime("%Y%m%dT000000Z")
|
|
e = datetime.strptime(date_to, "%Y-%m-%d").strftime("%Y%m%dT235959Z")
|
|
q = query.lower()
|
|
results = []
|
|
for cal in _discover(user, "VEVENT"):
|
|
for ev in _parse(_report(cal["href"], _auth(user), s, e)):
|
|
txt = f"{ev.summary.value if hasattr(ev,'summary') else ''} {ev.description.value if hasattr(ev,'description') else ''} {ev.location.value if hasattr(ev,'location') else ''}".lower()
|
|
if q in txt:
|
|
results.append(_fmt_ev(ev, cal["name"]))
|
|
return "\n\n".join(results[:30]) if results else "Keine Treffer"
|
|
|
|
|
|
@mcp.tool()
|
|
def list_task_lists() -> str:
|
|
"""List all task lists (Erinnerungen/Reminders). Call this first before get_tasks."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
lists = _discover(user, "VTODO")
|
|
return "\n".join(f"{l['name']}" for l in lists) if lists else "Keine Aufgabenlisten"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_tasks(
|
|
task_list: Annotated[str, Field(description="Task list name from list_task_lists, e.g. 'Erinnerungen'. Leave empty for all lists.")] = "",
|
|
include_completed: Annotated[bool, Field(description="Also show completed tasks")] = False,
|
|
) -> str:
|
|
"""Get tasks/reminders from a task list. Shows title, due date, priority, and status."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
lists = _discover(user, "VTODO")
|
|
if task_list: lists = [l for l in lists if task_list.lower() in l["name"].lower()]
|
|
results = []
|
|
for lst in lists:
|
|
for t in _parse(_report_tasks(lst["href"], _auth(user), include_completed), "VTODO"):
|
|
results.append(_fmt_task(t, lst["name"]))
|
|
return "\n\n".join(results) if results else "Keine Aufgaben"
|
|
|
|
|
|
def _geocode(address):
|
|
"""Geocode an address via Nominatim. Returns (lat, lon, display_name) or None."""
|
|
try:
|
|
r = httpx.get("https://nominatim.openstreetmap.org/search",
|
|
params={"q": address, "format": "json", "limit": 1},
|
|
headers={"User-Agent": "mcp-home/1.0"}, timeout=10)
|
|
results = r.json()
|
|
if results:
|
|
return float(results[0]["lat"]), float(results[0]["lon"]), results[0].get("display_name", address)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
@mcp.tool()
|
|
def create_event(
|
|
calendar: Annotated[str, Field(description="Calendar name, e.g. 'Stefan', 'Arbeit', 'Familie'")],
|
|
title: Annotated[str, Field(description="Event title")],
|
|
start: Annotated[str, Field(description="Start: YYYY-MM-DD for all-day, or YYYY-MM-DDTHH:MM for timed events, e.g. '2026-06-15T14:00'")],
|
|
end: Annotated[str, Field(description="End: YYYY-MM-DD for all-day, or YYYY-MM-DDTHH:MM, e.g. '2026-06-15T15:30'")],
|
|
location: Annotated[str, Field(description="Location or address, e.g. 'Marienplatz 1, Muenchen'. Gets geocoded for map display.")] = "",
|
|
description: Annotated[str, Field(description="Event description or notes")] = "",
|
|
allday: Annotated[bool, Field(description="True for all-day events (then use YYYY-MM-DD for start/end)")] = False,
|
|
travel_minutes: Annotated[int, Field(description="Travel time in minutes shown before the event on iPhone (0=none, e.g. 30 for 30 min commute)")] = 0,
|
|
reminder_minutes: Annotated[int, Field(description="Reminder alert X minutes before the event (0=none, e.g. 15, 30, 60)")] = 0,
|
|
) -> str:
|
|
"""Create a new calendar event. For all-day events, end date is exclusive (June 15 only → start=2026-06-15, end=2026-06-16). Location is geocoded for map pins. Travel time shows as block before the event on iPhone."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
cals = [c for c in _discover(user, "VEVENT") if calendar.lower() in c["name"].lower()]
|
|
if not cals: return f"Kalender '{calendar}' nicht gefunden"
|
|
uid = str(uuid.uuid4())
|
|
cal = vobject.iCalendar()
|
|
ev = cal.add("vevent")
|
|
ev.add("uid").value = uid
|
|
ev.add("summary").value = title
|
|
if allday:
|
|
ev.add("dtstart").value = datetime.strptime(start, "%Y-%m-%d").date()
|
|
ev.add("dtend").value = datetime.strptime(end, "%Y-%m-%d").date()
|
|
else:
|
|
ev.add("dtstart").value = datetime.fromisoformat(start)
|
|
ev.add("dtend").value = datetime.fromisoformat(end)
|
|
if location:
|
|
ev.add("location").value = location
|
|
geo = _geocode(location)
|
|
if geo:
|
|
lat, lon, display = geo
|
|
loc_prop = ev.add("x-apple-structured-location")
|
|
loc_prop.value = f"geo:{lat},{lon}"
|
|
loc_prop.params["VALUE"] = ["URI"]
|
|
loc_prop.params["X-APPLE-RADIUS"] = ["70"]
|
|
loc_prop.params["X-APPLE-REFERENCEFRAME"] = ["1"]
|
|
loc_prop.params["X-TITLE"] = [location]
|
|
if description:
|
|
ev.add("description").value = description
|
|
if travel_minutes > 0:
|
|
tp = ev.add("x-apple-travel-duration")
|
|
tp.value = f"PT{travel_minutes}M"
|
|
tp.params["VALUE"] = ["DURATION"]
|
|
if reminder_minutes > 0:
|
|
alarm = ev.add("valarm")
|
|
alarm.add("action").value = "DISPLAY"
|
|
alarm.add("description").value = title
|
|
alarm.add("trigger").value = timedelta(minutes=-reminder_minutes)
|
|
ev.add("dtstamp").value = datetime.utcnow()
|
|
r = httpx.put(RADICALE + cals[0]["href"] + uid + ".ics", content=cal.serialize(), auth=_auth(user), headers={"Content-Type": "text/calendar"}, timeout=15)
|
|
extras = []
|
|
if location and _geocode(location): extras.append("mit Karte")
|
|
if travel_minutes: extras.append(f"{travel_minutes} Min. Fahrzeit")
|
|
if reminder_minutes: extras.append(f"Erinnerung {reminder_minutes} Min. vorher")
|
|
extra_str = f" ({', '.join(extras)})" if extras else ""
|
|
return f"Termin erstellt: {title} am {start}{extra_str}" if r.status_code in (201, 204) else f"Fehler: HTTP {r.status_code}"
|
|
|
|
|
|
@mcp.tool()
|
|
def create_task(
|
|
task_list: Annotated[str, Field(description="Task list name, e.g. 'Erinnerungen', 'Langzeiterinnerungen'")],
|
|
title: Annotated[str, Field(description="Task title")],
|
|
due: Annotated[str, Field(description="Due date: YYYY-MM-DD or YYYY-MM-DDTHH:MM. Leave empty for no due date.")] = "",
|
|
priority: Annotated[int, Field(description="Priority 1-9 (1=highest, 9=lowest, 0=none)")] = 0,
|
|
description: Annotated[str, Field(description="Task notes")] = "",
|
|
) -> str:
|
|
"""Create a new task/reminder in the specified list."""
|
|
user = get_current_user()
|
|
if not user: return "Error: not authenticated"
|
|
lists = [l for l in _discover(user, "VTODO") if task_list.lower() in l["name"].lower()]
|
|
if not lists: return f"Liste '{task_list}' nicht gefunden"
|
|
uid = str(uuid.uuid4())
|
|
cal = vobject.iCalendar()
|
|
todo = cal.add("vtodo")
|
|
todo.add("uid").value = uid
|
|
todo.add("summary").value = title
|
|
todo.add("status").value = "NEEDS-ACTION"
|
|
if due: todo.add("due").value = datetime.fromisoformat(due) if "T" in due else datetime.strptime(due, "%Y-%m-%d").date()
|
|
if priority: todo.add("priority").value = str(priority)
|
|
if description: todo.add("description").value = description
|
|
todo.add("dtstamp").value = datetime.utcnow()
|
|
r = httpx.put(RADICALE + lists[0]["href"] + uid + ".ics", content=cal.serialize(), auth=_auth(user), headers={"Content-Type": "text/calendar"}, timeout=15)
|
|
return f"Aufgabe erstellt: {title}" if r.status_code in (201, 204) else f"Fehler: HTTP {r.status_code}"
|
|
|
|
|
|
def create_app():
|
|
from contextlib import asynccontextmanager
|
|
mcp_app = mcp.streamable_http_app()
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
async with contextlib.AsyncExitStack() as stack:
|
|
await stack.enter_async_context(mcp_app.router.lifespan_context(mcp_app))
|
|
yield
|
|
routes = list(OAUTH_ROUTES) + [Mount("/", app=mcp_app)]
|
|
app = Starlette(routes=routes, lifespan=lifespan)
|
|
app.add_middleware(BearerAuthMiddleware)
|
|
return app
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(create_app(), host="127.0.0.1", port=5101)
|