Oslo Marathon MCP

oslo_marathon_mcp.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "mcp[cli]",
#     "garminconnect",
#     "rank-bm25",
#     "icalendar",
# ]
# ///


from mcp.server.fastmcp import FastMCP
from rank_bm25 import BM25Okapi
from garminconnect import Garmin
from pathlib import Path
import datetime
from icalendar import Calendar
from zoneinfo import ZoneInfo


MCP = FastMCP("oslo_marathon")
GARMIN_EMAIL, GARMIN_PASSWORD, GARMIN_TOKENSTORE = (
    "XXXXX",
    "XXXXX",
    "~/.garminconnect",
)


def start_garmin() -> Garmin:
    """Initialize Garmin connection."""
    try:
        GARMIN = Garmin()
        GARMIN.login(GARMIN_TOKENSTORE)
        return GARMIN
    except Exception as e:
        print(f"Could not login with token store: {e}")

    try:
        GARMIN = Garmin(
            email=GARMIN_EMAIL,
            password=GARMIN_PASSWORD,
            is_cn=False,
        )
        GARMIN.login()
        GARMIN.garth.dump(GARMIN_TOKENSTORE)
        return GARMIN
    except Exception as e:
        print(f"Could not login with email and password: {e}")
        raise


def prep_book() -> tuple[BM25Okapi, list[str]]:
    script_dir = Path(__file__).resolve().parent
    book = script_dir / "marathon.md"
    with open(book, "r", encoding="utf-8") as file:
        content = file.read()

    words = content.split()
    corpus = []
    chunk_size_words = 800
    overlap = int(chunk_size_words * 0.2)
    for i in range(0, len(words), chunk_size_words - overlap):
        corpus.append(" ".join(words[i : i + chunk_size_words]))

    tokenized_corpus = [doc.split(" ") for doc in corpus]
    bm25 = BM25Okapi(tokenized_corpus)
    return bm25, corpus


@MCP.tool()
def search_marathon_training_book(
    query: str,
    total_results: int = 10,
) -> str:
    """
    Search the marathon training book for a specific query.
    Args:
        query (str): The search query.
        total_results (int): The number of results to return. (default: 10)
    """

    bm25, corpus = prep_book()
    tokenized_query = query.split(" ")
    results = bm25.get_top_n(tokenized_query, corpus, n=total_results)
    results_as_bullets = "\n".join(
        f"<result_{i + 1}>\n...{result}...\n</result_{i + 1}>"
        for i, result in enumerate(results)
    )

    return f"""
    <search_marathon_training_book>
    <search_query>{query}</search_query>
    {results_as_bullets}
    </search_marathon_training_book>
    """.strip()


@MCP.tool()
def fetch_athlete_runs(lookback_days: int) -> str:
    """
    Fetch athlete's running activities from Garmin Connect.
    Args:
        lookback_days (int): Number of days to look back for running activities.
    """
    garmin = start_garmin()
    today = datetime.date.today()
    startdate = today - datetime.timedelta(days=lookback_days)

    activities = garmin.get_activities_by_date(
        startdate.isoformat(), today.isoformat(), activitytype="running"
    )
    sorted_runs = sorted(activities, key=lambda x: x["startTimeLocal"], reverse=True)

    output = ["<Runs>"]
    for run in sorted_runs:
        start_time = run["startTimeLocal"]
        distance_km = round(run.get("distance", 0) / 1000, 2)
        duration_sec = run.get("duration", 0)
        duration_fmt = str(datetime.timedelta(seconds=int(duration_sec)))
        avg_hr = run.get("averageHR", "n/a")

        # Compute pace if distance and duration are valid
        if distance_km > 0 and duration_sec > 0:
            pace_sec_per_km = duration_sec / distance_km
            pace_min = int(pace_sec_per_km // 60)
            pace_sec = int(pace_sec_per_km % 60)
            pace = f"{pace_min}:{pace_sec:02d} min/km"
        else:
            pace = "n/a"

        output.append(f"""  <Run>
    <Date>{start_time}</Date>
    <Distance>{distance_km} km</Distance>
    <Duration>{duration_fmt}</Duration>
    <Pace>{pace}</Pace>
    <HeartRate>{avg_hr} bpm</HeartRate>
  </Run>""")

    output.append("</Runs>")
    return "\n".join(output)


@MCP.tool()
def get_training_program_events(start_date: str, end_date: str) -> str:
    """
    Fetches the athlete's training program events in the date range specified.

    Args:
        start_date: Start date in format "DD-MM-YYYY"
        end_date: End date in format "DD-MM-YYYY"

    Returns:
        A string with matching VEVENTS in XML-like format.
    """
    start = datetime.datetime.strptime(start_date, "%d-%m-%Y").replace(
        tzinfo=ZoneInfo("UTC")
    )
    end = datetime.datetime.strptime(end_date, "%d-%m-%Y").replace(
        tzinfo=ZoneInfo("UTC")
    )

    ics_path = Path(__file__).parent / "oslo_training_plan.ics"
    ics_content = ics_path.read_bytes()

    cal = Calendar.from_ical(ics_content)
    events_xml = []

    def to_datetime(val: datetime.datetime | datetime.date) -> datetime.datetime:
        if isinstance(val, datetime.date) and not isinstance(val, datetime.datetime):
            return datetime.datetime.combine(
                val, datetime.time.min, tzinfo=ZoneInfo("UTC")
            )
        if val.tzinfo is None:
            return val.replace(tzinfo=ZoneInfo("UTC"))
        return val

    for component in cal.walk():
        if component.name != "VEVENT":
            continue

        dtstart = to_datetime(component.decoded("DTSTART"))
        dtend_raw = component.get("DTEND")
        dtend = to_datetime(component.decoded("DTEND")) if dtend_raw else None

        if dtend is None or not (start <= dtstart <= end or start <= dtend <= end):
            continue

        summary = component.get("SUMMARY", "")
        location = component.get("LOCATION", "")

        events_xml.append(f"""<event>
  <summary>{summary}</summary>
  <location>{location}</location>
  <start>{dtstart.isoformat()}</start>
  <end>{dtend.isoformat()}</end>
</event>""")

    return "<events>\n" + "\n".join(events_xml) + "\n</events>"


@MCP.tool()
def get_current_date() -> str:
    """
    Returns the current date in ISO format.
    """
    return datetime.datetime.now(tz=ZoneInfo("UTC")).isoformat()


if __name__ == "__main__":
    # print(
    #     search_marathon_training_book(
    #         "What's the best strategy for recovery after a marathon?",
    #         total_results=5,
    #     )
    # )
    # print(fetch_athlete_runs(lookback_days=10))
    # print(
    #     extract_events_xml(
    #         start_date="01-05-2025",
    #         end_date="30-05-2025",
    #     )
    # )
    MCP.run(transport="stdio")