diff --git a/src/intervals_mcp_server/server.py b/src/intervals_mcp_server/server.py index 327f7c6..7e3d9fb 100644 --- a/src/intervals_mcp_server/server.py +++ b/src/intervals_mcp_server/server.py @@ -98,6 +98,10 @@ from intervals_mcp_server.tools.athlete import ( # pylint: disable=wrong-import get_athlete_summary, get_sport_settings, ) +from intervals_mcp_server.tools.workouts import ( # pylint: disable=wrong-import-position # noqa: E402 + get_workout, + get_workouts, +) 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 @@ -132,6 +136,8 @@ __all__ = [ "get_athlete_profile", "get_sport_settings", "get_athlete_summary", + "get_workouts", + "get_workout", "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 f897524..a010898 100644 --- a/src/intervals_mcp_server/tools/__init__.py +++ b/src/intervals_mcp_server/tools/__init__.py @@ -44,6 +44,7 @@ from intervals_mcp_server.tools.athlete import ( # noqa: F401 get_athlete_summary, get_sport_settings, ) +from intervals_mcp_server.tools.workouts import get_workout, get_workouts # noqa: F401 def register_tools(mcp_instance: FastMCP) -> None: @@ -88,4 +89,6 @@ __all__ = [ "get_athlete_profile", "get_sport_settings", "get_athlete_summary", + "get_workouts", + "get_workout", ] diff --git a/src/intervals_mcp_server/tools/workouts.py b/src/intervals_mcp_server/tools/workouts.py new file mode 100644 index 0000000..a33a5a8 --- /dev/null +++ b/src/intervals_mcp_server/tools/workouts.py @@ -0,0 +1,72 @@ +""" +Workout-library MCP tools for Intervals.icu. + +Read tools exposing the athlete's reusable workout library (distinct from the +calendar *events* handled in tools/events.py). +""" + +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_workout_details, format_workout_summary + +# Import mcp instance from shared module for tool registration +from intervals_mcp_server.mcp_instance import mcp # noqa: F401 + + +@mcp.tool() +async def get_workouts(folder_id: int | None = None, sport_type: str | None = None) -> str: + """List the athlete's reusable workout library. + + Filtering is applied client-side (the API returns the full library). + + Args: + folder_id: Optional folder ID to restrict results to one folder. + sport_type: Optional sport type to filter by (e.g. "Ride", "Run"). + """ + 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}/workouts", api_key=api_key) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching workouts: {result.get('message')}" + + workouts = [w for w in result if isinstance(w, dict)] if isinstance(result, list) else [] + if folder_id is not None: + workouts = [w for w in workouts if w.get("folder_id") == folder_id] + if sport_type: + want = sport_type.strip().lower() + workouts = [w for w in workouts if str(w.get("type", "")).lower() == want] + + if not workouts: + return "No workouts found." + + lines = [f"Workout Library ({len(workouts)}):", ""] + lines.extend(format_workout_summary(w) for w in workouts) + return "\n".join(lines) + + +@mcp.tool() +async def get_workout(workout_id: int) -> str: + """Get a single library workout's full structure (steps and targets). + + Args: + workout_id: The Intervals.icu workout ID. + """ + 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}/workouts/{workout_id}", api_key=api_key + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching workout: {result.get('message')}" + if not isinstance(result, dict) or not result: + return f"No workout found with ID {workout_id}." + return format_workout_details(result) diff --git a/src/intervals_mcp_server/utils/formatting.py b/src/intervals_mcp_server/utils/formatting.py index 089afc8..3833bb8 100644 --- a/src/intervals_mcp_server/utils/formatting.py +++ b/src/intervals_mcp_server/utils/formatting.py @@ -878,3 +878,101 @@ def format_interval_stats(interval: dict[str, Any]) -> str: if interval.get(key) is not None: lines.append(f"- {label}: {interval[key]}{(' ' + unit) if unit else ''}") return "\n".join(lines) + + +def _format_step_intensity(step: dict[str, Any]) -> str: + """Render a workout step's intensity target(s) from raw workout_doc JSON.""" + bits = [] + for key, label in [("power", ""), ("hr", "HR"), ("pace", "Pace"), ("cadence", "Cad")]: + v = step.get(key) + if not isinstance(v, dict): + continue + units = v.get("units", "") + if v.get("start") is not None and v.get("end") is not None: + val = f"{v['start']}-{v['end']}" + elif v.get("value") is not None: + val = f"{v['value']}" + else: + continue + bits.append(f"{(label + ' ') if label else ''}{val}{units}") + return ", ".join(bits) + + +def _format_workout_step(step: dict[str, Any], depth: int = 0) -> list[str]: + """Recursively render one workout_doc step (handles repeat blocks). Depth-capped.""" + indent = " " * (depth + 1) + if depth > 6: + return [f"{indent}- ...(nested too deep)"] + reps = step.get("reps") + substeps = step.get("steps") + if reps and isinstance(substeps, list): + lines = [f"{indent}{reps}x:"] + for sub in substeps: + if isinstance(sub, dict): + lines.extend(_format_workout_step(sub, depth + 1)) + return lines + parts = [] + if step.get("duration") is not None: + parts.append(_format_duration_label(int(step["duration"]))) + if step.get("distance") is not None: + parts.append(f"{step['distance']}m") + tag = "" + if step.get("warmup"): + tag = " (warmup)" + elif step.get("cooldown"): + tag = " (cooldown)" + elif step.get("freeride"): + tag = " (free ride)" + intensity = _format_step_intensity(step) + if step.get("ramp") and intensity: + intensity = "ramp " + intensity + label = " ".join(parts) if parts else "step" + detail = f" @ {intensity}" if intensity else "" + text = step.get("text") + return [f"{indent}- {label}{detail}{tag}{(' — ' + text) if text else ''}"] + + +def format_workout_summary(workout: dict[str, Any]) -> str: + """Format one library workout as a compact one-line list entry.""" + line = " | ".join([str(workout.get("name", "Unnamed")), str(workout.get("type", "?"))]) + extra = [] + if workout.get("icu_training_load") is not None: + extra.append(f"load {workout['icu_training_load']}") + if workout.get("moving_time") is not None: + extra.append(f"{workout['moving_time']}s") + if workout.get("folder_id") is not None: + extra.append(f"folder {workout['folder_id']}") + if extra: + line += " (" + ", ".join(extra) + ")" + return line + f" [id: {workout.get('id', 'N/A')}]" + + +def format_workout_details(workout: dict[str, Any]) -> str: + """Format a library workout in full, including its structured steps.""" + lines = [f"Workout: {workout.get('name', 'Unnamed')}", f"ID: {workout.get('id', 'N/A')}"] + for key, label, unit in [ + ("type", "Type", ""), + ("sub_type", "Sub-type", ""), + ("indoor", "Indoor", ""), + ("moving_time", "Duration", "s"), + ("distance", "Distance", "m"), + ("icu_training_load", "Training Load", ""), + ("icu_intensity", "Intensity", ""), + ("carbs_per_hour", "Carbs", "g/hr"), + ("folder_id", "Folder", ""), + ]: + if workout.get(key) is not None: + lines.append(f"{label}: {workout[key]}{(' ' + unit) if unit else ''}") + if workout.get("description"): + lines.append(f"Description: {workout['description']}") + tags = workout.get("tags") + if isinstance(tags, list) and tags: + lines.append("Tags: " + ", ".join(str(t) for t in tags)) + + doc = workout.get("workout_doc") + if isinstance(doc, dict) and isinstance(doc.get("steps"), list) and doc["steps"]: + lines += ["", "Steps:"] + for step in doc["steps"]: + if isinstance(step, dict): + lines.extend(_format_workout_step(step)) + return "\n".join(lines) diff --git a/tests/test_workouts.py b/tests/test_workouts.py new file mode 100644 index 0000000..92c7052 --- /dev/null +++ b/tests/test_workouts.py @@ -0,0 +1,131 @@ +""" +Tests for intervals_mcp_server.tools.workouts (0.3.0 workout library). + +Covers get_workouts (list + client-side filters) and get_workout (full detail +with a nested workout_doc), plus empty / error / credential branches. +""" + +import asyncio + +from intervals_mcp_server import credentials +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.tools import workouts + +LIBRARY = [ + {"id": 10, "name": "VO2 5x5", "type": "Ride", "icu_training_load": 95, "moving_time": 3600, "folder_id": 1}, + {"id": 11, "name": "Easy run", "type": "Run", "moving_time": 2400, "folder_id": 2}, +] + +WORKOUT_DETAIL = { + "id": 10, + "name": "VO2 5x5", + "type": "Ride", + "indoor": True, + "moving_time": 3600, + "icu_training_load": 95, + "description": "VO2max builder", + "tags": ["vo2", "key"], + "workout_doc": { + "steps": [ + {"duration": 900, "power": {"value": 60, "units": "%ftp"}, "warmup": True}, + { + "reps": 5, + "steps": [ + {"duration": 300, "power": {"value": 115, "units": "%ftp"}, "text": "hard"}, + {"duration": 300, "power": {"value": 50, "units": "%ftp"}, "text": "easy"}, + ], + }, + {"duration": 600, "power": {"start": 60, "end": 40, "units": "%ftp"}, "cooldown": True}, + ] + }, +} + + +def _patch_request(monkeypatch, result): + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return result + + monkeypatch.setattr(workouts, "make_intervals_request", fake) + return calls + + +def test_get_workouts_all(monkeypatch): + calls = _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts()) + assert calls[0]["url"] == "/athlete/i1/workouts" + assert "Workout Library (2)" in out + assert "VO2 5x5 | Ride (load 95, 3600s, folder 1) [id: 10]" in out + + +def test_get_workouts_filter_folder(monkeypatch): + _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts(folder_id=2)) + assert "Easy run" in out + assert "VO2 5x5" not in out + + +def test_get_workouts_filter_sport(monkeypatch): + _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts(sport_type="ride")) + assert "VO2 5x5" in out + assert "Easy run" not in out + + +def test_get_workouts_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No workouts found" in asyncio.run(workouts.get_workouts()) + + +def test_get_workouts_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + assert "Error fetching workouts: boom" in asyncio.run(workouts.get_workouts()) + + +def test_get_workout_detail_with_nested_doc(monkeypatch): + calls = _patch_request(monkeypatch, WORKOUT_DETAIL) + out = asyncio.run(workouts.get_workout(10)) + assert calls[0]["url"] == "/athlete/i1/workouts/10" + assert "Workout: VO2 5x5" in out + assert "Training Load: 95" in out + assert "Tags: vo2, key" in out + assert "Steps:" in out + assert "15m @ 60%ftp (warmup)" in out + assert "5x:" in out # repeat block rendered + assert "5m @ 115%ftp — hard" in out # nested step + assert "10m @ 60-40%ftp (cooldown)" in out # power range (no ramp flag set) + + +def test_get_workout_ramp_step(monkeypatch): + _patch_request( + monkeypatch, + { + "id": 12, + "name": "Ramp test", + "workout_doc": { + "steps": [{"ramp": True, "power": {"start": 100, "end": 300, "units": "w"}}] + }, + }, + ) + out = asyncio.run(workouts.get_workout(12)) + assert "ramp 100-300w" in out + + +def test_get_workout_not_found(monkeypatch): + _patch_request(monkeypatch, {}) + assert "No workout found with ID 99" in asyncio.run(workouts.get_workout(99)) + + +def test_get_workout_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "nope"}) + assert "Error fetching workout: nope" in asyncio.run(workouts.get_workout(10)) + + +def test_get_workout_credential_error(monkeypatch): + async def _deny(): + raise CredentialError("not approved") + + monkeypatch.setattr(credentials, "resolve_caller_credentials", _deny) + assert "not approved" in asyncio.run(workouts.get_workout(10))