add strava auth setup

strava_mcp/api.py

@@ -1,4 +1,5 @@
 import logging
+import os
 from datetime import datetime
 
 import httpx
@@ -71,27 +72,39 @@ class StravaAPI:
         if self.access_token and self.token_expires_at and now < self.token_expires_at:
             return self.access_token
 
-        # If we don't have a refresh token, try to get one through standalone OAuth flow
+        # If we don't have a refresh token, check if we're in a Hugging Face Space environment
         if not self.settings.refresh_token:
-            logger.warning("No refresh token available, launching standalone OAuth server")
-            try:
-                # Import here to avoid circular import
-                from strava_mcp.oauth_server import get_refresh_token_from_oauth
-
-                logger.info("Starting OAuth flow to get refresh token")
-                self.settings.refresh_token = await get_refresh_token_from_oauth(
-                    self.settings.client_id, self.settings.client_secret
-                )
-                logger.info("Successfully obtained refresh token from OAuth flow")
-            except Exception as e:
-                error_msg = f"Failed to get refresh token through OAuth flow: {e}"
-                logger.error(error_msg)
-
-                # No fallback to MCP-integrated auth flow anymore
+            # Check if we're running in Hugging Face Spaces
+            space_id = os.environ.get("SPACE_ID")
+            if space_id:
+                # We're in Hugging Face Spaces - can't do OAuth flow automatically
                 raise Exception(
-                    "No refresh token available and OAuth flow failed. "
-                    "Please set STRAVA_REFRESH_TOKEN manually in your environment variables."
-                ) from e
+                    "No refresh token available. In Hugging Face Spaces, you must provide a STRAVA_REFRESH_TOKEN "
+                    "environment variable or use the Authentication tab to set your refresh token manually. "
+                    "Use the OAuth Helper tab for instructions on how to get a refresh token."
+                )
+            else:
+                # We're running locally - try OAuth flow
+                logger.warning(
+                    "No refresh token available, launching standalone OAuth server"
+                )
+                try:
+                    # Import here to avoid circular import
+                    from strava_mcp.oauth_server import get_refresh_token_from_oauth
+
+                    logger.info("Starting OAuth flow to get refresh token")
+                    self.settings.refresh_token = await get_refresh_token_from_oauth(
+                        self.settings.client_id, self.settings.client_secret
+                    )
+                    logger.info("Successfully obtained refresh token from OAuth flow")
+                except Exception as e:
+                    error_msg = f"Failed to get refresh token through OAuth flow: {e}"
+                    logger.error(error_msg)
+
+                    raise Exception(
+                        "No refresh token available and OAuth flow failed. "
+                        "Please set STRAVA_REFRESH_TOKEN manually in your environment variables."
+                    ) from e
 
         # Now that we have a refresh token, refresh the access token
         async with httpx.AsyncClient() as client:
@@ -144,15 +157,30 @@ class StravaAPI:
         response = await self._client.request(method, url, headers=headers, **kwargs)
 
         if not response.is_success:
-            error_msg = f"Strava API request failed: {response.status_code} - {response.text}"
+            error_msg = (
+                f"Strava API request failed: {response.status_code} - {response.text}"
+            )
             logger.error(error_msg)
 
             try:
                 error_data = response.json()
                 error = ErrorResponse(**error_data)
-                raise Exception(f"Strava API error: {error.message} (code: {error.code})")
+                if error.errors:
+                    # Extract more specific error information
+                    error_details = []
+                    for err in error.errors:
+                        field = err.get("field", "unknown")
+                        code = err.get("code", "unknown")
+                        error_details.append(f"{field}: {code}")
+                    details = ", ".join(error_details)
+                    raise Exception(f"Strava API error: {error.message} ({details})")
+                else:
+                    raise Exception(f"Strava API error: {error.message}")
             except Exception as err:
-                msg = f"Strava API failed: {response.status_code} - {response.text[:50]}"
+                # If we can't parse the error response, fall back to the raw response
+                msg = (
+                    f"Strava API failed: {response.status_code} - {response.text[:200]}"
+                )
                 raise Exception(msg) from err
 
         return response
@@ -186,7 +214,9 @@ class StravaAPI:
 
         return [Activity(**activity) for activity in data]
 
-    async def get_activity(self, activity_id: int, include_all_efforts: bool = False) -> DetailedActivity:
+    async def get_activity(
+        self, activity_id: int, include_all_efforts: bool = False
+    ) -> DetailedActivity:
         """Get a specific activity.
 
         Args:
@@ -200,7 +230,9 @@ class StravaAPI:
         if include_all_efforts:
             params["include_all_efforts"] = "true"
 
-        response = await self._request("GET", f"/activities/{activity_id}", params=params)
+        response = await self._request(
+            "GET", f"/activities/{activity_id}", params=params
+        )
         data = response.json()
 
         return DetailedActivity(**data)

strava_mcp/gradio_server.py

@@ -17,17 +17,18 @@ logger = logging.getLogger(__name__)
 service: Optional[StravaService] = None
 
 
-async def initialize_service():
+async def initialize_service(refresh_token: str = None):
     """Initialize the Strava service with settings from environment variables."""
     global service
 
-    if service is not None:
+    if service is not None and refresh_token is None:
         return service
 
     try:
         settings = StravaSettings(
             client_id=os.getenv("STRAVA_CLIENT_ID", ""),
             client_secret=os.getenv("STRAVA_CLIENT_SECRET", ""),
+            refresh_token=refresh_token or os.getenv("STRAVA_REFRESH_TOKEN", ""),
             base_url="https://www.strava.com/api/v3",
         )
 
@@ -35,6 +36,10 @@ async def initialize_service():
             raise ValueError("STRAVA_CLIENT_ID environment variable is not set")
         if not settings.client_secret:
             raise ValueError("STRAVA_CLIENT_SECRET environment variable is not set")
+        if not settings.refresh_token:
+            raise ValueError(
+                "STRAVA_REFRESH_TOKEN is required for Hugging Face Spaces deployment"
+            )
 
         # Initialize service without FastAPI app for Gradio
         service = StravaService(settings, None)
@@ -46,6 +51,69 @@ async def initialize_service():
         raise
 
 
+async def setup_authentication(refresh_token: str) -> str:
+    """Set up authentication with the provided refresh token.
+
+    Args:
+        refresh_token: The Strava refresh token
+
+    Returns:
+        Status message
+    """
+    try:
+        if not refresh_token.strip():
+            raise ValueError("Refresh token cannot be empty")
+
+        global service
+        service = None  # Reset service to force re-initialization
+        await initialize_service(refresh_token.strip())
+        return "✅ Authentication successful! You can now use the Strava API functions."
+    except Exception as e:
+        logger.error(f"Authentication failed: {str(e)}")
+        return f"❌ Authentication failed: {str(e)}"
+
+
+def get_authorization_url() -> str:
+    """Get the Strava authorization URL for manual OAuth flow.
+
+    Returns:
+        The authorization URL and instructions
+    """
+    client_id = os.getenv("STRAVA_CLIENT_ID", "")
+    if not client_id:
+        return "❌ STRAVA_CLIENT_ID environment variable is not set"
+
+    # For Hugging Face Spaces, we need to provide a redirect URI that points to a manual flow
+    redirect_uri = "http://localhost:3008/exchange_token"  # This is just for display
+    auth_url = f"https://www.strava.com/oauth/authorize?client_id={client_id}&redirect_uri={redirect_uri}&response_type=code&scope=activity:read_all,read_all,profile:read_all&approval_prompt=force"
+
+    instructions = f"""
+🔐 **Manual OAuth Setup Instructions:**
+
+1. **Click this link to authorize with Strava:**
+   {auth_url}
+
+2. **After authorization, you'll be redirected to a page that might show an error (this is expected)**
+
+3. **Copy the 'code' parameter from the URL** (it will look like: `?code=abc123...`)
+
+4. **Exchange the code for a refresh token** using this curl command:
+   ```bash
+   curl -X POST https://www.strava.com/oauth/token \\
+     -d client_id={client_id} \\
+     -d client_secret=YOUR_CLIENT_SECRET \\
+     -d code=THE_CODE_FROM_STEP_3 \\
+     -d grant_type=authorization_code
+   ```
+
+5. **Copy the 'refresh_token' from the response** and paste it in the "Refresh Token" field above.
+
+⚠️ **Note:** You'll need your STRAVA_CLIENT_SECRET for step 4. Contact the app administrator if you don't have it.
+"""
+
+    return instructions
+
+
 async def get_user_activities(
     before: Optional[int] = None,
     after: Optional[int] = None,
@@ -124,6 +192,31 @@ async def get_activity_segments(activity_id: int) -> List[Dict]:
 def create_interface():
     """Create the Gradio interface for the Strava MCP server."""
 
+    # Authentication interface
+    auth_interface = gr.Interface(
+        fn=setup_authentication,
+        inputs=[
+            gr.Textbox(
+                label="Refresh Token",
+                placeholder="Enter your Strava refresh token here",
+                type="password",
+                lines=1,
+            )
+        ],
+        outputs=gr.Textbox(label="Status"),
+        title="🔐 Authentication",
+        description="Enter your Strava refresh token to authenticate. If you don't have one, use the OAuth Helper tab to get it.",
+    )
+
+    # OAuth helper interface
+    oauth_interface = gr.Interface(
+        fn=get_authorization_url,
+        inputs=[],
+        outputs=gr.Markdown(label="OAuth Instructions"),
+        title="🔗 OAuth Helper",
+        description="Get instructions for manually obtaining a Strava refresh token",
+    )
+
     # Activities interface
     activities_interface = gr.Interface(
         fn=get_user_activities,
@@ -134,7 +227,7 @@ def create_interface():
             gr.Number(label="Per Page", value=30, precision=0),
         ],
         outputs=gr.JSON(label="Activities"),
-        title="Get User Activities",
+        title="📊 Get User Activities",
         description="Retrieve the authenticated user's activities from Strava",
     )
 
@@ -146,7 +239,7 @@ def create_interface():
             gr.Checkbox(label="Include All Efforts", value=False),
         ],
         outputs=gr.JSON(label="Activity Details"),
-        title="Get Activity Details",
+        title="🔍 Get Activity Details",
         description="Get detailed information about a specific activity",
     )
 
@@ -157,14 +250,26 @@ def create_interface():
             gr.Number(label="Activity ID", precision=0),
         ],
         outputs=gr.JSON(label="Activity Segments"),
-        title="Get Activity Segments",
+        title="🏃 Get Activity Segments",
         description="Get segment efforts for a specific activity",
     )
 
     # Combine interfaces in a tabbed interface
     demo = gr.TabbedInterface(
-        [activities_interface, details_interface, segments_interface],
-        ["Activities", "Activity Details", "Activity Segments"],
+        [
+            auth_interface,
+            oauth_interface,
+            activities_interface,
+            details_interface,
+            segments_interface,
+        ],
+        [
+            "🔐 Authentication",
+            "🔗 OAuth Helper",
+            "📊 Activities",
+            "🔍 Activity Details",
+            "🏃 Activity Segments",
+        ],
         title="Strava MCP Server",
     )
 

strava_mcp/models.py

@@ -11,11 +11,15 @@ class Activity(BaseModel):
     distance: float = Field(..., description="The distance in meters")
     moving_time: int = Field(..., description="Moving time in seconds")
     elapsed_time: int = Field(..., description="Elapsed time in seconds")
-    total_elevation_gain: float = Field(..., description="Total elevation gain in meters")
+    total_elevation_gain: float = Field(
+        ..., description="Total elevation gain in meters"
+    )
     type: str = Field(..., description="Type of activity")
     sport_type: str = Field(..., description="Type of sport")
     start_date: datetime = Field(..., description="Start date and time in UTC")
-    start_date_local: datetime = Field(..., description="Start date and time in athlete's timezone")
+    start_date_local: datetime = Field(
+        ..., description="Start date and time in athlete's timezone"
+    )
     timezone: str = Field(..., description="The timezone of the activity")
     achievement_count: int = Field(..., description="The number of achievements")
     kudos_count: int = Field(..., description="The number of kudos")
@@ -23,7 +27,9 @@ class Activity(BaseModel):
     athlete_count: int = Field(..., description="The number of athletes")
     photo_count: int = Field(..., description="The number of photos")
     map: dict | None = Field(None, description="The map of the activity")
-    trainer: bool = Field(..., description="Whether this activity was recorded on a training machine")
+    trainer: bool = Field(
+        ..., description="Whether this activity was recorded on a training machine"
+    )
     commute: bool = Field(..., description="Whether this activity is a commute")
     manual: bool = Field(..., description="Whether this activity was created manually")
     private: bool = Field(..., description="Whether this activity is private")
@@ -31,9 +37,15 @@ class Activity(BaseModel):
     workout_type: int | None = Field(None, description="The workout type")
     average_speed: float = Field(..., description="Average speed in meters per second")
     max_speed: float = Field(..., description="Maximum speed in meters per second")
-    has_heartrate: bool = Field(..., description="Whether the activity has heartrate data")
-    average_heartrate: float | None = Field(None, description="Average heartrate during activity")
-    max_heartrate: float | None = Field(None, description="Maximum heartrate during activity")
+    has_heartrate: bool = Field(
+        ..., description="Whether the activity has heartrate data"
+    )
+    average_heartrate: float | None = Field(
+        None, description="Average heartrate during activity"
+    )
+    max_heartrate: float | None = Field(
+        None, description="Maximum heartrate during activity"
+    )
     elev_high: float | None = Field(None, description="The highest elevation")
     elev_low: float | None = Field(None, description="The lowest elevation")
 
@@ -44,13 +56,19 @@ class DetailedActivity(Activity):
     description: str | None = Field(None, description="The description of the activity")
     athlete: dict = Field(..., description="The athlete who performed the activity")
     calories: float | None = Field(None, description="Calories burned during activity")
-    segment_efforts: list[dict] | None = Field(None, description="List of segment efforts")
+    segment_efforts: list[dict] | None = Field(
+        None, description="List of segment efforts"
+    )
     splits_metric: list[dict] | None = Field(None, description="Splits in metric units")
-    splits_standard: list[dict] | None = Field(None, description="Splits in standard units")
+    splits_standard: list[dict] | None = Field(
+        None, description="Splits in standard units"
+    )
     best_efforts: list[dict] | None = Field(None, description="List of best efforts")
     photos: dict | None = Field(None, description="Photos associated with activity")
     gear: dict | None = Field(None, description="Gear used during activity")
-    device_name: str | None = Field(None, description="Name of device used to record activity")
+    device_name: str | None = Field(
+        None, description="Name of device used to record activity"
+    )
 
 
 class Segment(BaseModel):
@@ -60,19 +78,35 @@ class Segment(BaseModel):
     name: str = Field(..., description="The name of the segment")
     activity_type: str = Field(..., description="The activity type of the segment")
     distance: float = Field(..., description="The segment's distance in meters")
-    average_grade: float = Field(..., description="The segment's average grade, in percents")
-    maximum_grade: float = Field(..., description="The segments's maximum grade, in percents")
-    elevation_high: float = Field(..., description="The segments's highest elevation, in meters")
-    elevation_low: float = Field(..., description="The segments's lowest elevation, in meters")
-    total_elevation_gain: float = Field(..., description="The segments's total elevation gain, in meters")
-    start_latlng: list[float] = Field(..., description="Start coordinates [latitude, longitude]")
-    end_latlng: list[float] = Field(..., description="End coordinates [latitude, longitude]")
+    average_grade: float = Field(
+        ..., description="The segment's average grade, in percents"
+    )
+    maximum_grade: float = Field(
+        ..., description="The segments's maximum grade, in percents"
+    )
+    elevation_high: float = Field(
+        ..., description="The segments's highest elevation, in meters"
+    )
+    elevation_low: float = Field(
+        ..., description="The segments's lowest elevation, in meters"
+    )
+    total_elevation_gain: float = Field(
+        ..., description="The segments's total elevation gain, in meters"
+    )
+    start_latlng: list[float] = Field(
+        ..., description="Start coordinates [latitude, longitude]"
+    )
+    end_latlng: list[float] = Field(
+        ..., description="End coordinates [latitude, longitude]"
+    )
     climb_category: int = Field(..., description="The category of the climb [0, 5]")
     city: str | None = Field(None, description="The city this segment is in")
     state: str | None = Field(None, description="The state this segment is in")
     country: str | None = Field(None, description="The country this segment is in")
     private: bool = Field(..., description="Whether this segment is private")
-    starred: bool = Field(..., description="Whether this segment is starred by the authenticated athlete")
+    starred: bool = Field(
+        ..., description="Whether this segment is starred by the authenticated athlete"
+    )
 
 
 class SegmentEffort(BaseModel):
@@ -85,13 +119,19 @@ class SegmentEffort(BaseModel):
     elapsed_time: int = Field(..., description="The elapsed time in seconds")
     moving_time: int = Field(..., description="The moving time in seconds")
     start_date: datetime = Field(..., description="Start date and time in UTC")
-    start_date_local: datetime = Field(..., description="Start date and time in athlete's timezone")
+    start_date_local: datetime = Field(
+        ..., description="Start date and time in athlete's timezone"
+    )
     distance: float = Field(..., description="The effort's distance in meters")
     average_watts: float | None = Field(None, description="Average wattage")
-    device_watts: bool | None = Field(None, description="Whether power data comes from a power meter")
+    device_watts: bool | None = Field(
+        None, description="Whether power data comes from a power meter"
+    )
     average_heartrate: float | None = Field(None, description="Average heartrate")
     max_heartrate: float | None = Field(None, description="Maximum heartrate")
-    pr_rank: int | None = Field(None, description="Personal record rank (1-3), 0 if not a PR")
+    pr_rank: int | None = Field(
+        None, description="Personal record rank (1-3), 0 if not a PR"
+    )
     achievements: list[dict] | None = Field(None, description="List of achievements")
     athlete: dict = Field(..., description="The athlete who performed the effort")
     segment: Segment = Field(..., description="The segment")
@@ -101,4 +141,5 @@ class ErrorResponse(BaseModel):
     """Represents an error response from the Strava API."""
 
     message: str = Field(..., description="Error message")
-    code: int = Field(..., description="Error code")
+    errors: list[dict] | None = Field(None, description="Detailed error information")
+    code: int | None = Field(None, description="Error code (optional)")