farhoodlabs/intervalsicu-mcp
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat(athlete): add profile, sport-settings, and summary read tools

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN
This commit is contained in:
Chris Farhood
2026-07-20 15:44:05 -04:00
parent e724be283f
commit 28119f1761
5 changed files with 474 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/intervals_mcp_server/server.py
+9
View File
@@ -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",
src/intervals_mcp_server/tools/__init__.py
+8
View File
@@ -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",
]
src/intervals_mcp_server/tools/athlete.py
+129
View File
@@ -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)
src/intervals_mcp_server/utils/formatting.py
+140
View File
@@ -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."""
tests/test_athlete.py
+188
View File
@@ -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())