feat: Implement a CalDAV client and integrate it as tools within the FastMCP server for calendar and event management.

src/caldav_client.py

@@ -0,0 +1,121 @@
+import os
+import caldav
+from datetime import datetime
+from typing import List, Optional, Dict, Any
+from dotenv import load_dotenv
+import icalendar
+
+load_dotenv()
+
+class CalDAVClient:
+    def __init__(self):
+        self.url = os.getenv("CALDAV_URL")
+        self.username = os.getenv("CALDAV_USERNAME")
+        self.password = os.getenv("CALDAV_PASSWORD")
+        if not all([self.url, self.username, self.password]):
+            raise ValueError("Missing CalDAV credentials in .env")
+        
+        self.client = caldav.DAVClient(
+            url=self.url,
+            username=self.username,
+            password=self.password
+        )
+        self.principal = self.client.principal()
+
+    def list_calendars(self) -> List[Dict[str, str]]:
+        calendars = self.principal.calendars()
+        return [{"name": c.name, "url": str(c.url)} for c in calendars]
+
+    def _get_calendar(self, name: str):
+        calendars = self.principal.calendars()
+        for c in calendars:
+            if c.name == name:
+                return c
+        raise ValueError(f"Calendar '{name}' not found")
+
+    def list_events(self, calendar_name: Optional[str] = None, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None) -> List[Dict[str, Any]]:
+        if calendar_name:
+            calendars = [self._get_calendar(calendar_name)]
+        else:
+            calendars = self.principal.calendars()
+
+        results = []
+        for calendar in calendars:
+            if start_date and end_date:
+                events = calendar.date_search(start=start_date, end=end_date, expand=True)
+            else:
+                events = calendar.events()
+
+            for event in events:
+                # event.data is the raw ical data
+                ical = icalendar.Calendar.from_ical(event.data)
+                for component in ical.walk():
+                    if component.name == "VEVENT":
+                        summary = component.get("summary")
+                        description = component.get("description")
+                        dtstart = component.get("dtstart").dt
+                        dtend = component.get("dtend").dt if component.get("dtend") else None
+                        uid = component.get("uid")
+                        
+                        # Handle datetime/date objects (convert to isoformat string)
+                        start_str = dtstart.isoformat() if hasattr(dtstart, 'isoformat') else str(dtstart)
+                        end_str = dtend.isoformat() if dtend and hasattr(dtend, 'isoformat') else str(dtend) if dtend else None
+
+                        results.append({
+                            "calendar": calendar.name,
+                            "uid": str(uid),
+                            "summary": str(summary) if summary else "No Title",
+                            "description": str(description) if description else "",
+                            "start": start_str,
+                            "end": end_str
+                        })
+        return results
+
+    def create_event(self, calendar_name: str, summary: str, start_time: datetime, end_time: datetime, description: str = "") -> str:
+        calendar = self._get_calendar(calendar_name)
+        event = calendar.save_event(
+            dtstart=start_time,
+            dtend=end_time,
+            summary=summary,
+            description=description
+        )
+        # Re-parse to get the UID, though library might provide it
+        ical = icalendar.Calendar.from_ical(event.data)
+        for component in ical.walk():
+            if component.name == "VEVENT":
+                return str(component.get("uid"))
+        return "Event Created (UID Unknown)"
+
+    def update_event(self, calendar_name: str, event_uid: str, summary: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, description: Optional[str] = None) -> bool:
+        calendar = self._get_calendar(calendar_name)
+        event = calendar.event_by_uid(event_uid)
+        
+        # We need to modify the underlying ical component
+        ical = icalendar.Calendar.from_ical(event.data)
+        changed = False
+        
+        for component in ical.walk():
+            if component.name == "VEVENT":
+                if summary:
+                    component["summary"] = summary
+                    changed = True
+                if description is not None:
+                    component["description"] = description
+                    changed = True
+                if start_time:
+                    component["dtstart"] = icalendar.vDatetime(start_time)
+                    changed = True
+                if end_time:
+                    component["dtend"] = icalendar.vDatetime(end_time)
+                    changed = True
+        
+        if changed:
+            event.data = ical.to_ical()
+            event.save()
+            return True
+        return False
+
+    def delete_event(self, calendar_name: str, event_uid: str):
+        calendar = self._get_calendar(calendar_name)
+        event = calendar.event_by_uid(event_uid)
+        event.delete()

src/server.py

@@ -0,0 +1,119 @@
+from mcp.server.fastmcp import FastMCP
+from src.caldav_client import CalDAVClient
+from datetime import datetime
+from typing import Optional
+
+# Initialize FastMCP server
+mcp = FastMCP("CalDAV Server")
+
+# Initialize CalDAV client
+# We initialize it lazily or here if credentials are guaranteed to be present
+try:
+    client = CalDAVClient()
+except Exception as e:
+    import sys
+    sys.stderr.write(f"Error initializing CalDAV client: {e}\n")
+    client = None
+
+@mcp.tool()
+def list_calendars() -> str:
+    """Lists all available calendars on the CalDAV server."""
+    if not client:
+        return "Error: CalDAV client not initialized (check credentials)"
+    try:
+        calendars = client.list_calendars()
+        return str(calendars)
+    except Exception as e:
+        return f"Error listing calendars: {str(e)}"
+
+@mcp.tool()
+def list_events(calendar_name: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None) -> str:
+    """
+    Lists events. If calendar_name is not provided, lists events from ALL calendars.
+    
+    Args:
+        calendar_name: Optional. The name of the calendar to query.
+        start_date: Optional start date in YYYY-MM-DD or ISO format.
+        end_date: Optional end date in YYYY-MM-DD or ISO format.
+    """
+    if not client:
+        return "Error: CalDAV client not initialized"
+    
+    try:
+        dt_start = datetime.fromisoformat(start_date) if start_date else None
+        dt_end = datetime.fromisoformat(end_date) if end_date else None
+        
+        events = client.list_events(calendar_name, dt_start, dt_end)
+        return str(events)
+    except Exception as e:
+        return f"Error listing events: {str(e)}"
+
+@mcp.tool()
+def create_event(calendar_name: str, summary: str, start_time: str, end_time: str, description: str = "") -> str:
+    """
+    Creates a new event.
+    
+    Args:
+        calendar_name: Target calendar.
+        summary: Event title.
+        start_time: Start time (ISO format).
+        end_time: End time (ISO format).
+        description: Event description.
+    """
+    if not client:
+        return "Error: CalDAV client not initialized"
+    
+    try:
+        dt_start = datetime.fromisoformat(start_time)
+        dt_end = datetime.fromisoformat(end_time)
+        
+        result = client.create_event(calendar_name, summary, dt_start, dt_end, description)
+        return f"Event created: {result}"
+    except Exception as e:
+        return f"Error creating event: {str(e)}"
+
+@mcp.tool()
+def update_event(calendar_name: str, event_uid: str, summary: Optional[str] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, description: Optional[str] = None) -> str:
+    """
+    Updates an existing event.
+    
+    Args:
+        calendar_name: Calendar containing the event.
+        event_uid: UID of the event to update.
+        summary: New title (optional).
+        start_time: New start time (ISO format, optional).
+        end_time: New end time (ISO format, optional).
+        description: New description (optional).
+    """
+    if not client:
+        return "Error: CalDAV client not initialized"
+    
+    try:
+        dt_start = datetime.fromisoformat(start_time) if start_time else None
+        dt_end = datetime.fromisoformat(end_time) if end_time else None
+        
+        success = client.update_event(calendar_name, event_uid, summary, dt_start, dt_end, description)
+        return "Event updated successfully" if success else "Event update failed (or no changes made)"
+    except Exception as e:
+        return f"Error updating event: {str(e)}"
+
+@mcp.tool()
+def delete_event(calendar_name: str, event_uid: str) -> str:
+    """
+    Deletes an event.
+    
+    Args:
+        calendar_name: Calendar containing the event.
+        event_uid: UID of the event to delete.
+    """
+    if not client:
+        return "Error: CalDAV client not initialized"
+    
+    try:
+        client.delete_event(calendar_name, event_uid)
+        return "Event deleted successfully"
+    except Exception as e:
+        return f"Error deleting event: {str(e)}"
+
+if __name__ == "__main__":
+    mcp.run()