feat: add ATP season planning and race event tools#112
Open
Jochem-van-Appeldoorn wants to merge 15 commits into
Open
feat: add ATP season planning and race event tools#112Jochem-van-Appeldoorn wants to merge 15 commits into
Jochem-van-Appeldoorn wants to merge 15 commits into
Conversation
Adds four new MCP tools for training plan management: - create_atp_plan: auto-determines phases (Base/Build/Peak/Race) based on current CTL vs goal CTL and posts each phase as a NOTE event - get_atp_plan: reads all ATP phase notes from the calendar - get_atp_week_note: returns the phase note covering a specific week - get_planning_context: single call returning ATP note + CTL/ATL/TSB/HRV + scheduled workouts + upcoming races for weekly planning Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- Fix mypy has-type errors: use _raw list with explicit type annotations instead of reassigning gather results (asyncio.gather return_exceptions=True) - Add noqa: F401 to get_gear_list re-export in server.py (used by tests) - Remove unused resolve_activity_type import from power_curves.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- add_race_event: creates RACE_A/B/C events with correct API fields (start_date_local as datetime, moving_time in seconds, icu_training_load) - Fix get_planning_context to filter on RACE_A/RACE_B/RACE_C instead of the incorrect A/B/C category names Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
All phase labels, focus/intensity descriptions, note headers, error messages and section headers in planning.py are now in English. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Helps diagnose whether the Intervals.icu API accepts RACE_A/B/C as category on POST or silently overrides it. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds an Annual Training Plan (ATP) feature set as a new MCP tool module, with logic to auto-select training phases based on the gap between current and goal CTL, and registers the new tools.
Changes:
- New
planning.pymodule with five MCP tools:create_atp_plan,get_atp_plan,get_atp_week_note,get_planning_context,add_race_event. - Registers the new planning tools in
tools/__init__.py. - Minor import cleanup in
power_curves.pyandserver.py.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 7 comments.
| File | Description |
|---|---|
| src/intervals_mcp_server/tools/planning.py | New module implementing ATP phase generation, plan retrieval, weekly planning context, and race event creation. |
| src/intervals_mcp_server/tools/init.py | Exposes and re-exports new planning tools. |
| src/intervals_mcp_server/tools/power_curves.py | Removes unused resolve_activity_type import. |
| src/intervals_mcp_server/server.py | Adds F401 to get_gear_list import noqa to silence unused-import warning. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Filter ATP notes by name prefix 'ATP' in get_atp_plan, get_atp_week_note and get_planning_context to avoid mixing in unrelated user notes - Validate date and priority in additional_races parsing - Validate start_time format (HH:MM:SS) in add_race_event - Apply consistent build_wks < 3 guard in large-gap branch of _determine_phases (was build_wks < 1, inconsistent with moderate-gap) - Remove unused _recovery_week_numbers helper Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+129
to
+133
| def _reason_for_phases(current_ctl: float, goal_ctl: float, total_weeks: int) -> str: | ||
| ratio = current_ctl / max(goal_ctl, 1.0) | ||
| gap = round(goal_ctl - current_ctl) | ||
| if total_weeks <= 5: | ||
| return f"Only {total_weeks} weeks available — going straight to peak phase." |
Comment on lines
+360
to
+367
| event_data: dict[str, Any] = { | ||
| "category": "NOTE", | ||
| "name": f"ATP — {_PHASES[ph['name']]['label']}", | ||
| "description": description, | ||
| "start_date_local": ph["start"].isoformat() + "T00:00:00", | ||
| "end_date_local": ph["end"].isoformat() + "T00:00:00", | ||
| "color": _PHASES[ph["name"]]["color"], | ||
| } |
Comment on lines
+742
to
+743
| if distance_km is not None: | ||
| event_data["distance"] = int(distance_km * 1000) |
Comment on lines
+163
to
+170
| def _week_tss(phase: str, goal_tss: int, week: int, total_weeks: int, cycle: int) -> str: | ||
| factor = _PHASES[phase]["tss_factor"] | ||
| if week % cycle == 0: | ||
| return f"~{round(goal_tss * factor * 0.60 / 50) * 50} TSS ← recovery week" | ||
| load_weeks = [w for w in range(1, total_weeks + 1) if w % cycle != 0] | ||
| idx = load_weeks.index(week) if week in load_weeks else 0 | ||
| prog = 0.85 + 0.15 * (idx / max(len(load_weeks) - 1, 1)) | ||
| tss = round(goal_tss * factor * prog / 50) * 50 |
Comment on lines
+443
to
+447
| events = result if isinstance(result, list) else [] | ||
| notes = [ | ||
| e for e in events | ||
| if e.get("category") == "NOTE" and (e.get("name") or "").startswith("ATP") | ||
| ] |
- _reason_for_phases: derive actual phase sequence from _determine_phases instead of hardcoding inaccurate "going straight to peak phase" message - end_date_local: add +1 day so calendar events cover through the last day (exclusive endpoint semantics used by Intervals) - distance_km → meters: use round() instead of int() to avoid float truncation - _week_tss: accept precomputed load_weeks to eliminate O(n²) rebuild per week - notes filter: guard with isinstance(e, dict) before calling .get() Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+85
to
+92
| peak_wks = min(2, max(1, total_weeks // 7)) | ||
| race_wks = 1 | ||
| remaining = total_weeks - peak_wks - race_wks | ||
|
|
||
| tail = [("peak", peak_wks), ("race", race_wks)] | ||
|
|
||
| if remaining <= 0: | ||
| return tail |
Comment on lines
+459
to
+463
| name = note.get("name", "") | ||
| s = (note.get("start_date_local") or "")[:10] | ||
| e_end = (note.get("end_date_local") or note.get("start_date_local") or "")[:10] | ||
| desc = note.get("description", "") | ||
| lines.append(f"## {name} ({s} – {e_end})") |
Comment on lines
+604
to
+607
| atp_notes = [ | ||
| e for e in week_list | ||
| if e.get("category") == "NOTE" and (e.get("name") or "").startswith("ATP") | ||
| ] |
- _determine_phases: clamp peak_wks + race_wks <= total_weeks to prevent invalid phase date ranges when total_weeks is very small (e.g. 1–2 weeks) - get_atp_plan display: subtract 1 day from end_date_local before rendering so the shown range matches the inclusive end stored on the event - get_planning_context: add isinstance(e, dict) guard to atp_notes, workouts, and races comprehensions for consistency with other filtering paths Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+85
to
+90
| if total_weeks <= 1: | ||
| return [("race", total_weeks)] | ||
| race_wks = 1 | ||
| peak_wks = min(2, max(1, (total_weeks - race_wks) // 7)) | ||
| if peak_wks + race_wks >= total_weeks: | ||
| peak_wks = total_weeks - race_wks |
Comment on lines
+169
to
+174
| def _week_tss(phase: str, goal_tss: int, week: int, cycle: int, load_weeks: list[int]) -> str: | ||
| factor = _PHASES[phase]["tss_factor"] | ||
| if week % cycle == 0: | ||
| return f"~{round(goal_tss * factor * 0.60 / 50) * 50} TSS ← recovery week" | ||
| idx = load_weeks.index(week) if week in load_weeks else 0 | ||
| prog = 0.85 + 0.15 * (idx / max(len(load_weeks) - 1, 1)) |
Comment on lines
+737
to
+740
| from datetime import datetime as _dt | ||
| try: | ||
| _dt.strptime(start_time, "%H:%M:%S") | ||
| except ValueError: |
Comment on lines
+315
to
+318
| for line in additional_races.strip().splitlines(): | ||
| parts = line.strip().split(None, 1) # split on first space only: [date, rest] | ||
| if len(parts) != 2: | ||
| continue |
- _determine_phases: compute peak_wks directly (2 if enough weeks, else 1)
instead of the broken //7 scaling that almost always yielded 1 week
- _week_tss: replace two O(n) list scans with O(1) dict lookup by accepting
a precomputed {week: idx} mapping from _build_phase_note
- additional_races parsing: return a clear error on malformed lines instead
of silently skipping them
- add_race_event: remove inline datetime import; use module-level datetime
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+134
to
+154
| def _reason_for_phases(current_ctl: float, goal_ctl: float, total_weeks: int) -> str: | ||
| ratio = current_ctl / max(goal_ctl, 1.0) | ||
| gap = round(goal_ctl - current_ctl) | ||
| if total_weeks <= 5: | ||
| phases = _determine_phases(total_weeks, current_ctl, goal_ctl) | ||
| phase_seq = " → ".join(_PHASES[name]["label"] for name, _ in phases) | ||
| return f"Only {total_weeks} weeks available — {phase_seq}." | ||
| if ratio >= 0.90: | ||
| return ( | ||
| f"CTL {round(current_ctl)} is close to goal CTL {goal_ctl} " | ||
| f"(gap: {gap}) — going straight to build phase." | ||
| ) | ||
| if ratio >= 0.70: | ||
| return ( | ||
| f"CTL {round(current_ctl)} is {gap} points below goal CTL {goal_ctl} " | ||
| f"({round(ratio * 100)}%) — short base block followed by build phase." | ||
| ) | ||
| return ( | ||
| f"CTL {round(current_ctl)} is {gap} points below goal CTL {goal_ctl} " | ||
| f"({round(ratio * 100)}%) — preparation and base block needed before build." | ||
| ) |
Comment on lines
+352
to
+381
| for i, ph in enumerate(phase_ranges, start=1): | ||
| phase_races = [r for r in extra_races if ph["start"].isoformat() <= r["date"] <= ph["end"].isoformat()] | ||
| description = _build_phase_note( | ||
| phase=ph["name"], | ||
| phase_num=i, | ||
| total_phases=total_phases, | ||
| start=ph["start"], | ||
| end=ph["end"], | ||
| cycle=recovery_cycle, | ||
| goal_tss=goal_weekly_tss, | ||
| race_name=race_name, | ||
| race_date=race_dt, | ||
| phase_races=phase_races, | ||
| ) | ||
|
|
||
| event_data: dict[str, Any] = { | ||
| "category": "NOTE", | ||
| "name": f"ATP — {_PHASES[ph['name']]['label']}", | ||
| "description": description, | ||
| "start_date_local": ph["start"].isoformat() + "T00:00:00", | ||
| "end_date_local": (ph["end"] + timedelta(days=1)).isoformat() + "T00:00:00", | ||
| "color": _PHASES[ph["name"]]["color"], | ||
| } | ||
|
|
||
| result = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| data=event_data, | ||
| method="POST", | ||
| ) |
Comment on lines
+450
to
+454
| events = result if isinstance(result, list) else [] | ||
| notes = [ | ||
| e for e in events | ||
| if isinstance(e, dict) and e.get("category") == "NOTE" and (e.get("name") or "").startswith("ATP") | ||
| ] |
- _reason_for_phases: handle gap <= 0 (current CTL >= goal) with accurate message; also round goal_ctl to int for clean user-facing strings - create_atp_plan: delete existing ATP notes in plan window before posting so re-running the tool is idempotent and doesn't clutter the calendar - extract _ATP_PREFIX constant and _is_atp_note() helper; replace three duplicated category/prefix filter expressions across get_atp_plan, get_atp_week_note, and get_planning_context Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+333
to
+337
| if additional_races: | ||
| for line in additional_races.strip().splitlines(): | ||
| parts = line.strip().split(None, 1) # split on first space only: [date, rest] | ||
| if len(parts) != 2: | ||
| return f"Error: malformed additional_races line '{line.strip()}'. Expected format: 'YYYY-MM-DD Race name [A/B/C]'." |
Comment on lines
+343
to
+348
| priority = "C" | ||
| name_part = rest | ||
| if "[" in rest: | ||
| priority = rest.split("[")[-1].strip("] ").upper() | ||
| name_part = rest.split("[")[0].strip() | ||
| if priority not in ("A", "B", "C"): |
Comment on lines
+245
to
+249
| async def create_atp_plan( | ||
| race_date: str, | ||
| race_name: str, | ||
| goal_ctl: int, | ||
| recovery_cycle: int = 4, |
Comment on lines
+606
to
+639
| _raw = await asyncio.gather( | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/wellness", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": (today - timedelta(days=14)).isoformat(), | ||
| "newest": today.isoformat(), | ||
| }, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": monday.isoformat(), "newest": sunday.isoformat()}, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": today.isoformat(), | ||
| "newest": (today + timedelta(days=120)).isoformat(), | ||
| }, | ||
| ), | ||
| return_exceptions=True, | ||
| ) | ||
| _fallback: list[Any] = [] | ||
| wellness_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[0] if not isinstance(_raw[0], BaseException) else _fallback | ||
| ) | ||
| week_events: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[1] if not isinstance(_raw[1], BaseException) else _fallback | ||
| ) | ||
| races_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[2] if not isinstance(_raw[2], BaseException) else _fallback | ||
| ) |
- create_atp_plan: validate goal_ctl >= 1 early with a clear error message - additional_races parsing: skip blank/whitespace lines; use trailing-only regex [A/B/C] match so brackets in race names aren't mis-parsed as priority - get_planning_context: collect and surface API call failures as a warnings section instead of silently collapsing exceptions to empty lists - move import re to module level Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+611
to
+640
| _raw = await asyncio.gather( | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/wellness", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": (today - timedelta(days=14)).isoformat(), | ||
| "newest": today.isoformat(), | ||
| }, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": monday.isoformat(), "newest": sunday.isoformat()}, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": today.isoformat(), | ||
| "newest": (today + timedelta(days=120)).isoformat(), | ||
| }, | ||
| ), | ||
| return_exceptions=True, | ||
| ) | ||
| _fallback: list[Any] = [] | ||
| _api_warnings: list[str] = [] | ||
| _labels = ("wellness", "week events", "upcoming races") | ||
| for _label, _raw_item in zip(_labels, _raw): | ||
| if isinstance(_raw_item, BaseException): | ||
| _api_warnings.append(f" • {_label}: {_raw_item}") |
Comment on lines
+364
to
+370
| for ev in (existing if isinstance(existing, list) else []): | ||
| if _is_atp_note(ev) and isinstance(ev, dict) and ev.get("id"): | ||
| await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events/{ev['id']}", | ||
| api_key=api_key, | ||
| method="DELETE", | ||
| ) |
Comment on lines
+390
to
+419
| for i, ph in enumerate(phase_ranges, start=1): | ||
| phase_races = [r for r in extra_races if ph["start"].isoformat() <= r["date"] <= ph["end"].isoformat()] | ||
| description = _build_phase_note( | ||
| phase=ph["name"], | ||
| phase_num=i, | ||
| total_phases=total_phases, | ||
| start=ph["start"], | ||
| end=ph["end"], | ||
| cycle=recovery_cycle, | ||
| goal_tss=goal_weekly_tss, | ||
| race_name=race_name, | ||
| race_date=race_dt, | ||
| phase_races=phase_races, | ||
| ) | ||
|
|
||
| event_data: dict[str, Any] = { | ||
| "category": "NOTE", | ||
| "name": f"ATP — {_PHASES[ph['name']]['label']}", | ||
| "description": description, | ||
| "start_date_local": ph["start"].isoformat() + "T00:00:00", | ||
| "end_date_local": (ph["end"] + timedelta(days=1)).isoformat() + "T00:00:00", | ||
| "color": _PHASES[ph["name"]]["color"], | ||
| } | ||
|
|
||
| result = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| data=event_data, | ||
| method="POST", | ||
| ) |
- get_planning_context: check asyncio.CancelledError before Exception so task cancellation propagates correctly; use Exception instead of BaseException - create_atp_plan: delete existing ATP notes concurrently (semaphore=3) instead of serially to reduce latency on large calendars - create_atp_plan: post all phase events concurrently (semaphore=3); roll back any created events and return a clear error if any POST fails Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+357
to
+380
| # Delete any existing ATP notes in the plan window to avoid duplicates | ||
| plan_end = _monday_of(race_dt) + timedelta(days=6) | ||
| existing = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": today_monday.isoformat(), "newest": plan_end.isoformat()}, | ||
| ) | ||
| _sem = asyncio.Semaphore(3) | ||
|
|
||
| async def _delete(ev_id: str) -> None: | ||
| async with _sem: | ||
| await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events/{ev_id}", | ||
| api_key=api_key, | ||
| method="DELETE", | ||
| ) | ||
|
|
||
| to_delete = [ | ||
| ev["id"] for ev in (existing if isinstance(existing, list) else []) | ||
| if _is_atp_note(ev) and isinstance(ev, dict) and ev.get("id") | ||
| ] | ||
| if to_delete: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in to_delete)) | ||
|
|
Comment on lines
+668
to
+681
| for _label, _raw_item in zip(_labels, _raw): | ||
| if isinstance(_raw_item, asyncio.CancelledError): | ||
| raise _raw_item | ||
| if isinstance(_raw_item, Exception): | ||
| _api_warnings.append(f" • {_label}: {_raw_item}") | ||
| wellness_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[0] if not isinstance(_raw[0], Exception) else _fallback | ||
| ) | ||
| week_events: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[1] if not isinstance(_raw[1], Exception) else _fallback | ||
| ) | ||
| races_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[2] if not isinstance(_raw[2], Exception) else _fallback | ||
| ) |
Comment on lines
+443
to
+447
| failed = [r for r in post_results if isinstance(r, Exception) or (isinstance(r, dict) and "error" in r)] | ||
| if failed: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in created_ids)) | ||
| err_msg = failed[0].get("message") if isinstance(failed[0], dict) else str(failed[0]) | ||
| return f"Error creating ATP plan (all changes rolled back): {err_msg}" |
Comment on lines
+518
to
+543
| events = result if isinstance(result, list) else [] | ||
| notes = [ | ||
| e for e in events | ||
| if _is_atp_note(e) | ||
| ] | ||
|
|
||
| if not notes: | ||
| return ( | ||
| f"No ATP notes found between {start_date} and {end_date}. " | ||
| "Use create_atp_plan to create a plan." | ||
| ) | ||
|
|
||
| lines = [f"ATP overview ({start_date} – {end_date})\n"] | ||
| for note in notes: | ||
| name = note.get("name", "") | ||
| s = (note.get("start_date_local") or "")[:10] | ||
| e_end_raw = (note.get("end_date_local") or note.get("start_date_local") or "")[:10] | ||
| try: | ||
| e_end = (date.fromisoformat(e_end_raw) - timedelta(days=1)).isoformat() if e_end_raw else "" | ||
| except ValueError: | ||
| e_end = e_end_raw | ||
| desc = note.get("description", "") | ||
| lines.append(f"## {name} ({s} – {e_end})") | ||
| if desc: | ||
| lines.append(desc) | ||
| lines.append("") |
Comment on lines
+337
to
+339
| if additional_races: | ||
| _trailing_priority = re.compile(r'^(.*?)\s*\[([A-Ca-c])\]\s*$') | ||
| for line in additional_races.strip().splitlines(): |
Comment on lines
+299
to
+303
| if goal_ctl < 1: | ||
| return "Error: goal_ctl must be at least 1." | ||
|
|
||
| if recovery_cycle not in (3, 4): | ||
| return "Error: recovery_cycle must be 3 or 4." |
Comment on lines
+332
to
+333
| # Weekly TSS in peak build weeks ≈ goal_ctl × 7 | ||
| goal_weekly_tss = round(goal_ctl * 7 / 50) * 50 |
Comment on lines
+443
to
+455
| failed = [r for r in post_results if isinstance(r, Exception) or (isinstance(r, dict) and "error" in r)] | ||
| if failed: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in created_ids)) | ||
| err_msg = failed[0].get("message") if isinstance(failed[0], dict) else str(failed[0]) | ||
| return f"Error creating ATP plan (all changes rolled back): {err_msg}" | ||
|
|
||
| posted_lines: list[str] = [] | ||
| for (ph, _), result in zip(phase_events, post_results): | ||
| label = _PHASES[ph["name"]]["label"] | ||
| if isinstance(result, dict) and "error" in result: | ||
| posted_lines.append(f" ✗ {label} ({ph['start']} – {ph['end']}): {result.get('message')}") | ||
| else: | ||
| posted_lines.append(f" ✓ {label}: {ph['start']} – {ph['end']} ({ph['weeks']} wk)") |
| ), | ||
| return_exceptions=True, | ||
| ) | ||
| _fallback: list[Any] = [] |
Comment on lines
+673
to
+681
| wellness_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[0] if not isinstance(_raw[0], Exception) else _fallback | ||
| ) | ||
| week_events: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[1] if not isinstance(_raw[1], Exception) else _fallback | ||
| ) | ||
| races_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[2] if not isinstance(_raw[2], Exception) else _fallback | ||
| ) |
Comment on lines
+74
to
+82
| _ATP_PREFIX = "ATP" | ||
|
|
||
|
|
||
| def _is_atp_note(e: object) -> bool: | ||
| return ( | ||
| isinstance(e, dict) | ||
| and e.get("category") == "NOTE" # type: ignore[union-attr] | ||
| and (e.get("name") or "").startswith(_ATP_PREFIX) # type: ignore[union-attr] | ||
| ) |
Comment on lines
+357
to
+363
| # Delete any existing ATP notes in the plan window to avoid duplicates | ||
| plan_end = _monday_of(race_dt) + timedelta(days=6) | ||
| existing = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": today_monday.isoformat(), "newest": plan_end.isoformat()}, | ||
| ) |
Comment on lines
+444
to
+447
| if failed: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in created_ids)) | ||
| err_msg = failed[0].get("message") if isinstance(failed[0], dict) else str(failed[0]) | ||
| return f"Error creating ATP plan (all changes rolled back): {err_msg}" |
- narrow _ATP_PREFIX to "ATP — " so _is_atp_note/delete only matches tool-created notes, not any user note starting with "ATP" - clamp goal_weekly_tss to min 50 so very low goal_ctl values don't produce zero TSS guidance - fix err_msg fallback to "error" key and str(f) so it's never None - remove unreachable ✗ branch in the posting loop (all error dicts already trigger rollback+return before the loop runs) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+357
to
+379
| # Delete any existing ATP notes in the plan window to avoid duplicates | ||
| plan_end = _monday_of(race_dt) + timedelta(days=6) | ||
| existing = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": today_monday.isoformat(), "newest": plan_end.isoformat()}, | ||
| ) | ||
| _sem = asyncio.Semaphore(3) | ||
|
|
||
| async def _delete(ev_id: str) -> None: | ||
| async with _sem: | ||
| await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events/{ev_id}", | ||
| api_key=api_key, | ||
| method="DELETE", | ||
| ) | ||
|
|
||
| to_delete = [ | ||
| ev["id"] for ev in (existing if isinstance(existing, list) else []) | ||
| if _is_atp_note(ev) and isinstance(ev, dict) and ev.get("id") | ||
| ] | ||
| if to_delete: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in to_delete)) |
Comment on lines
+374
to
+377
| to_delete = [ | ||
| ev["id"] for ev in (existing if isinstance(existing, list) else []) | ||
| if _is_atp_note(ev) and isinstance(ev, dict) and ev.get("id") | ||
| ] |
Comment on lines
+639
to
+679
| _raw = await asyncio.gather( | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/wellness", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": (today - timedelta(days=14)).isoformat(), | ||
| "newest": today.isoformat(), | ||
| }, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": monday.isoformat(), "newest": sunday.isoformat()}, | ||
| ), | ||
| make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={ | ||
| "oldest": today.isoformat(), | ||
| "newest": (today + timedelta(days=120)).isoformat(), | ||
| }, | ||
| ), | ||
| return_exceptions=True, | ||
| ) | ||
| _fallback: list[Any] = [] | ||
| _api_warnings: list[str] = [] | ||
| _labels = ("wellness", "week events", "upcoming races") | ||
| for _label, _raw_item in zip(_labels, _raw): | ||
| if isinstance(_raw_item, asyncio.CancelledError): | ||
| raise _raw_item | ||
| if isinstance(_raw_item, Exception): | ||
| _api_warnings.append(f" • {_label}: {_raw_item}") | ||
| wellness_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[0] if not isinstance(_raw[0], Exception) else _fallback | ||
| ) | ||
| week_events: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[1] if not isinstance(_raw[1], Exception) else _fallback | ||
| ) | ||
| races_result: dict[str, Any] | list[dict[str, Any]] = ( | ||
| _raw[2] if not isinstance(_raw[2], Exception) else _fallback | ||
| ) |
Comment on lines
+517
to
+520
| notes = [ | ||
| e for e in events | ||
| if _is_atp_note(e) | ||
| ] |
- create_atp_plan: fetch old notes first but delete them only after all
POSTs succeed, so a failed re-plan never loses the existing calendar data
- old_ids collection: use ev.get("id") consistently, no KeyError risk
- get_planning_context: surface API error dicts (not just exceptions) in
_api_warnings via _is_failed/_warn_msg helpers; use _fallback for failed
dict responses so downstream sees empty list, not an error dict
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+444
to
+451
| await asyncio.gather(*(_delete(ev_id) for ev_id in created_ids)) | ||
| f = failed[0] | ||
| err_msg = (f.get("message") or f.get("error") or str(f)) if isinstance(f, dict) else str(f) | ||
| return f"Error creating ATP plan (rolled back, existing plan preserved): {err_msg}" | ||
|
|
||
| # All POSTs succeeded — now safe to remove the old notes | ||
| if old_ids: | ||
| await asyncio.gather(*(_delete(ev_id) for ev_id in old_ids)) |
Comment on lines
+189
to
+196
| def _week_tss(phase: str, goal_tss: int, week: int, cycle: int, load_week_idx: dict[int, int]) -> str: | ||
| factor = _PHASES[phase]["tss_factor"] | ||
| if week % cycle == 0: | ||
| return f"~{round(goal_tss * factor * 0.60 / 50) * 50} TSS ← recovery week" | ||
| idx = load_week_idx.get(week, 0) | ||
| prog = 0.85 + 0.15 * (idx / max(len(load_week_idx) - 1, 1)) | ||
| tss = round(goal_tss * factor * prog / 50) * 50 | ||
| return f"~{tss} TSS" |
Comment on lines
+516
to
+517
| if isinstance(result, dict) and "error" in result: | ||
| return f"Error fetching ATP plan: {result.get('message')}" |
Comment on lines
+853
to
+854
| if isinstance(result, dict) and "error" in result: | ||
| return f"Error adding race: {result.get('message')}" |
Comment on lines
+338
to
+339
| _trailing_priority = re.compile(r'^(.*?)\s*\[([A-Ca-c])\]\s*$') | ||
| for line in additional_races.strip().splitlines(): |
Internal review: - race_dt <= today → < today: same-day races were incorrectly rejected - _determine_phases: explicit > 0 guards before appending base/build phases - old_ids: simplify cryptic tuple-unwrap comprehension - recovery_factor moved into _PHASES dict; _week_tss reads it per phase instead of using hardcoded 0.60 everywhere - _TRAILING_PRIORITY_RE: move regex to module scope (was re-compiled per call) Copilot round 10: - asyncio.gather for rollback and old-note deletes now use return_exceptions=True; delete failures are surfaced in the return message instead of raising - _week_tss: clamp computed TSS to min 50 for both load and recovery weeks - Error handlers in get_atp_plan and add_race_event: fall back to 'error' key and str(result) when 'message' is missing, so user never sees None Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Comment on lines
+380
to
+386
| async def _delete(ev_id: str) -> None: | ||
| async with _sem: | ||
| await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events/{ev_id}", | ||
| api_key=api_key, | ||
| method="DELETE", | ||
| ) |
Comment on lines
+459
to
+465
| delete_results = await asyncio.gather(*(_delete(ev_id) for ev_id in old_ids), return_exceptions=True) | ||
| delete_failures = [r for r in delete_results if isinstance(r, Exception)] | ||
| if delete_failures: | ||
| return ( | ||
| f"ATP plan created but {len(delete_failures)} old note(s) could not be deleted " | ||
| f"(may cause duplicates): {delete_failures[0]}" | ||
| ) |
Comment on lines
+366
to
+376
| # Fetch existing ATP notes now; delete them only after new ones are created | ||
| plan_end = _monday_of(race_dt) + timedelta(days=6) | ||
| existing = await make_intervals_request( | ||
| url=f"/athlete/{athlete_id_to_use}/events", | ||
| api_key=api_key, | ||
| params={"oldest": today_monday.isoformat(), "newest": plan_end.isoformat()}, | ||
| ) | ||
| old_ids = [ | ||
| ev.get("id") for ev in (existing if isinstance(existing, list) else []) | ||
| if _is_atp_note(ev) and isinstance(ev, dict) and ev.get("id") | ||
| ] |
Comment on lines
+597
to
+598
| if isinstance(result, dict) and "error" in result: | ||
| return f"Error fetching events: {result.get('message')}" |
Comment on lines
+533
to
+537
| events = result if isinstance(result, list) else [] | ||
| notes = [ | ||
| e for e in events | ||
| if _is_atp_note(e) | ||
| ] |
Comment on lines
+545
to
+547
| lines = [f"ATP overview ({start_date} – {end_date})\n"] | ||
| for note in notes: | ||
| name = note.get("name", "") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR adds a new
planningmodule with five MCP tools that give an AI assistant full season-planning capabilities on top of the Intervals.icu calendar.New tools
create_atp_planBuilds a complete Annual Training Plan from today up to an A-race date.
current_ctl / goal_ctl(scales to any athlete level — no hardcoded thresholds):get_atp_planReads all ATP phase notes from the calendar for a given date range and returns a structured overview of the season plan.
get_atp_week_noteReturns the ATP phase note for the week containing a given date — useful for week-by-week planning context (phase focus, TSS target, key sessions).
get_planning_contextOne-shot context snapshot for planning a specific week, combining:
add_race_eventCreates a race event on the calendar with the correct Intervals.icu fields:
category:RACE_A,RACE_B, orRACE_C(priority embedded in category name)type: sport (Ride, Run, Swim, …)start_date_local:YYYY-MM-DDTHH:MM:SSmoving_time: expected duration in secondsdistance: metersicu_training_load: expected TSSOther changes
tools/__init__.py: registers all five new tools for exporttools/power_curves.py: removed genuinely unused import flagged by rufftools/server.py: added# noqa: F401to preserve deliberate re-export ofget_gear_listfor test compatibilityTest plan
create_atp_planwith a race date 8–24 weeks out at varying CTL levels (high/mid/low ratio) — verify correct phase selection and NOTE events on calendarget_atp_planafter creating a plan — verify all phase notes are returnedget_atp_week_notefor a date inside each phase — verify correct note is returnedget_planning_context— verify CTL/ATL/TSB, ATP note, workouts and races all appearadd_race_eventwith priority A, B and C — verify correct category saved by APIruff check src/,mypy src tests,pytestall pass ✅ (verified before each commit)🤖 Generated with Claude Code