From 28119f17614c36858602414a66d0363a7351c434 Mon Sep 17 00:00:00 2001 From: Chris Farhood Date: Mon, 20 Jul 2026 15:44:05 -0400 Subject: [PATCH] feat(athlete): add profile, sport-settings, and summary read tools Closes the biggest coaching-context gap: expose the athlete's identity/physiology (get_athlete_profile), per-sport FTP/zones/thresholds (get_sport_settings, with an optional sport filter and the settings id needed for future writes), and a training-load summary over a range (get_athlete_summary). New tools/athlete.py plus formatters in utils/formatting.py; registered in server.py and tools/__init__.py. Implements #2. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN --- src/intervals_mcp_server/server.py | 9 + src/intervals_mcp_server/tools/__init__.py | 8 + src/intervals_mcp_server/tools/athlete.py | 129 +++++++++++++ src/intervals_mcp_server/utils/formatting.py | 140 ++++++++++++++ tests/test_athlete.py | 188 +++++++++++++++++++ 5 files changed, 474 insertions(+) create mode 100644 src/intervals_mcp_server/tools/athlete.py create mode 100644 tests/test_athlete.py diff --git a/src/intervals_mcp_server/server.py b/src/intervals_mcp_server/server.py index 5efe59a..320b701 100644 --- a/src/intervals_mcp_server/server.py +++ b/src/intervals_mcp_server/server.py @@ -90,6 +90,12 @@ from intervals_mcp_server.tools.wellness import ( # pylint: disable=wrong-impor get_wellness_data, update_wellness, ) +from intervals_mcp_server.tools.athlete import ( # pylint: disable=wrong-import-position # noqa: E402 + get_athlete_profile, + get_athlete_summary, + get_sport_settings, +) + from intervals_mcp_server.tools.power_curves import get_athlete_power_curves # pylint: disable=wrong-import-position # noqa: E402 from intervals_mcp_server.tools.custom_items import ( # pylint: disable=wrong-import-position # noqa: E402 create_custom_item, @@ -117,6 +123,9 @@ __all__ = [ "add_or_update_event", "get_wellness_data", "update_wellness", + "get_athlete_profile", + "get_sport_settings", + "get_athlete_summary", "get_gear_list", "get_athlete_power_curves", "get_custom_items", diff --git a/src/intervals_mcp_server/tools/__init__.py b/src/intervals_mcp_server/tools/__init__.py index a6b3128..d713d49 100644 --- a/src/intervals_mcp_server/tools/__init__.py +++ b/src/intervals_mcp_server/tools/__init__.py @@ -36,6 +36,11 @@ from intervals_mcp_server.tools.wellness import ( # noqa: F401 get_wellness_data, update_wellness, ) +from intervals_mcp_server.tools.athlete import ( # noqa: F401 + get_athlete_profile, + get_athlete_summary, + get_sport_settings, +) def register_tools(mcp_instance: FastMCP) -> None: @@ -74,4 +79,7 @@ __all__ = [ "get_gear_list", "get_wellness_data", "update_wellness", + "get_athlete_profile", + "get_sport_settings", + "get_athlete_summary", ] diff --git a/src/intervals_mcp_server/tools/athlete.py b/src/intervals_mcp_server/tools/athlete.py new file mode 100644 index 0000000..96d51fc --- /dev/null +++ b/src/intervals_mcp_server/tools/athlete.py @@ -0,0 +1,129 @@ +""" +Athlete-profile and configuration MCP tools for Intervals.icu. + +Read tools exposing the athlete's profile, per-sport training settings +(FTP / zones / thresholds), and training-load summaries — the context an AI +coach needs to reason about intensity and readiness. +""" + +from typing import Any + +from intervals_mcp_server import credentials +from intervals_mcp_server.api.client import make_intervals_request +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.utils.formatting import ( + format_athlete_profile, + format_athlete_summary, + format_sport_settings, +) +from intervals_mcp_server.utils.validation import resolve_date_params + +# Import mcp instance from shared module for tool registration +from intervals_mcp_server.mcp_instance import mcp # noqa: F401 + + +@mcp.tool() +async def get_athlete_profile() -> str: + """Get the signed-in athlete's profile from Intervals.icu. + + Returns identity and physiology basics (name, sex, weight, resting HR, + timezone, units, location). For per-sport FTP / zones / thresholds use + get_sport_settings instead. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request(url=f"/athlete/{athlete_id}", api_key=api_key) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching athlete profile: {result.get('message')}" + if not isinstance(result, dict): + return "No athlete profile found." + return format_athlete_profile(result) + + +@mcp.tool() +async def get_sport_settings(sport: str | None = None) -> str: + """Get the athlete's per-sport training settings (FTP, zones, thresholds). + + These are the values that drive load, intensity and zone calculations across + Intervals.icu. Each record's "Settings ID" is the identifier update_sport_settings + uses to target a specific sport. + + Args: + sport: Optional sport type to filter by (e.g. "Ride", "Run"). Matches the + record's sport types case-insensitively. If omitted, all sports are returned. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/sport-settings", api_key=api_key + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching sport settings: {result.get('message')}" + + records = [r for r in result if isinstance(r, dict)] if isinstance(result, list) else [] + if not records: + return "No sport settings found." + + if sport: + want = sport.strip().lower() + records = [ + r for r in records if any(want == str(t).lower() for t in (r.get("types") or [])) + ] + if not records: + return f"No sport settings found for sport '{sport}'." + + return "\n\n".join(format_sport_settings(r) for r in records) + + +@mcp.tool() +async def get_athlete_summary( + start_date: str | None = None, + end_date: str | None = None, + tags: str | None = None, +) -> str: + """Get a training-load summary (fitness/fatigue/form and totals) over a date range. + + Args: + start_date: Start date in YYYY-MM-DD format (optional, defaults to 30 days ago). + end_date: End date in YYYY-MM-DD format (optional, defaults to today). + tags: Optional comma-separated activity tags to filter by. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + start_date, end_date = resolve_date_params(start_date, end_date) + params: dict[str, Any] = {"start": start_date, "end": end_date} + if tags: + tag_list = [t.strip() for t in tags.split(",") if t.strip()] + if tag_list: + params["tags"] = tag_list + + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/athlete-summary", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching athlete summary: {result.get('message')}" + + if isinstance(result, list): + summaries = [s for s in result if isinstance(s, dict)] + elif isinstance(result, dict): + summaries = [result] + else: + summaries = [] + + if not summaries: + return "No summary data found for the specified date range." + + header = f"Athlete Summary ({start_date} to {end_date}):\n\n" + return header + "\n\n".join(format_athlete_summary(s) for s in summaries) diff --git a/src/intervals_mcp_server/utils/formatting.py b/src/intervals_mcp_server/utils/formatting.py index 3e93efe..b2e04a2 100644 --- a/src/intervals_mcp_server/utils/formatting.py +++ b/src/intervals_mcp_server/utils/formatting.py @@ -420,6 +420,146 @@ def format_wellness_entry(entries: dict[str, Any], include_all_fields: bool = Fa return "\n".join(lines) +def format_athlete_profile(athlete: dict[str, Any]) -> str: + """Format an athlete profile into a readable string. + + Renders identity/physiology basics. The embedded per-sport settings blob is + only summarised (a count) — get_sport_settings renders the detail. + """ + name = athlete.get("name") or " ".join( + p for p in [athlete.get("firstname"), athlete.get("lastname")] if p + ) or "Unknown" + lines = ["Athlete Profile:", "", f"Name: {name}", f"ID: {athlete.get('id', 'N/A')}"] + + weight = athlete.get("weight") + if weight is None: + weight = athlete.get("icu_weight") + for label, value, unit in [ + ("Sex", athlete.get("sex"), ""), + ("Date of Birth", athlete.get("icu_date_of_birth"), ""), + ("Weight", weight, "kg"), + ("Resting HR", athlete.get("icu_resting_hr"), "bpm"), + ("Timezone", athlete.get("timezone"), ""), + ("Units", athlete.get("measurement_preference"), ""), + ]: + if value is not None and value != "": + lines.append(f"{label}: {value}{(' ' + unit) if unit else ''}") + + location = ", ".join( + p for p in [athlete.get("city"), athlete.get("state"), athlete.get("country")] if p + ) + if location: + lines.append(f"Location: {location}") + if athlete.get("icu_coach"): + lines.append("Role: Coach") + if athlete.get("bio"): + lines.append(f"Bio: {athlete['bio']}") + + sport_settings = athlete.get("sportSettings") or athlete.get("icu_type_settings") + if isinstance(sport_settings, list) and sport_settings: + lines.append( + f"Sport Settings: {len(sport_settings)} sport(s) configured " + "(use get_sport_settings for FTP/zones/thresholds)" + ) + return "\n".join(lines) + + +def _format_zone_line(label: str, zones: Any, names: Any, unit: str = "") -> str | None: + """Render a zone-boundary array, pairing with names when they line up.""" + if not isinstance(zones, list) or not zones: + return None + if isinstance(names, list) and len(names) == len(zones): + parts = [f"{n}: {z}{unit}" for n, z in zip(names, zones, strict=True)] + else: + parts = [f"{z}{unit}" for z in zones] + return f"{label}: " + ", ".join(parts) + + +def format_sport_settings(settings: dict[str, Any]) -> str: + """Format one per-sport settings record (FTP, zones, thresholds) into text. + + The record ``id`` is always shown — it is the identifier update_sport_settings + needs to target a specific sport's settings. + """ + types = settings.get("types") or [] + sport = ", ".join(str(t) for t in types) if types else "Unknown" + lines = [f"Sport Settings — {sport}:", f"Settings ID: {settings.get('id', 'N/A')}"] + + power_bits = [] + for key, label, unit in [ + ("ftp", "FTP", "W"), + ("indoor_ftp", "Indoor FTP", "W"), + ("w_prime", "W'", "J"), + ("p_max", "Pmax", "W"), + ]: + if settings.get(key) is not None: + power_bits.append(f"{label}: {settings[key]}{unit}") + if power_bits: + lines += ["", "Power:"] + [f"- {b}" for b in power_bits] + zone_line = _format_zone_line("Zones", settings.get("power_zones"), settings.get("power_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + hr_bits = [] + for key, label in [("lthr", "LTHR"), ("max_hr", "Max HR")]: + if settings.get(key) is not None: + hr_bits.append(f"{label}: {settings[key]} bpm") + if hr_bits: + lines += ["", "Heart Rate:"] + [f"- {b}" for b in hr_bits] + zone_line = _format_zone_line("Zones", settings.get("hr_zones"), settings.get("hr_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + if settings.get("threshold_pace") is not None: + units = settings.get("pace_units", "") + lines += ["", "Pace:", f"- Threshold: {settings['threshold_pace']} {units}".rstrip()] + zone_line = _format_zone_line("Zones", settings.get("pace_zones"), settings.get("pace_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + defaults = [] + for key, label in [("warmup_time", "Warmup"), ("cooldown_time", "Cooldown")]: + if settings.get(key) is not None: + defaults.append(f"{label}: {settings[key]}s") + if defaults: + lines += ["", "Defaults: " + ", ".join(defaults)] + return "\n".join(lines) + + +def format_athlete_summary(summary: dict[str, Any]) -> str: + """Format a training-load summary (fitness/fatigue/form + totals) into text.""" + lines: list[str] = [] + if summary.get("date"): + lines.append(f"Period ending {summary['date']}:") + for key, label, unit in [ + ("count", "Activities", ""), + ("moving_time", "Moving Time", "s"), + ("distance", "Distance", "m"), + ("total_elevation_gain", "Elevation Gain", "m"), + ("training_load", "Training Load", ""), + ("calories", "Calories", "kcal"), + ("fitness", "Fitness (CTL)", ""), + ("fatigue", "Fatigue (ATL)", ""), + ("form", "Form (TSB)", ""), + ("rampRate", "Ramp Rate", ""), + ("eftp", "eFTP", "W"), + ]: + if summary.get(key) is not None: + lines.append(f"- {label}: {summary[key]}{(' ' + unit) if unit else ''}") + + categories = summary.get("byCategory") + if isinstance(categories, list) and categories: + lines.append("By category:") + for cat in categories: + if not isinstance(cat, dict): + continue + lines.append( + f" - {cat.get('category', '?')}: {cat.get('count', 0)} activities, " + f"load {cat.get('training_load', 'N/A')}, {cat.get('moving_time', 'N/A')}s" + ) + return "\n".join(lines) if lines else "No summary metrics available." + + def format_event_summary(event: dict[str, Any]) -> str: """Format a basic event summary into a readable string.""" diff --git a/tests/test_athlete.py b/tests/test_athlete.py new file mode 100644 index 0000000..5faa53f --- /dev/null +++ b/tests/test_athlete.py @@ -0,0 +1,188 @@ +""" +Tests for intervals_mcp_server.tools.athlete. + +Covers the athlete-context read tools (get_athlete_profile, get_sport_settings, +get_athlete_summary): request shape, sport filtering, formatting of realistic +fixtures, and the empty / error / credential branches. Default caller credentials +come from the autouse fixture in conftest (athlete ``i1``). +""" + +import asyncio + +from intervals_mcp_server import credentials +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.tools import athlete + +PROFILE = { + "id": "i1", + "name": "Test Athlete", + "sex": "M", + "weight": 72.5, + "icu_resting_hr": 48, + "timezone": "Europe/Madrid", + "measurement_preference": "meters", + "city": "Girona", + "country": "Spain", + "icu_coach": True, + "icu_type_settings": [{"id": 1}, {"id": 2}], +} + +SPORT_SETTINGS = [ + { + "id": 100, + "types": ["Ride", "VirtualRide"], + "ftp": 280, + "indoor_ftp": 275, + "w_prime": 22000, + "power_zones": [55, 75, 90, 105, 120], + "power_zone_names": ["Z1", "Z2", "Z3", "Z4", "Z5"], + "lthr": 165, + "max_hr": 190, + "hr_zones": [120, 145, 160, 175], + "threshold_pace": 4.2, + "pace_units": "MINS_KM", + "pace_zones": [3.5, 4.0, 4.5], + "warmup_time": 600, + "cooldown_time": 300, + }, + {"id": 101, "types": ["Run"], "threshold_pace": 3.8, "pace_units": "MINS_KM"}, +] + +SUMMARY = [ + { + "date": "2026-07-20", + "count": 12, + "moving_time": 43200, + "distance": 320000, + "training_load": 640, + "fitness": 78.5, + "fatigue": 71.0, + "form": 7.5, + "eftp": 285, + "byCategory": [{"category": "Ride", "count": 8, "training_load": 500, "moving_time": 32400}], + } +] + + +def _patch_request(monkeypatch, result): + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return result + + monkeypatch.setattr(athlete, "make_intervals_request", fake) + return calls + + +# --------------------------------------------------------------------------- # +# get_athlete_profile +# --------------------------------------------------------------------------- # +def test_get_athlete_profile_success(monkeypatch): + calls = _patch_request(monkeypatch, PROFILE) + out = asyncio.run(athlete.get_athlete_profile()) + assert calls[0]["url"] == "/athlete/i1" + assert "Name: Test Athlete" in out + assert "Weight: 72.5 kg" in out + assert "Resting HR: 48 bpm" in out + assert "Location: Girona, Spain" in out + assert "Role: Coach" in out + assert "2 sport(s) configured" in out + + +def test_get_athlete_profile_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "nope"}) + assert "Error fetching athlete profile: nope" in asyncio.run(athlete.get_athlete_profile()) + + +def test_get_athlete_profile_non_dict(monkeypatch): + _patch_request(monkeypatch, []) + assert "No athlete profile found" in asyncio.run(athlete.get_athlete_profile()) + + +def test_get_athlete_profile_credential_error(monkeypatch): + async def _deny(): + raise CredentialError("not approved") + + monkeypatch.setattr(credentials, "resolve_caller_credentials", _deny) + assert "not approved" in asyncio.run(athlete.get_athlete_profile()) + + +# --------------------------------------------------------------------------- # +# get_sport_settings +# --------------------------------------------------------------------------- # +def test_get_sport_settings_all(monkeypatch): + calls = _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings()) + assert calls[0]["url"] == "/athlete/i1/sport-settings" + assert "Sport Settings — Ride, VirtualRide" in out + assert "Settings ID: 100" in out + assert "FTP: 280W" in out + assert "Z1: 55, Z2: 75" in out # power zones paired with names + assert "LTHR: 165 bpm" in out + assert "Threshold: 4.2 MINS_KM" in out + assert "Settings ID: 101" in out # second record rendered too + + +def test_get_sport_settings_filter_hit(monkeypatch): + _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings(sport="run")) # case-insensitive + assert "Settings ID: 101" in out + assert "Settings ID: 100" not in out + + +def test_get_sport_settings_filter_miss(monkeypatch): + _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings(sport="Swim")) + assert "No sport settings found for sport 'Swim'" in out + + +def test_get_sport_settings_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No sport settings found" in asyncio.run(athlete.get_sport_settings()) + + +def test_get_sport_settings_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + assert "Error fetching sport settings: boom" in asyncio.run(athlete.get_sport_settings()) + + +# --------------------------------------------------------------------------- # +# get_athlete_summary +# --------------------------------------------------------------------------- # +def test_get_athlete_summary_success(monkeypatch): + calls = _patch_request(monkeypatch, SUMMARY) + out = asyncio.run(athlete.get_athlete_summary(start_date="2026-06-20", end_date="2026-07-20")) + call = calls[0] + assert call["url"] == "/athlete/i1/athlete-summary" + assert call["params"]["start"] == "2026-06-20" + assert call["params"]["end"] == "2026-07-20" + assert "tags" not in call["params"] + assert "Fitness (CTL): 78.5" in out + assert "Form (TSB): 7.5" in out + assert "By category:" in out + assert "Ride: 8 activities" in out + + +def test_get_athlete_summary_tags_split(monkeypatch): + calls = _patch_request(monkeypatch, SUMMARY) + asyncio.run(athlete.get_athlete_summary(tags="race, key-workout")) + assert calls[0]["params"]["tags"] == ["race", "key-workout"] + + +def test_get_athlete_summary_defaults_dates(monkeypatch): + calls = _patch_request(monkeypatch, SUMMARY) + asyncio.run(athlete.get_athlete_summary()) + # resolve_date_params fills both ends with YYYY-MM-DD + assert len(calls[0]["params"]["start"]) == 10 + assert len(calls[0]["params"]["end"]) == 10 + + +def test_get_athlete_summary_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No summary data found" in asyncio.run(athlete.get_athlete_summary()) + + +def test_get_athlete_summary_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "bad"}) + assert "Error fetching athlete summary: bad" in asyncio.run(athlete.get_athlete_summary())