feat: coaching-context tools (0.3.0) #6
@@ -6,6 +6,7 @@ Read tools exposing the athlete's profile, per-sport training settings
|
||||
coach needs to reason about intensity and readiness.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from mcp.server.fastmcp import Context # pylint: disable=import-error
|
||||
@@ -24,6 +25,8 @@ 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
|
||||
|
||||
logger = logging.getLogger("intervals_icu_mcp_server")
|
||||
|
||||
|
||||
class _ConfirmThresholdChange(BaseModel):
|
||||
"""Elicitation schema: the user confirms (or not) a threshold change."""
|
||||
@@ -96,14 +99,15 @@ async def get_sport_settings(sport: str | None = None) -> str:
|
||||
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.
|
||||
|
||||
Note: the underlying endpoint's ``tags`` parameter filters *athletes* (a
|
||||
coach-facing feature), not activities, so no tag filter is offered here.
|
||||
|
||||
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()
|
||||
@@ -112,10 +116,6 @@ async def get_athlete_summary(
|
||||
|
||||
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
|
||||
@@ -213,19 +213,28 @@ async def update_sport_settings( # pylint: disable=too-many-arguments,too-many-
|
||||
sport = ", ".join(str(t) for t in (current.get("types") or [])) or f"settings {settings_id}"
|
||||
diff = "\n".join(diff_lines)
|
||||
|
||||
# Guardrail. If the client answers an elicitation prompt, that answer is
|
||||
# authoritative: anything short of accept-with-confirm is a refusal and we
|
||||
# stop WITHOUT emitting the confirm=true fallback instructions (an agentic
|
||||
# client could otherwise use them to bypass the refusal it just received).
|
||||
# Only when elicitation is unavailable (no ctx, or the request itself fails)
|
||||
# do we fall back to requiring the explicit confirm flag.
|
||||
approved = False
|
||||
elicitation_answered = False
|
||||
if ctx is not None:
|
||||
try:
|
||||
elicited = await ctx.elicit(
|
||||
message=f"Update {sport} thresholds?\n{diff}", schema=_ConfirmThresholdChange
|
||||
)
|
||||
elicitation_answered = True
|
||||
action = getattr(elicited, "action", None)
|
||||
if action in ("decline", "cancel"):
|
||||
return "Sport settings unchanged — you declined."
|
||||
data = getattr(elicited, "data", None)
|
||||
approved = action == "accept" and bool(getattr(data, "confirm", False))
|
||||
except Exception: # noqa: BLE001 - client without elicitation capability falls through
|
||||
approved = False
|
||||
except Exception as exc: # noqa: BLE001 - capability absent or elicitation failed
|
||||
logger.warning("Elicitation unavailable, falling back to confirm flag: %s", exc)
|
||||
|
||||
if elicitation_answered and not approved:
|
||||
return "Sport settings unchanged — you did not confirm the change."
|
||||
|
||||
if not approved and not confirm:
|
||||
return (
|
||||
@@ -246,5 +255,6 @@ async def update_sport_settings( # pylint: disable=too-many-arguments,too-many-
|
||||
if isinstance(result, dict) and "error" in result:
|
||||
return f"Error updating sport settings: {result.get('message')}"
|
||||
|
||||
body = result if isinstance(result, dict) else updated
|
||||
# An empty-body 200 parses to {}; render the merged record in that case.
|
||||
body = result if isinstance(result, dict) and result else updated
|
||||
return f"Updated {sport} settings:\n\n" + format_sport_settings(body)
|
||||
|
||||
+27
-9
@@ -187,19 +187,12 @@ def test_get_athlete_summary_success(monkeypatch):
|
||||
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())
|
||||
@@ -266,7 +259,8 @@ def test_update_sport_settings_elicit_decline(monkeypatch):
|
||||
calls = _patch_seq(monkeypatch, [CURRENT_SS])
|
||||
ctx = _StubCtx(action="decline")
|
||||
out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx))
|
||||
assert "you declined" in out
|
||||
assert "did not confirm" in out
|
||||
assert "confirm=true" not in out # no bypass instructions after a refusal
|
||||
assert len(calls) == 1 # no PUT
|
||||
|
||||
|
||||
@@ -274,7 +268,21 @@ def test_update_sport_settings_elicit_cancel(monkeypatch):
|
||||
_patch_seq(monkeypatch, [CURRENT_SS])
|
||||
ctx = _StubCtx(action="cancel")
|
||||
out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx))
|
||||
assert "you declined" in out
|
||||
assert "did not confirm" in out
|
||||
|
||||
|
||||
def test_update_sport_settings_accept_without_confirm_refuses_hard(monkeypatch):
|
||||
# Submitting the elicitation without ticking confirm is a refusal: the tool
|
||||
# must stop and must NOT emit the confirm=true bypass instructions — and an
|
||||
# explicit confirm=True param must not override the answered elicitation.
|
||||
calls = _patch_seq(monkeypatch, [CURRENT_SS])
|
||||
ctx = _StubCtx(action="accept", confirm=False)
|
||||
out = asyncio.run(
|
||||
athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True, ctx=ctx)
|
||||
)
|
||||
assert "did not confirm" in out
|
||||
assert "confirm=true" not in out
|
||||
assert len(calls) == 1 # no PUT
|
||||
|
||||
|
||||
def test_update_sport_settings_elicit_unsupported_falls_back(monkeypatch):
|
||||
@@ -307,3 +315,13 @@ def test_update_sport_settings_put_error(monkeypatch):
|
||||
_patch_seq(monkeypatch, [CURRENT_SS, {"error": True, "message": "rejected"}])
|
||||
out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True))
|
||||
assert "Error updating sport settings: rejected" in out
|
||||
|
||||
|
||||
def test_update_sport_settings_empty_echo_renders_merged(monkeypatch):
|
||||
# An empty-body 200 parses to {}; the confirmation must render the merged
|
||||
# record (with the new FTP), not format_sport_settings({}).
|
||||
calls = _patch_seq(monkeypatch, [CURRENT_SS, {}])
|
||||
out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True))
|
||||
assert len(calls) == 2
|
||||
assert "FTP: 300W" in out
|
||||
assert "Settings ID: 100" in out
|
||||
|
||||
Reference in New Issue
Block a user