Initial commit: Modularized Strava MCP Server with UV and Hatchling

2026-05-09 01:06:04 +02:00
commit ed43e1928e
21 changed files with 2688 additions and 0 deletions
@@ -0,0 +1,204 @@
+import httpx
+import os
+import time
+from typing import Optional, Dict, Any, List
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class StravaClient:
+    def __init__(self):
+        self.client_id = os.getenv("STRAVA_CLIENT_ID")
+        self.client_secret = os.getenv("STRAVA_CLIENT_SECRET")
+        self.refresh_token = os.getenv("STRAVA_REFRESH_TOKEN")
+        self.access_token: Optional[str] = None
+        self.expires_at: int = 0
+        self.base_url = "https://www.strava.com/api/v3"
+
+    async def _refresh_access_token(self):
+        """Refreshes the access token using the refresh token."""
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                "https://www.strava.com/oauth/token",
+                data={
+                    "client_id": self.client_id,
+                    "client_secret": self.client_secret,
+                    "grant_type": "refresh_token",
+                    "refresh_token": self.refresh_token,
+                },
+            )
+            response.raise_for_status()
+            data = response.json()
+            self.access_token = data["access_token"]
+            self.refresh_token = data["refresh_token"]
+            self.expires_at = data["expires_at"]
+
+    async def get_valid_token(self) -> str:
+        """Returns a valid access token, refreshing it if necessary."""
+        if not self.access_token or time.time() > self.expires_at - 60:
+            await self._refresh_access_token()
+        return self.access_token
+
+    async def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Any:
+        """Helper for authenticated GET requests to the Strava API."""
+        token = await self.get_valid_token()
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{self.base_url}/{endpoint}",
+                headers={"Authorization": f"Bearer {token}"},
+                params=params,
+            )
+            response.raise_for_status()
+            return response.json()
+
+    # ── Athletes ──────────────────────────────────────────────────────────────
+
+    async def get_athlete(self) -> Dict[str, Any]:
+        """Gets the current athlete profile."""
+        return await self._get("athlete")
+
+    async def get_athlete_zones(self) -> Dict[str, Any]:
+        """Gets heart rate and power zones configured for the current athlete."""
+        return await self._get("athlete/zones")
+
+    async def get_athlete_stats(self) -> Dict[str, Any]:
+        """Gets cumulative stats for the current athlete (totals for runs, rides, swims)."""
+        athlete = await self.get_athlete()
+        return await self._get(f"athletes/{athlete['id']}/stats")
+
+    # ── Activities ────────────────────────────────────────────────────────────
+
+    async def list_activities(
+        self,
+        limit: int = 10,
+        page: int = 1,
+        before: int | None = None,
+        after: int | None = None,
+    ) -> List[Dict[str, Any]]:
+        """Lists activities for the current athlete."""
+        params: Dict[str, Any] = {"per_page": limit, "page": page}
+        if before is not None:
+            params["before"] = before
+        if after is not None:
+            params["after"] = after
+        return await self._get("athlete/activities", params=params)
+
+    async def get_activity(self, activity_id: int) -> Dict[str, Any]:
+        """Gets a specific activity by ID."""
+        return await self._get(f"activities/{activity_id}")
+
+    async def get_activity_laps(self, activity_id: int) -> List[Dict[str, Any]]:
+        """Gets the laps for a specific activity."""
+        return await self._get(f"activities/{activity_id}/laps")
+
+    async def get_activity_zones(self, activity_id: int) -> List[Dict[str, Any]]:
+        """Gets the heart rate and power zones for a specific activity."""
+        return await self._get(f"activities/{activity_id}/zones")
+
+    async def get_activity_comments(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets comments on a specific activity."""
+        return await self._get(f"activities/{activity_id}/comments", params={"page": page, "per_page": per_page})
+
+    async def get_activity_kudoers(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets athletes who kudoed a specific activity."""
+        return await self._get(f"activities/{activity_id}/kudos", params={"page": page, "per_page": per_page})
+
+    async def get_activity_streams(self, activity_id: int, keys: List[str]) -> Dict[str, Any]:
+        """Gets data streams for a specific activity."""
+        return await self._get(
+            f"activities/{activity_id}/streams",
+            params={"keys": ",".join(keys), "key_by_type": True},
+        )
+
+    # ── Clubs ─────────────────────────────────────────────────────────────────
+
+    async def get_athlete_clubs(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets clubs the current athlete belongs to."""
+        return await self._get("athlete/clubs", params={"page": page, "per_page": per_page})
+
+    async def get_club(self, club_id: int) -> Dict[str, Any]:
+        """Gets a specific club by ID."""
+        return await self._get(f"clubs/{club_id}")
+
+    async def get_club_activities(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets recent activities from a club."""
+        return await self._get(f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page})
+
+    async def get_club_members(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets members of a club."""
+        return await self._get(f"clubs/{club_id}/members", params={"page": page, "per_page": per_page})
+
+    # ── Routes ────────────────────────────────────────────────────────────────
+
+    async def get_routes_by_athlete(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets routes created by the current athlete."""
+        athlete = await self.get_athlete()
+        return await self._get(f"athletes/{athlete['id']}/routes", params={"page": page, "per_page": per_page})
+
+    async def get_route_by_id(self, route_id: str) -> Dict[str, Any]:
+        """Gets a specific route by its ID."""
+        return await self._get(f"routes/{route_id}")
+
+    async def get_route_streams(self, route_id: str) -> Dict[str, Any]:
+        """Gets data streams for a specific route."""
+        return await self._get(f"routes/{route_id}/streams")
+
+    # ── Segments ──────────────────────────────────────────────────────────────
+
+    async def get_segment(self, segment_id: int) -> Dict[str, Any]:
+        """Gets a specific segment by ID."""
+        return await self._get(f"segments/{segment_id}")
+
+    async def get_starred_segments(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
+        """Gets segments starred by the current athlete."""
+        return await self._get("segments/starred", params={"page": page, "per_page": per_page})
+
+    async def explore_segments(self, bounds: str, activity_type: str = "riding") -> Dict[str, Any]:
+        """Explores segments within a given bounding box."""
+        return await self._get(
+            "segments/explore",
+            params={"bounds": bounds, "activity_type": activity_type},
+        )
+
+
+    async def get_segment_streams(self, segment_id: int, keys: List[str]) -> Dict[str, Any]:
+        """Gets data streams for a specific segment."""
+        return await self._get(
+            f"segments/{segment_id}/streams",
+            params={"keys": ",".join(keys), "key_by_type": True},
+        )
+
+    # ── Segment Efforts ───────────────────────────────────────────────────────
+
+    async def get_segment_effort(self, effort_id: str) -> Dict[str, Any]:
+        """Gets a specific segment effort by ID."""
+        return await self._get(f"segment_efforts/{effort_id}")
+
+    async def list_segment_efforts(
+        self,
+        segment_id: int,
+        start_date_local: Optional[str] = None,
+        end_date_local: Optional[str] = None,
+        per_page: int = 30,
+    ) -> List[Dict[str, Any]]:
+        """Lists efforts on a segment for the current athlete."""
+        params: Dict[str, Any] = {"segment_id": segment_id, "per_page": per_page}
+        if start_date_local:
+            params["start_date_local"] = start_date_local
+        if end_date_local:
+            params["end_date_local"] = end_date_local
+        return await self._get("segment_efforts", params=params)
+
+    async def get_segment_effort_streams(self, effort_id: str, keys: List[str]) -> Dict[str, Any]:
+        """Gets data streams for a specific segment effort."""
+        return await self._get(
+            f"segment_efforts/{effort_id}/streams",
+            params={"keys": ",".join(keys), "key_by_type": True},
+        )
+
+    # ── Gear ──────────────────────────────────────────────────────────────────
+
+    async def get_gear_by_id(self, gear_id: str) -> Dict[str, Any]:
+        """Gets details for a specific piece of gear (bike or shoes) by its ID."""
+        return await self._get(f"gear/{gear_id}")