Strava MCP Server

Instructions

Implementation Reference

• server.py:120-139 (handler)

  The MCP tool handler for get_activity_streams_tool. This function is decorated with @mcp.tool() and serves as the entry point that retrieves activity streams by calling get_client() and get_activity_streams(), then returns the result as a dictionary.

  @mcp.tool()
  def get_activity_streams_tool(
      activity_id: int,
      types: Optional[list[str]] = None,
      resolution: Optional[Literal["low", "medium", "high"]] = None,
  ) -> dict:
      """
      Get raw stream data (GPS, HR, power, etc.) for a specific activity.
  
      Args:
          activity_id: The ID of the activity to retrieve streams for
          types: List of stream types to fetch. Options: time, latlng, distance, altitude,
                 velocity_smooth, heartrate, cadence, watts, temp, moving, grade_smooth.
                 If None, all available streams will be returned.
          resolution: Data point resolution - 'low' (100 points), 'medium' (1000 points),
                     'high' (10000 points), or None (all points)
      """
      client = get_client()
      streams = get_activity_streams(client, activity_id, types, resolution)
      return streams.to_dict()

• strava_mcp/services/streams.py:43-74 (handler)

  Core implementation of get_activity_streams that fetches raw stream data (GPS, HR, power, etc.) from the Strava API. It calls client.get_activity_streams() with optional parameters for types and resolution, then extracts data into an ActivityStreams model.

  def get_activity_streams(
      client: Client,
      activity_id: int,
      types: Optional[list[str]] = None,
      resolution: Optional[Literal["low", "medium", "high"]] = None,
  ) -> ActivityStreams:
      """Get raw stream data (GPS, HR, power, etc.) for a specific activity."""
      streams = client.get_activity_streams(
          activity_id, types=types, resolution=resolution
      )
  
      # Extract data from each stream type if available
      # Each stream is a Stream object with a .data attribute
      def get_stream_data(key: str) -> Optional[list]:
          if key in streams:
              stream = streams[key]
              return getattr(stream, "data", None)
          return None
  
      return ActivityStreams(
          time=get_stream_data("time"),
          latlng=get_stream_data("latlng"),
          distance=get_stream_data("distance"),
          altitude=get_stream_data("altitude"),
          velocity_smooth=get_stream_data("velocity_smooth"),
          heartrate=get_stream_data("heartrate"),
          cadence=get_stream_data("cadence"),
          watts=get_stream_data("watts"),
          temp=get_stream_data("temp"),
          moving=get_stream_data("moving"),
          grade_smooth=get_stream_data("grade_smooth"),
      )

• strava_mcp/models.py:106-124 (schema)

  ActivityStreams dataclass that defines the schema for activity stream data. Includes optional fields for time, latlng, distance, altitude, velocity_smooth, heartrate, cadence, watts, temp, moving, and grade_smooth, with a to_dict() method for serialization.

  @dataclass
  class ActivityStreams:
      """Raw stream data for a Strava activity."""
  
      time: Optional[list[int]] = None
      latlng: Optional[list[list[float]]] = None
      distance: Optional[list[float]] = None
      altitude: Optional[list[float]] = None
      velocity_smooth: Optional[list[float]] = None
      heartrate: Optional[list[int]] = None
      cadence: Optional[list[int]] = None
      watts: Optional[list[int]] = None
      temp: Optional[list[int]] = None
      moving: Optional[list[bool]] = None
      grade_smooth: Optional[list[float]] = None
  
      def to_dict(self) -> dict:
          """Convert to dictionary for serialization, excluding None values."""
          return {k: v for k, v in asdict(self).items() if v is not None}

• server.py:17-17 (registration)

  Import statement that brings get_activity_streams into scope from strava_mcp.services.streams, enabling its use in the tool handler.

  from strava_mcp.services.streams import get_activity_laps, get_activity_streams

• strava_mcp/auth.py:13-41 (helper)

  Helper function get_client() that returns an authenticated Strava client with automatic token refresh. Used by get_activity_streams_tool to obtain a valid API client before fetching streams.

  def get_client() -> Client:
      """
      Returns an authenticated Strava client, refreshing the access token if necessary.
      """
      global _client, _token_expires_at
  
      # Buffer time to refresh before actual expiration (e.g., 5 minutes)
      if time.time() > _token_expires_at - 300:
          try:
              sys.stderr.write("Refreshing Strava access token...\n")
              # Ensure CLIENT_ID is correctly typed if needed, though stravalib handles str often
              response = _client.refresh_access_token(
                  client_id=int(CLIENT_ID)
                  if CLIENT_ID and CLIENT_ID.isdigit()
                  else CLIENT_ID,  # type: ignore
                  client_secret=CLIENT_SECRET,  # type: ignore
                  refresh_token=REFRESH_TOKEN,  # type: ignore
              )
              _client.access_token = response["access_token"]
              _client.refresh_token = response["refresh_token"]
              _token_expires_at = response["expires_at"]
              sys.stderr.write("Token refreshed successfully.\n")
          except Exception as e:
              # Log full error to stderr for debugging
              sys.stderr.write(f"Auth Error: {e}\n")
              # Return generic error to client to avoid leaking secrets
              raise RuntimeError("Failed to authenticate with Strava. Check server logs.")
  
      return _client