feat(multi-tenant): resolve per-caller credentials in every tool
build-image / test (push) Successful in 10s
build-image / build (push) Successful in 42s

All 20 tools now drop the athlete_id/api_key parameters and instead resolve the
authenticated caller's stored, enabled credentials via
credentials.resolve_caller_credentials() (get_access_token().subject -> store).

Security: there is no tool parameter a caller can pass to supply a key, so a
disabled/unapproved user cannot bypass the admin-approval gate — each tool
returns a helpful "not approved / set up your credentials" message instead.
Gear resolution now uses the caller's athlete id rather than an env var.

Tests: conftest autouse fixture runs tool tests as an enabled user; a
parametrized test asserts every tool refuses when unauthorized; existing tool
tests updated (no more athlete_id/api_key kwargs). 221 passing at 91.5%.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-04 19:28:27 -04:00
parent 31eb45c3f8
commit f067f9639a
14 changed files with 256 additions and 195 deletions
+4 -1
View File
@@ -32,7 +32,10 @@ class CredentialError(Exception):
async def resolve_caller_credentials() -> tuple[str, str]:
"""Return ``(athlete_id, api_key)`` for the current caller or raise CredentialError."""
token = get_access_token()
try:
token = get_access_token()
except Exception: # noqa: BLE001 - no auth context (stdio/local dev)
token = None
if token is not None and token.subject:
creds = await store.get_active_credentials(token.subject)
if creds is None:
+39 -26
View File
@@ -7,20 +7,19 @@ This module contains tools for retrieving and managing athlete activities.
from datetime import datetime, timedelta
from typing import Any
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.tools.gear import (
resolve_gear_for_activity,
resolve_gear_for_activities,
)
from intervals_mcp_server.utils.formatting import format_activity_message, format_activity_summary, format_intervals
from intervals_mcp_server.utils.validation import resolve_athlete_id, resolve_date_params
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
config = get_config()
def _parse_activities_from_result(result: Any) -> list[dict[str, Any]]:
"""Extract a list of activity dictionaries from the API result."""
@@ -105,28 +104,24 @@ def _format_activities_response(
@mcp.tool()
async def get_activities( # pylint: disable=too-many-arguments,too-many-return-statements,too-many-branches,too-many-positional-arguments
athlete_id: str | None = None,
api_key: str | None = None,
async def get_activities( # pylint: disable=too-many-return-statements,too-many-branches
start_date: str | None = None,
end_date: str | None = None,
limit: int = 10,
include_unnamed: bool = False,
) -> str:
"""Get a list of activities for an athlete from Intervals.icu
"""Get a list of activities for the signed-in athlete from Intervals.icu
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
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)
limit: Maximum number of activities to return (optional, defaults to 10)
include_unnamed: Whether to include unnamed activities (optional, defaults to False)
"""
# Resolve athlete ID and date parameters
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, 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)
@@ -176,13 +171,17 @@ async def get_activities( # pylint: disable=too-many-arguments,too-many-return-
@mcp.tool()
async def get_activity_details(activity_id: str, api_key: str | None = None) -> str:
async def get_activity_details(activity_id: str) -> str:
"""Get detailed information for a specific activity from Intervals.icu
Args:
activity_id: The Intervals.icu activity ID
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Call the Intervals.icu API
result = await make_intervals_request(url=f"/activity/{activity_id}", api_key=api_key)
@@ -199,8 +198,8 @@ async def get_activity_details(activity_id: str, api_key: str | None = None) ->
if not isinstance(activity_data, dict):
return f"Invalid activity format for activity {activity_id}."
# Resolve gear name (uses configured athlete_id via ATHLETE_ID env var)
await resolve_gear_for_activity(activity_data, api_key=api_key)
# Resolve gear name for the signed-in athlete
await resolve_gear_for_activity(activity_data, athlete_id=athlete_id_to_use, api_key=api_key)
# Return a more detailed view of the activity
detailed_view = format_activity_summary(activity_data)
@@ -220,7 +219,7 @@ async def get_activity_details(activity_id: str, api_key: str | None = None) ->
@mcp.tool()
async def get_activity_intervals(activity_id: str, api_key: str | None = None) -> str:
async def get_activity_intervals(activity_id: str) -> str:
"""Get interval data for a specific activity from Intervals.icu
This endpoint returns detailed metrics for each interval in an activity, including power, heart rate,
@@ -228,8 +227,12 @@ async def get_activity_intervals(activity_id: str, api_key: str | None = None) -
Args:
activity_id: The Intervals.icu activity ID
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
try:
_athlete_id, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Call the Intervals.icu API
result = await make_intervals_request(url=f"/activity/{activity_id}/intervals", api_key=api_key)
@@ -254,7 +257,6 @@ async def get_activity_intervals(activity_id: str, api_key: str | None = None) -
@mcp.tool()
async def get_activity_streams(
activity_id: str,
api_key: str | None = None,
stream_types: str | None = None,
) -> str:
"""Get stream data for a specific activity from Intervals.icu
@@ -264,11 +266,15 @@ async def get_activity_streams(
Args:
activity_id: The Intervals.icu activity ID
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
stream_types: Comma-separated list of stream types to retrieve (optional, defaults to all available types)
Available types: time, watts, heartrate, cadence, altitude, distance,
core_temperature, skin_temperature, velocity_smooth
"""
try:
_athlete_id, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Build query parameters
params = {}
if stream_types:
@@ -330,13 +336,17 @@ async def get_activity_streams(
@mcp.tool()
async def get_activity_messages(activity_id: str, api_key: str | None = None) -> str:
async def get_activity_messages(activity_id: str) -> str:
"""Get messages (notes/comments) for a specific activity from Intervals.icu
Args:
activity_id: The Intervals.icu activity ID
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
try:
_athlete_id, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
result = await make_intervals_request(
url=f"/activity/{activity_id}/messages",
api_key=api_key,
@@ -365,15 +375,18 @@ async def get_activity_messages(activity_id: str, api_key: str | None = None) ->
async def add_activity_message(
activity_id: str,
content: str,
api_key: str | None = None,
) -> str:
"""Add a message (note/comment) to an activity on Intervals.icu
Args:
activity_id: The Intervals.icu activity ID
content: The message text to add
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
try:
_athlete_id, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
result = await make_intervals_request(
url=f"/activity/{activity_id}/messages",
api_key=api_key,
+22 -39
View File
@@ -7,31 +7,26 @@ This module contains tools for managing athlete custom items (charts, fields, zo
import json
from typing import Any
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.utils.formatting import format_custom_item_details
from intervals_mcp_server.utils.validation import resolve_athlete_id
# Import mcp instance from shared module for tool registration
from intervals_mcp_server.mcp_instance import mcp # noqa: F401
config = get_config()
@mcp.tool()
async def get_custom_items(
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Get custom items (charts, custom fields, zones, etc.) for an athlete from Intervals.icu
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
result = await make_intervals_request(
url=f"/athlete/{athlete_id_to_use}/custom-item", api_key=api_key
@@ -58,19 +53,16 @@ async def get_custom_items(
@mcp.tool()
async def get_custom_item_by_id(
item_id: int,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Get detailed information for a specific custom item from Intervals.icu
Args:
item_id: The custom item ID
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
result = await make_intervals_request(
url=f"/athlete/{athlete_id_to_use}/custom-item/{item_id}", api_key=api_key
@@ -89,8 +81,6 @@ async def get_custom_item_by_id(
async def create_custom_item(
name: str,
item_type: str,
athlete_id: str | None = None,
api_key: str | None = None,
description: str | None = None,
content: dict[str, Any] | None = None,
visibility: str | None = None,
@@ -100,17 +90,16 @@ async def create_custom_item(
Args:
name: Name of the custom item
item_type: Type of custom item (e.g. FITNESS_CHART, TRACE_CHART, INPUT_FIELD, ACTIVITY_FIELD, INTERVAL_FIELD, ACTIVITY_STREAM, ACTIVITY_CHART, ACTIVITY_HISTOGRAM, ACTIVITY_HEATMAP, ACTIVITY_MAP, ACTIVITY_PANEL, ZONES)
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
description: Description of the custom item (optional)
content: Configuration content for the custom item as a dict (optional). Important enum values:
- "type" field for INPUT_FIELD/ACTIVITY_FIELD: must be "numeric", "text", or "select" (NOT "number")
- "aggregate" field: must be "MIN", "SUM", "MAX", or "AVERAGE" (NOT "AVG")
visibility: Visibility setting: PRIVATE, FOLLOWERS, or PUBLIC (optional)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
data: dict[str, Any] = {"name": name, "type": item_type}
if description is not None:
@@ -144,8 +133,6 @@ async def create_custom_item(
@mcp.tool()
async def update_custom_item(
item_id: int,
athlete_id: str | None = None,
api_key: str | None = None,
name: str | None = None,
item_type: str | None = None,
description: str | None = None,
@@ -156,8 +143,6 @@ async def update_custom_item(
Args:
item_id: The custom item ID to update
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
name: New name for the custom item (optional)
item_type: New type for the custom item (optional)
description: New description for the custom item (optional)
@@ -166,9 +151,10 @@ async def update_custom_item(
- "aggregate" field: must be "MIN", "SUM", "MAX", or "AVERAGE" (NOT "AVG")
visibility: New visibility setting: PRIVATE, FOLLOWERS, or PUBLIC (optional)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
data: dict[str, Any] = {}
if name is not None:
@@ -206,19 +192,16 @@ async def update_custom_item(
@mcp.tool()
async def delete_custom_item(
item_id: int,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Delete a custom item for an athlete from Intervals.icu
Args:
item_id: The custom item ID to delete
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
result = await make_intervals_request(
url=f"/athlete/{athlete_id_to_use}/custom-item/{item_id}",
+27 -44
View File
@@ -8,18 +8,17 @@ import json
from datetime import datetime
from typing import Any
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.utils.dates import get_default_end_date, get_default_future_end_date
from intervals_mcp_server.utils.formatting import format_event_details, format_event_summary
from intervals_mcp_server.utils.types import WorkoutDoc
from intervals_mcp_server.utils.validation import resolve_activity_type, resolve_athlete_id, validate_date
from intervals_mcp_server.utils.validation import resolve_activity_type, validate_date
# Import mcp instance from shared module for tool registration
from intervals_mcp_server.mcp_instance import mcp # noqa: F401
config = get_config()
def _prepare_event_data( # pylint: disable=too-many-arguments,too-many-positional-arguments
name: str,
@@ -89,23 +88,20 @@ async def _delete_events_list(
@mcp.tool()
async def get_events(
athlete_id: str | None = None,
api_key: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
) -> str:
"""Get events for an athlete from Intervals.icu
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
start_date: Start date in YYYY-MM-DD format (optional, defaults to today)
end_date: End date in YYYY-MM-DD format (optional, defaults to 30 days from today)
"""
# Resolve athlete ID
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Parse date parameters (events use different defaults)
if not start_date:
@@ -147,20 +143,17 @@ async def get_events(
@mcp.tool()
async def get_event_by_id(
event_id: str,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Get detailed information for a specific event from Intervals.icu
Args:
event_id: The Intervals.icu event ID
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
"""
# Resolve athlete ID
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Call the Intervals.icu API
result = await make_intervals_request(
@@ -184,18 +177,15 @@ async def get_event_by_id(
@mcp.tool()
async def delete_event(
event_id: str,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Delete event for an athlete from Intervals.icu
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
event_id: The Intervals.icu event ID
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
if not event_id:
return "Error: No event ID provided."
result = await make_intervals_request(
@@ -234,20 +224,17 @@ async def _fetch_events_for_deletion(
async def delete_events_by_date_range(
start_date: str,
end_date: str,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Delete events for an athlete from Intervals.icu in the specified date range.
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
events, error_msg = await _fetch_events_for_deletion(
athlete_id_to_use, api_key, start_date, end_date
@@ -264,8 +251,6 @@ async def delete_events_by_date_range(
async def add_or_update_event( # pylint: disable=too-many-arguments,too-many-positional-arguments
workout_type: str,
name: str,
athlete_id: str | None = None,
api_key: str | None = None,
event_id: str | None = None,
start_date: str | None = None,
workout_doc: WorkoutDoc | None = None,
@@ -278,8 +263,6 @@ async def add_or_update_event( # pylint: disable=too-many-arguments,too-many-po
Many arguments are required as this MCP tool function maps directly to the Intervals.icu API parameters.
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
event_id: The Intervals.icu event ID (optional, will use event_id from .env if not provided)
start_date: Start date in YYYY-MM-DD format (optional, defaults to today)
name: Name of the activity
@@ -337,9 +320,10 @@ async def add_or_update_event( # pylint: disable=too-many-arguments,too-many-po
- Use "reps" with nested steps to define repeat intervals (as in example above)
- Define one of "power", "hr" or "pace" to define step intensity
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
if not start_date:
start_date = datetime.now().strftime("%Y-%m-%d")
@@ -362,8 +346,6 @@ async def add_or_update_note(
description: str,
start_date: str | None = None,
color: str | None = "green",
athlete_id: str | None = None,
api_key: str | None = None,
event_id: str | None = None,
) -> str:
"""Add or update a plain text note (category NOTE) on the Intervals.icu calendar.
@@ -377,9 +359,10 @@ async def add_or_update_note(
api_key: The Intervals.icu API key (optional)
event_id: The Intervals.icu event ID (optional, for updates)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
if not start_date:
start_date = datetime.now().strftime("%Y-%m-%d")
+7 -10
View File
@@ -19,8 +19,10 @@ Call `get_gear_list(refresh=True)` to bust the cache.
from typing import Any
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.utils.validation import resolve_athlete_id
# Import mcp instance from shared module for tool registration
@@ -151,26 +153,21 @@ async def resolve_gear_for_activities(
@mcp.tool()
async def get_gear_list(
athlete_id: str | None = None,
api_key: str | None = None,
refresh: bool = False,
) -> str:
"""Get the gear catalog (bikes, shoes, etc.) for an athlete from Intervals.icu.
"""Get the gear catalog (bikes, shoes, etc.) for the signed-in athlete from Intervals.icu.
Returns one line per gear item with id, type, name, and basic stats.
The result is cached for the MCP process lifetime; pass refresh=True to
re-fetch.
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
refresh: If True, bypass the cache and re-fetch from the API (default False)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
if not athlete_id_to_use:
return "Error: athlete_id is required (either as argument or via ATHLETE_ID env var)."
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
# Single fetch path: get_gear_raw consults the cache and only hits the API
# on a cold cache or when refresh=True.
+6 -11
View File
@@ -8,16 +8,14 @@ import json
from datetime import datetime
from typing import Any
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.utils.formatting import format_power_curves
from intervals_mcp_server.utils.validation import resolve_activity_type, resolve_athlete_id
# Import mcp instance from shared module for tool registration
from intervals_mcp_server.mcp_instance import mcp # noqa: F401
config = get_config()
# 5s, 15s, 30s, 1min, 2min, 5min, 10min, 20min, 60min
DEFAULT_DURATIONS: tuple[int, ...] = (5, 15, 30, 60, 120, 300, 600, 1200, 3600)
@@ -135,8 +133,6 @@ async def get_athlete_power_curves(
this_season: bool = True,
last_season: bool = True,
include_normalised: bool = True,
athlete_id: str | None = None,
api_key: str | None = None,
) -> str:
"""Get power curves for an athlete from Intervals.icu.
@@ -152,15 +148,14 @@ async def get_athlete_power_curves(
this_season: Include this season's curve (default True)
last_season: Include last season's curve (default True)
include_normalised: Include weight-normalised W/kg values (default True)
athlete_id: Intervals.icu athlete ID (optional, uses ATHLETE_ID from .env if not provided)
api_key: Optional API key override. Uses API_KEY from .env if not provided.
"""
if durations is None:
durations = list(DEFAULT_DURATIONS)
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, api_key = await credentials.resolve_caller_credentials()
except CredentialError as exc:
return str(exc)
if indoor_outdoor and indoor_outdoor not in ("indoor", "outdoor"):
return "Error: indoor_outdoor must be 'indoor', 'outdoor', or omitted."
+8 -12
View File
@@ -4,41 +4,37 @@ Wellness-related MCP tools for Intervals.icu.
This module contains tools for retrieving athlete wellness data.
"""
from intervals_mcp_server import credentials
from intervals_mcp_server.api.client import make_intervals_request
from intervals_mcp_server.config import get_config
from intervals_mcp_server.credentials import CredentialError
from intervals_mcp_server.utils.formatting import format_wellness_entry
from intervals_mcp_server.utils.validation import resolve_athlete_id, resolve_date_params
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
config = get_config()
@mcp.tool()
async def get_wellness_data(
athlete_id: str | None = None,
api_key: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
include_all_fields: bool = False,
) -> str:
"""Get wellness data for an athlete from Intervals.icu.
"""Get wellness data for the signed-in athlete from Intervals.icu.
By default returns standard wellness fields (training metrics, vitals, sleep,
subjective scores, etc.). Set include_all_fields=True to also include any
additional or custom fields configured by the user in Intervals.icu.
Args:
athlete_id: The Intervals.icu athlete ID (optional, will use ATHLETE_ID from .env if not provided)
api_key: The Intervals.icu API key (optional, will use API_KEY from .env if not provided)
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)
include_all_fields: If True, include additional and custom fields beyond the standard set (optional, defaults to False)
"""
athlete_id_to_use, error_msg = resolve_athlete_id(athlete_id, config.athlete_id)
if error_msg:
return error_msg
try:
athlete_id_to_use, 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)