feat(workouts): add workout library read tools (get_workouts, get_workout)

get_workouts lists the reusable library (client-side folder/sport filters);
get_workout renders a single workout including its structured workout_doc steps
via a defensive, depth-capped recursive formatter (repeats, ramps, warmup/
cooldown, power/hr/pace targets). New tools/workouts.py; registered.

Implements #4.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN
This commit is contained in:
2026-07-20 15:49:17 -04:00
parent 65585c53b5
commit e18e05e02c
5 changed files with 310 additions and 0 deletions
+6
View File
@@ -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",
@@ -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",
]
@@ -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)
@@ -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)
+131
View File
@@ -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))