Calendar MCP Server

from typing import Any, Dict, List, Optional from datetime import datetime, timedelta import time import os import httpx from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP from icalendar import Calendar from dateutil.parser import parse from dateutil.tz import gettz import recurring_ical_events # Load environment variables load_dotenv() # Initialize FastMCP server mcp = FastMCP("calendar") # Constants ICS_URL = os.getenv("ICS_URL") if not ICS_URL: raise ValueError("ICS_URL environment variable is not set") CACHE_DURATION = 300 # 5 minutes DEFAULT_TZ = gettz("Europe/Paris") # Global cache class EventCache: def __init__(self): self.calendar_data: bytes = b"" self.last_fetch: float = 0 def is_valid(self) -> bool: return time.time() - self.last_fetch < CACHE_DURATION cache = EventCache() async def get_raw_calendar() -> bytes: """Fetches the raw ICS content, using a cache.""" if cache.is_valid() and cache.calendar_data: return cache.calendar_data async with httpx.AsyncClient() as client: response = await client.get(ICS_URL) response.raise_for_status() cache.calendar_data = response.content cache.last_fetch = time.time() return cache.calendar_data @mcp.tool() async def list_events(start_date: str, end_date: str) -> List[Dict[str, Any]]: """ List calendar events within a specific date range, expanding recurring events. Args: start_date: Start of the range (ISO format, e.g., '2023-01-01') end_date: End of the range (ISO format) Returns: List of events falling within the range. """ try: start_dt = parse(start_date) end_dt = parse(end_date) if start_dt.tzinfo is None: start_dt = start_dt.replace(tzinfo=DEFAULT_TZ) if end_dt.tzinfo is None: end_dt = end_dt.replace(tzinfo=DEFAULT_TZ) except ValueError: return [{"error": "Invalid date format. Please use ISO 8601 format."}] raw_cal = await get_raw_calendar() cal = Calendar.from_ical(raw_cal) # Use recurring-ical-events to expand events in the range events_in_range = recurring_ical_events.of(cal).between(start_dt, end_dt) formatted_events = [] for component in events_in_range: start = component.get("dtstart").dt end = component.get("dtend").dt if component.get("dtend") else None # Normalize to datetime objects if they are date objects if not isinstance(start, datetime): start = datetime.combine(start, datetime.min.time(), tzinfo=DEFAULT_TZ) if end and not isinstance(end, datetime): end = datetime.combine(end, datetime.min.time(), tzinfo=DEFAULT_TZ) # Ensure timezone awareness if start.tzinfo is None: start = start.replace(tzinfo=DEFAULT_TZ) if end and end.tzinfo is None: end = end.replace(tzinfo=DEFAULT_TZ) formatted_events.append({ "summary": str(component.get("summary", "")), "start": start.isoformat(), "end": end.isoformat() if end else None, "description": str(component.get("description", "")), "location": str(component.get("location", "")), }) # Sort by start time formatted_events.sort(key=lambda x: x["start"]) return formatted_events @mcp.tool() async def search_events(query: str) -> List[Dict[str, Any]]: """ Search for events matching a text query in summary or description. Note: This searches a wide range (current year +/- 1 year) to be useful, as text search on infinite recurrence is impossible without a range. Args: query: The text to search for. Returns: List of matching events. """ # For search, we need a range to expand recurrence. # Let's search from 1 month ago to 6 months ahead as a reasonable default for "search" # or we could revert to non-expanded search if the user wants to search *patterns* # but usually users want *instances*. Let's do a reasonable window. now = datetime.now(DEFAULT_TZ) search_start = now - timedelta(days=30) search_end = now + timedelta(days=180) raw_cal = await get_raw_calendar() cal = Calendar.from_ical(raw_cal) events_in_range = recurring_ical_events.of(cal).between(search_start, search_end) query_lower = query.lower() matching_events = [] for component in events_in_range: summary = str(component.get("summary", "")).lower() description = str(component.get("description", "")).lower() if query_lower in summary or query_lower in description: start = component.get("dtstart").dt end = component.get("dtend").dt if component.get("dtend") else None # Handle date/datetime and timezone normalization (same as above) if not isinstance(start, datetime): start = datetime.combine(start, datetime.min.time(), tzinfo=DEFAULT_TZ) if end and not isinstance(end, datetime): end = datetime.combine(end, datetime.min.time(), tzinfo=DEFAULT_TZ) if start.tzinfo is None: start = start.replace(tzinfo=DEFAULT_TZ) if end and end.tzinfo is None: end = end.replace(tzinfo=DEFAULT_TZ) matching_events.append({ "summary": str(component.get("summary", "")), "start": start.isoformat(), "end": end.isoformat() if end else None, "description": str(component.get("description", "")), "location": str(component.get("location", "")), }) return matching_events def main(): # Initialize and run the server mcp.run(transport="stdio") if __name__ == "__main__": main()