feat: init
All checks were successful
perso/mcp-ics/pipeline/head This commit looks good

This commit is contained in:
Julien Cabillot
2026-03-15 19:22:18 -04:00
commit ee0d719142
7 changed files with 434 additions and 0 deletions

3
.env.example Normal file
View File

@@ -0,0 +1,3 @@
CALDAV_DATA_PATH=/path/to/radicale/data
EXCLUDE_PATTERN=^Poubelle
BIRTHDAY_PATTERN=^Anniversaire|^Anniv

25
.gitignore vendored Normal file
View File

@@ -0,0 +1,25 @@
# Virtual environments
venv/
.venv/
env/
env.bak/
# Environment variables
.env
.env.local
# Python cache and compiled files
__pycache__/
*.py[cod]
*$py.class
*.so
# Distribution / packaging
build/
dist/
*.egg-info/
.eggs/
# OS generated files
.DS_Store
Thumbs.db

38
Jenkinsfile vendored Normal file
View File

@@ -0,0 +1,38 @@
pipeline {
environment {
registry = 'https://registry.hub.docker.com'
registryCredential = 'dockerhub_jcabillot'
dockerImage = 'jcabillot/mcp-caldav'
}
agent any
triggers {
cron('@midnight')
}
stages {
stage('Clone repository') {
steps{
checkout scm
}
}
stage('Build image') {
steps{
sh 'docker build --force-rm=true --no-cache=true --pull -t ${dockerImage} -f pkg/Dockerfile .'
}
}
stage('Deploy Image') {
steps{
script {
withCredentials([usernamePassword(credentialsId: 'dockerhub_jcabillot', usernameVariable: 'DOCKER_USER', passwordVariable: 'DOCKER_PASS')]) {
sh 'docker login --username ${DOCKER_USER} --password ${DOCKER_PASS}'
sh 'docker push ${dockerImage}'
}
}
}
}
}
}

17
docker-compose.yml Normal file
View File

@@ -0,0 +1,17 @@
services:
mcp-server:
build:
context: .
dockerfile: pkg/Dockerfile
container_name: mcp_caldav_server
ports:
- "127.0.0.1:8000:8000"
env_file:
- .env
volumes:
# Mount the source code for hot-reloading (optional)
- ./src:/app/src:ro,z
# Mount the Radicale data directory as read-only
- ${CALDAV_DATA_PATH:-./example_data}:/data:ro,z
environment:
- CALDAV_DATA_PATH=/data

20
pkg/Dockerfile Normal file
View File

@@ -0,0 +1,20 @@
FROM docker.io/library/python:3.14-slim
# Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
# Set the working directory in the container
WORKDIR /app
# Copy the requirements file into the container
COPY requirements.txt .
# Install dependencies using buildkit cache
RUN --mount=type=cache,target=/root/.cache/pip pip install -r requirements.txt
# Copy the source code
COPY src/ ./src/
# Command to run the MCP server
CMD ["python", "src/server.py"]

8
requirements.txt Normal file
View File

@@ -0,0 +1,8 @@
mcp>=1,<2
fastmcp>=3,<4
python-dotenv>=1,<2
uvicorn>=0.34,<1
starlette>=0.46,<1
icalendar>=7,<8
recurring-ical-events>=3,<4
python-dateutil>=2,<3

323
src/server.py Normal file
View File

@@ -0,0 +1,323 @@
"""
MCP Server exposing a read-only agenda tool for Radicale CalDAV data.
Reads .ics files directly from the Radicale data directory, parses
VEVENT and VTODO components, expands recurring events (RRULE), and
returns them sorted chronologically for a given date range.
"""
import json
import logging
import os
import re
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Optional
import recurring_ical_events
from dateutil import parser as date_parser
from dotenv import load_dotenv
from icalendar import Calendar
from starlette.requests import Request
from starlette.responses import JSONResponse
from fastmcp import FastMCP
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger(__name__)
CALDAV_DATA_PATH = os.environ.get("CALDAV_DATA_PATH", "")
EXCLUDE_PATTERN = os.environ.get("EXCLUDE_PATTERN", "")
BIRTHDAY_PATTERN = os.environ.get("BIRTHDAY_PATTERN", r"^Anniversaire|^Anniv")
if not CALDAV_DATA_PATH:
raise ValueError("CALDAV_DATA_PATH environment variable is required.")
logger.info("Starting MCP CalDAV server with data path: %s", CALDAV_DATA_PATH)
if EXCLUDE_PATTERN:
logger.info("Exclude pattern: %s", EXCLUDE_PATTERN)
logger.info("Birthday pattern: %s", BIRTHDAY_PATTERN)
# Compile patterns once at startup
_exclude_re: Optional[re.Pattern] = None
if EXCLUDE_PATTERN:
try:
_exclude_re = re.compile(EXCLUDE_PATTERN, re.IGNORECASE)
logger.info("Compiled exclude pattern: %s", EXCLUDE_PATTERN)
except re.error as exc:
logger.error("Invalid EXCLUDE_PATTERN '%s': %s", EXCLUDE_PATTERN, exc)
_birthday_re: Optional[re.Pattern] = None
if BIRTHDAY_PATTERN:
try:
_birthday_re = re.compile(BIRTHDAY_PATTERN, re.IGNORECASE)
logger.info("Compiled birthday pattern: %s", BIRTHDAY_PATTERN)
except re.error as exc:
logger.error("Invalid BIRTHDAY_PATTERN '%s': %s", BIRTHDAY_PATTERN, exc)
# Initialize FastMCP server
mcp = FastMCP("mcp-caldav")
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request):
"""Simple health check endpoint for Kubernetes liveness/readiness probes."""
return JSONResponse({"status": "ok"})
def discover_calendars(data_path: str) -> list[dict[str, Any]]:
"""
Discovers Radicale calendar collections.
Walks the collection-root directory and reads .Radicale.props files
to identify VCALENDAR collections (skipping VADDRESSBOOK).
Returns a list of dicts with keys: path, name, description, user.
"""
calendars = []
collection_root = Path(data_path) / "collections" / "collection-root"
if not collection_root.is_dir():
logger.warning("Collection root not found: %s", collection_root)
return calendars
for user_dir in sorted(collection_root.iterdir()):
if not user_dir.is_dir() or user_dir.name.startswith("."):
continue
for cal_dir in sorted(user_dir.iterdir()):
if not cal_dir.is_dir() or cal_dir.name.startswith("."):
continue
props_file = cal_dir / ".Radicale.props"
if not props_file.is_file():
continue
try:
props = json.loads(props_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to read props for %s: %s", cal_dir, exc)
continue
# Only process calendar collections
if props.get("tag") != "VCALENDAR":
continue
calendars.append({
"path": str(cal_dir),
"name": props.get("D:displayname", cal_dir.name),
"description": props.get("C:calendar-description", ""),
"user": user_dir.name,
})
logger.info("Discovered %d calendar(s)", len(calendars))
for cal in calendars:
logger.info(" - %s (%s) [user: %s]", cal["name"], cal["path"], cal["user"])
return calendars
def parse_ics_files(calendar_path: str) -> list[Calendar]:
"""
Reads and parses all .ics files from a calendar directory.
Returns a list of icalendar.Calendar objects.
"""
cal_dir = Path(calendar_path)
calendars = []
for ics_file in cal_dir.glob("*.ics"):
try:
raw = ics_file.read_bytes()
cal = Calendar.from_ical(raw)
calendars.append(cal)
except Exception as exc:
logger.warning("Failed to parse %s: %s", ics_file, exc)
continue
return calendars
def component_to_dt(value: Any) -> Optional[datetime]:
"""
Extracts a datetime from an iCalendar date/datetime property.
Handles both date and datetime types, converting dates to datetime
at midnight for consistent comparison.
"""
if value is None:
return None
dt = value.dt if hasattr(value, "dt") else value
if isinstance(dt, datetime):
return dt
if isinstance(dt, date):
return datetime(dt.year, dt.month, dt.day)
return None
def format_dt(dt: Optional[datetime | date]) -> Optional[str]:
"""Formats a date or datetime to ISO 8601 string."""
if dt is None:
return None
if isinstance(dt, datetime):
return dt.isoformat()
if isinstance(dt, date):
return dt.isoformat()
return str(dt)
def format_component(
component: Any, calendar_name: str, include_birthdays: bool = False,
) -> Optional[dict[str, Any]]:
"""
Formats a VEVENT or VTODO component into a dict for API response.
Returns None if the component should be excluded (matches EXCLUDE_PATTERN
or matches BIRTHDAY_PATTERN when include_birthdays is False).
"""
summary = str(component.get("SUMMARY", ""))
# Apply exclude filter
if _exclude_re and _exclude_re.search(summary):
return None
# Apply birthday filter (excluded by default, included on demand)
if not include_birthdays and _birthday_re and _birthday_re.search(summary):
return None
comp_type = component.name # "VEVENT" or "VTODO"
dtstart = component_to_dt(component.get("DTSTART"))
dtend = component_to_dt(component.get("DTEND"))
due = component_to_dt(component.get("DUE"))
result: dict[str, Any] = {
"type": "event" if comp_type == "VEVENT" else "todo",
"summary": summary,
"start": format_dt(dtstart),
"calendar": calendar_name,
"uid": str(component.get("UID", "")),
}
if comp_type == "VEVENT":
result["end"] = format_dt(dtend)
elif comp_type == "VTODO":
result["due"] = format_dt(due)
status = str(component.get("STATUS", ""))
if status:
result["status"] = status
location = str(component.get("LOCATION", ""))
if location:
result["location"] = location
description = str(component.get("DESCRIPTION", ""))
if description:
result["description"] = description
return result
def get_events_in_range(
start: datetime,
end: datetime,
include_birthdays: bool = False,
) -> list[dict[str, Any]]:
"""
Returns all VEVENT and VTODO components across all calendars
that fall within [start, end], with RRULE expansion.
"""
calendars = discover_calendars(CALDAV_DATA_PATH)
all_items: list[dict[str, Any]] = []
for cal_info in calendars:
ics_calendars = parse_ics_files(cal_info["path"])
for cal in ics_calendars:
try:
events = recurring_ical_events.of(cal, components=["VEVENT", "VTODO"]).between(
start, end
)
except Exception as exc:
logger.warning(
"Error expanding events in %s: %s", cal_info["name"], exc
)
continue
for event in events:
formatted = format_component(event, cal_info["name"], include_birthdays)
if formatted is not None:
all_items.append(formatted)
# Sort by start date
def sort_key(item: dict[str, Any]) -> str:
return item.get("start") or item.get("due") or ""
all_items.sort(key=sort_key)
return all_items
@mcp.tool()
def get_agenda(
start_date: str,
end_date: str = "",
include_birthdays: bool = False,
):
"""
Returns all calendar events and todos within a date range.
All calendars are aggregated. Recurring events are automatically expanded.
Results are sorted chronologically.
Birthdays are excluded by default to reduce noise. Set include_birthdays
to True when the user explicitly asks about birthdays or anniversaries.
Args:
start_date: Start date in ISO 8601 format (e.g. "2026-03-15" or "2026-03-15T09:00:00").
end_date: End date in ISO 8601 format. Defaults to 7 days after start_date if not provided.
include_birthdays: Whether to include birthday/anniversary events (default: False).
"""
logger.info(
"Tool get_agenda called: start_date=%r, end_date=%r, include_birthdays=%r",
start_date,
end_date,
include_birthdays,
)
try:
start = date_parser.parse(start_date)
if end_date:
end = date_parser.parse(end_date)
else:
end = start + timedelta(days=7)
logger.info("Fetching agenda from %s to %s", start.isoformat(), end.isoformat())
items = get_events_in_range(start, end, include_birthdays)
logger.info("Found %d item(s) in range", len(items))
return {
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"count": len(items),
"items": items,
}
except Exception as exc:
logger.error("Error in get_agenda: %s", exc, exc_info=True)
return {"error": f"get_agenda failed: {exc}"}
if __name__ == "__main__":
logger.info("Starting SSE server on 0.0.0.0:8000...")
mcp.run(transport="sse", host="0.0.0.0", port=8000)