Strava MCP Server

Instructions

Implementation Reference

• src/strava_mcp_server/server.py:292-319 (handler)

  The main handler function for the 'get_recent_activities' tool. It is decorated with @mcp.tool(), which also handles registration in the FastMCP server. Fetches activities from Strava API for the past specified days using the StravaClient.
  @mcp.tool() def get_recent_activities(days: int = 7, limit: int = 10) -> dict[str, Any]: """ Get activities from the past X days. Args: days: Number of days to look back (default: 7) limit: Maximum number of activities to return (default: 10) Returns: Dictionary containing activities data """ if strava_client is None: return { "error": "Strava client not initialized. Please provide refresh token, client ID, and client secret." # noqa: E501 } try: # Calculate timestamp for X days ago now = datetime.now() days_ago = now - timedelta(days=days) after = int(days_ago.timestamp()) activities = strava_client.get_activities(limit=limit, after=after) return {"data": activities} except Exception as e: return {"error": str(e)}
• src/strava_mcp_server/server.py:82-106 (helper)

  Helper method in StravaClient class used by get_recent_activities to fetch filtered activities from Strava API.
  def get_activities( self, limit: int = 10, before: Optional[int] = None, after: Optional[int] = None ) -> list: """ Get the authenticated athlete's activities. Args: limit: Maximum number of activities to return before: Unix timestamp to filter activities before this time after: Unix timestamp to filter activities after this time Returns: List of activities """ params = {"per_page": limit} if before: params["before"] = before if after: params["after"] = after activities = self._make_request("athlete/activities", params) return self._filter_activities(activities)
• src/strava_mcp_server/server.py:19-155 (helper)

  The StravaClient class provides the core API interaction logic, including token management, requests, and activity fetching, used by the tool handler.
  class StravaClient: """Client for interacting with the Strava API.""" BASE_URL = "https://www.strava.com/api/v3" def __init__(self, refresh_token: str, client_id: str, client_secret: str): """ Initialize the Strava API client. Args: refresh_token: Refresh token for Strava API client_id: Client ID for Strava API client_secret: Client secret for Strava API """ self.refresh_token = refresh_token self.client_id = client_id self.client_secret = client_secret self.access_token = None self.expires_at = 0 self.client = httpx.Client(timeout=30.0) def _ensure_valid_token(self) -> None: """Ensure we have a valid access token, refreshing if necessary.""" current_time = int(time.time()) # If token is missing or expired, refresh it if not self.access_token or current_time >= self.expires_at: self._refresh_token() def _refresh_token(self) -> None: """Refresh the access token using the refresh token.""" refresh_url = "https://www.strava.com/oauth/token" payload = { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "grant_type": "refresh_token", } response = self.client.post(refresh_url, data=payload) if response.status_code != 200: error_msg = f"Error {response.status_code}: {response.text}" raise Exception(error_msg) token_data = response.json() self.access_token = token_data["access_token"] self.expires_at = token_data["expires_at"] print("Token refreshed successfully") def _make_request(self, endpoint: str, params: Optional[dict] = None) -> Any: """Make an authenticated request to the Strava API.""" self._ensure_valid_token() url = f"{self.BASE_URL}/{endpoint}" headers = {"Authorization": f"Bearer {self.access_token}"} response = self.client.get(url, headers=headers, params=params) if response.status_code != 200: error_msg = f"Error {response.status_code}: {response.text}" raise Exception(error_msg) return response.json() def get_activities( self, limit: int = 10, before: Optional[int] = None, after: Optional[int] = None ) -> list: """ Get the authenticated athlete's activities. Args: limit: Maximum number of activities to return before: Unix timestamp to filter activities before this time after: Unix timestamp to filter activities after this time Returns: List of activities """ params = {"per_page": limit} if before: params["before"] = before if after: params["after"] = after activities = self._make_request("athlete/activities", params) return self._filter_activities(activities) def get_activity(self, activity_id: int) -> dict: """ Get detailed information about a specific activity. Args: activity_id: ID of the activity to retrieve Returns: Activity details """ activity = self._make_request(f"activities/{activity_id}") return self._filter_activity(activity) def _filter_activity(self, activity: dict) -> dict: """Filter activity to only include specific keys and rename with units.""" # Define field mappings with units field_mappings = { "calories": "calories", "distance": "distance_metres", "elapsed_time": "elapsed_time_seconds", "elev_high": "elev_high_metres", "elev_low": "elev_low_metres", "end_latlng": "end_latlng", "average_speed": "average_speed_mps", # metres per second "max_speed": "max_speed_mps", # metres per second "moving_time": "moving_time_seconds", "sport_type": "sport_type", "start_date": "start_date", "start_latlng": "start_latlng", "total_elevation_gain": "total_elevation_gain_metres", "name": "name", # Keep name for display purposes } # Create a new dictionary with renamed fields filtered_activity = {} for old_key, new_key in field_mappings.items(): if old_key in activity: filtered_activity[new_key] = activity[old_key] return filtered_activity def _filter_activities(self, activities: list) -> list: """Filter a list of activities to only include specific keys with units.""" return [self._filter_activity(activity) for activity in activities] def close(self) -> None: """Close the HTTP client.""" self.client.close()