strava_mcp/gradio_server.py

import logging
import os
from typing import Dict, List, Optional

import gradio as gr

from strava_mcp.config import StravaSettings
from strava_mcp.service import StravaService

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Global service instance
service: Optional[StravaService] = None


async def initialize_service(refresh_token: str = None):
    """Initialize the Strava service with settings from environment variables."""
    global service

    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",
        )

        if not settings.client_id:
            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)
        await service.initialize()
        logger.info("Initialized Strava service for Gradio")
        return service
    except Exception as e:
        logger.error(f"Failed to initialize Strava service: {str(e)}")
        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=read_all,activity:read,activity:read_all,profile:read_all&approval_prompt=force"

    instructions = f"""
🔐 **Manual OAuth Setup Instructions:**

**⚠️ IMPORTANT:** If you're getting "activity:read_permission missing" errors, it means your current refresh token was created without the correct scopes. You MUST follow these steps to get a new token.

1. **Click this link to authorize with Strava (with correct scopes):**
   {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 "Authentication" tab above.

🔍 **Required Scopes:** This URL includes the correct scopes: `read_all`, `activity:read`, `activity:read_all`, `profile:read_all`

⚠️ **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: str = "",
    after: str = "",
    page: str = "1",
    per_page: str = "30",
) -> list[dict]:
    """Get the authenticated user's activities from Strava.

    Args:
        before (str): An epoch timestamp for filtering activities before a certain time (leave empty for no filter)
        after (str): An epoch timestamp for filtering activities after a certain time (leave empty for no filter)
        page (str): Page number (default: 1)
        per_page (str): Number of items per page (default: 30, max: 200)

    Returns:
        List of activities with details like name, distance, moving time, etc.
    """
    # Convert string parameters to appropriate types
    before_int = int(before) if before.strip() else None
    after_int = int(after) if after.strip() else None
    page_int = int(page) if page.strip() else 1
    per_page_int = int(per_page) if per_page.strip() else 30
    try:
        await initialize_service()
        if service is None:
            raise ValueError("Service not initialized")

        activities = await service.get_activities(
            before_int, after_int, page_int, per_page_int
        )
        return [activity.model_dump() for activity in activities]
    except Exception as e:
        logger.error(f"Error getting user activities: {str(e)}")
        error_msg = str(e)

        # Check for scope-related errors and provide helpful guidance
        if "activity:read_permission" in error_msg and "missing" in error_msg:
            raise gr.Error(
                "❌ Authorization Error: Your refresh token doesn't have the required 'activity:read' permission. "
                "This means your token was created without the correct scopes. "
                "Please use the 'OAuth Helper' tab to get a new authorization URL with the correct scopes, "
                "then follow the steps to get a new refresh token with the proper permissions."
            )
        else:
            raise gr.Error(f"Failed to get activities: {error_msg}")


async def get_activity_details(
    activity_id: str,
    include_all_efforts: str = "false",
) -> dict:
    """Get detailed information about a specific activity.

    Args:
        activity_id (str): The unique ID of the activity
        include_all_efforts (str): Whether to include all segment efforts (true/false, default: false)

    Returns:
        Detailed activity information including stats, segments, and metadata
    """
    # Convert string parameters to appropriate types
    activity_id_int = int(activity_id)
    include_efforts_bool = include_all_efforts.lower().strip() == "true"
    try:
        await initialize_service()
        if service is None:
            raise ValueError("Service not initialized")

        activity = await service.get_activity(activity_id_int, include_efforts_bool)
        return activity.model_dump()
    except Exception as e:
        logger.error(f"Error getting activity details: {str(e)}")
        error_msg = str(e)

        # Check for scope-related errors and provide helpful guidance
        if "activity:read_permission" in error_msg and "missing" in error_msg:
            raise gr.Error(
                "❌ Authorization Error: Your refresh token doesn't have the required 'activity:read' permission. "
                "Please use the 'OAuth Helper' tab to get a new refresh token with the correct scopes."
            )
        else:
            raise gr.Error(f"Failed to get activity details: {error_msg}")


async def get_activity_segments(activity_id: str) -> list[dict]:
    """Get segment efforts for a specific activity.

    Args:
        activity_id (str): The unique ID of the activity

    Returns:
        List of segment efforts with performance data and rankings
    """
    # Convert string parameter to appropriate type
    activity_id_int = int(activity_id)
    try:
        await initialize_service()
        if service is None:
            raise ValueError("Service not initialized")

        segments = await service.get_activity_segments(activity_id_int)
        return [segment.model_dump() for segment in segments]
    except Exception as e:
        logger.error(f"Error getting activity segments: {str(e)}")
        error_msg = str(e)

        # Check for scope-related errors and provide helpful guidance
        if "activity:read_permission" in error_msg and "missing" in error_msg:
            raise gr.Error(
                "❌ Authorization Error: Your refresh token doesn't have the required 'activity:read' permission. "
                "Please use the 'OAuth Helper' tab to get a new refresh token with the correct scopes."
            )
        else:
            raise gr.Error(f"Failed to get activity segments: {error_msg}")


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,
        inputs=[
            gr.Textbox(
                label="Before (epoch timestamp)",
                value="",
                placeholder="Leave empty for no filter",
            ),
            gr.Textbox(
                label="After (epoch timestamp)",
                value="",
                placeholder="Leave empty for no filter",
            ),
            gr.Textbox(label="Page", value="1", placeholder="1"),
            gr.Textbox(label="Per Page", value="30", placeholder="30"),
        ],
        outputs=gr.JSON(label="Activities"),
        title="📊 Get User Activities",
        description="Retrieve the authenticated user's activities from Strava",
        api_name="get_user_activities",
    )

    # Activity details interface
    details_interface = gr.Interface(
        fn=get_activity_details,
        inputs=[
            gr.Textbox(label="Activity ID", placeholder="Enter activity ID"),
            gr.Textbox(
                label="Include All Efforts", value="false", placeholder="true or false"
            ),
        ],
        outputs=gr.JSON(label="Activity Details"),
        title="🔍 Get Activity Details",
        description="Get detailed information about a specific activity",
        api_name="get_activity_details",
    )

    # Activity segments interface
    segments_interface = gr.Interface(
        fn=get_activity_segments,
        inputs=[
            gr.Textbox(label="Activity ID", placeholder="Enter activity ID"),
        ],
        outputs=gr.JSON(label="Activity Segments"),
        title="🏃 Get Activity Segments",
        description="Get segment efforts for a specific activity",
        api_name="get_activity_segments",
    )

    # Combine interfaces in a tabbed interface
    demo = gr.TabbedInterface(
        [
            auth_interface,
            oauth_interface,
            activities_interface,
            details_interface,
            segments_interface,
        ],
        [
            "🔐 Authentication",
            "🔗 OAuth Helper",
            "📊 Activities",
            "🔍 Activity Details",
            "🏃 Activity Segments",
        ],
        title="Strava MCP Server",
    )

    return demo


def main():
    """Run the Gradio-based Strava MCP server."""
    logger.info("Starting Gradio web interface")

    demo = create_interface()

    # Launch Gradio web interface
    # Note: MCP server functionality requires Gradio 5.32.0+
    # Will be enabled automatically when HF Spaces updates Gradio version
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,  # Standard port for HF Spaces
        share=False,
        show_api=True,  # Ensure API endpoints are visible
    )


if __name__ == "__main__":
    main()