Strava MCP Server
by ctvidic
- src
#!/usr/bin/env python3
"""
HTTP server for Strava API integration.
This server exposes HTTP endpoints to query the Strava API for activities, athletes, and other data.
"""
import os
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from datetime import datetime, timedelta
from dotenv import load_dotenv
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import HTMLResponse
import uvicorn
import requests
import time
from map_utils import format_activity_with_map
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
stream=sys.stderr
)
logger = logging.getLogger(__name__)
# Strava API configuration
STRAVA_API_BASE_URL = "https://www.strava.com/api/v3"
strava_client: Optional[Dict[str, str]] = None
def refresh_access_token() -> bool:
"""Refresh the Strava access token using the refresh token."""
global strava_client
client_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET")
refresh_token = os.getenv("STRAVA_REFRESH_TOKEN")
try:
response = requests.post(
"https://www.strava.com/oauth/token",
data={
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token"
}
)
response.raise_for_status()
strava_client = response.json()
logger.info("Successfully refreshed Strava access token")
return True
except Exception as e:
logger.error(f"Failed to refresh access token: {str(e)}")
return False
def initialize_strava_client() -> None:
"""Initialize the Strava client using environment variables."""
global strava_client
# Load environment variables
env_path = Path(__file__).parent.parent / 'config' / '.env'
logger.info(f"Looking for .env file at: {env_path}")
if not env_path.exists():
logger.error(f"Environment file not found at {env_path}")
return
load_dotenv(dotenv_path=env_path)
logger.info("Environment variables loaded")
# Get credentials
client_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET")
refresh_token = os.getenv("STRAVA_REFRESH_TOKEN")
if not all([client_id, client_secret, refresh_token]):
logger.error("Missing Strava credentials in environment variables")
return
# Initialize by getting a fresh access token
if not refresh_access_token():
logger.error("Failed to initialize Strava client")
return
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for FastAPI application."""
# Startup
initialize_strava_client()
yield
# Shutdown
pass
# Create FastAPI app with lifespan
app = FastAPI(
title="Strava API Server",
description="HTTP server for Strava API integration",
lifespan=lifespan
)
def make_strava_request(url: str, method: str = "GET", params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Make a request to the Strava API with automatic token refresh."""
global strava_client
if not strava_client:
if not refresh_access_token():
raise HTTPException(status_code=401, detail="Not authenticated with Strava")
try:
headers = {"Authorization": f"Bearer {strava_client['access_token']}"}
response = requests.request(
method=method,
url=url,
headers=headers,
params=params
)
# If unauthorized, try refreshing token once
if response.status_code == 401:
if refresh_access_token():
headers = {"Authorization": f"Bearer {strava_client['access_token']}"}
response = requests.request(
method=method,
url=url,
headers=headers,
params=params
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Strava API request failed: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/auth/status")
def check_auth_status() -> Dict[str, Any]:
"""Check if we're authenticated with Strava."""
if not strava_client:
return {
"authenticated": False,
"message": "Not authenticated with Strava"
}
try:
profile = make_strava_request(f"{STRAVA_API_BASE_URL}/athlete")
return {
"authenticated": True,
"message": "Successfully authenticated with Strava",
"profile": profile
}
except Exception as e:
return {
"authenticated": False,
"message": f"Authentication error: {str(e)}"
}
@app.get("/activities/recent")
def get_recent_activities(limit: int = 10) -> List[Dict[str, Any]]:
"""Get recent activities from Strava."""
return make_strava_request(
f"{STRAVA_API_BASE_URL}/athlete/activities",
params={"per_page": limit}
)
@app.get("/activities/{activity_id}")
def get_activity(activity_id: int) -> Dict[str, Any]:
"""Get detailed activity data from Strava."""
return make_strava_request(f"{STRAVA_API_BASE_URL}/activities/{activity_id}")
@app.get("/athlete/stats")
def get_athlete_stats() -> Dict[str, Any]:
"""Get athlete statistics from Strava."""
# First get athlete ID
athlete = make_strava_request(f"{STRAVA_API_BASE_URL}/athlete")
athlete_id = athlete["id"]
# Then get stats
return make_strava_request(f"{STRAVA_API_BASE_URL}/athletes/{athlete_id}/stats")
@app.get("/activities/{activity_id}/map")
def get_activity_with_map(activity_id: int, format: str = 'html') -> Response:
"""Get detailed activity data from Strava with map visualization."""
try:
activity_data = make_strava_request(f"{STRAVA_API_BASE_URL}/activities/{activity_id}")
logger.debug(f"Retrieved activity data for ID {activity_id}")
formatted_activity = format_activity_with_map(activity_data, format)
logger.debug(f"Formatted activity data with {format} format")
if format == 'html':
return HTMLResponse(content=formatted_activity, media_type="text/html")
else:
return {
"formatted_activity": formatted_activity,
"activity": activity_data
}
except Exception as e:
logger.error(f"Error processing activity {activity_id}: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
def main() -> None:
"""Main entry point for the server."""
try:
logger.info("Starting Strava HTTP Server...")
uvicorn.run(app, host="0.0.0.0", port=8000)
except Exception as e:
logger.error(f"Server error: {str(e)}", exc_info=True)
raise
if __name__ == "__main__":
main()