import os
import webbrowser
import httpx
import asyncio
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent
auth_code: str | None = None
class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
global auth_code
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
if "code" in params:
auth_code = params["code"][0]
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(b"""
✅ Authorization successful!
You can close this window and return to your terminal/chat.
""")
else:
error = params.get("error", ["unknown"])[0]
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(f"❌ Error: {error}
".encode())
def log_message(self, format, *args):
pass # Suppress server logs
def register(mcp: FastMCP, strava) -> None:
@mcp.tool()
async def get_new_oauth_token(ctx: Context) -> list[TextContent]:
"""
Start the interactive Strava OAuth2 authorization flow.
This opens a browser window for the user to log in and authorize the app.
It then intercepts the redirect locally, obtains the token, and returns the tokens.
"""
global auth_code
auth_code = None
client_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET")
if not client_id or not client_secret:
return [TextContent(type="text", text="Error: Missing STRAVA_CLIENT_ID or STRAVA_CLIENT_SECRET in .env")]
redirect_uri = "http://localhost:8765/callback"
scopes = "profile:read_all,activity:read_all,activity:read,profile:write"
auth_url = (
f"https://www.strava.com/oauth/authorize"
f"?client_id={client_id}"
f"&redirect_uri={redirect_uri}"
f"&response_type=code"
f"&approval_prompt=force"
f"&scope={scopes}"
)
await ctx.info("Opening browser for Strava Authorization...")
webbrowser.open(auth_url)
await ctx.info("Waiting for you to log in and authorize (Browser opened on your computer)...")
server = HTTPServer(("localhost", 8765), CallbackHandler)
# Run handle_request in a separate thread so it doesn't block the async event loop
await asyncio.to_thread(server.handle_request)
if not auth_code:
return [TextContent(type="text", text="Error: No authorization code received.")]
await ctx.info("Authorization code received. Exchanging for tokens...")
async with httpx.AsyncClient() as client:
response = await client.post(
"https://www.strava.com/oauth/token",
data={
"client_id": client_id,
"client_secret": client_secret,
"code": auth_code,
"grant_type": "authorization_code",
},
)
if response.status_code != 200:
return [TextContent(type="text", text=f"Error: Token exchange failed: {response.status_code} {response.text}")]
data = response.json()
refresh_token = data.get("refresh_token")
# Update the .env file if it exists
env_msg = ""
try:
env_path = ".env"
if os.path.exists(env_path):
with open(env_path, "r") as f:
lines = f.readlines()
with open(env_path, "w") as f:
for line in lines:
if line.startswith("STRAVA_REFRESH_TOKEN="):
f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n")
else:
f.write(line)
env_msg = "\nI have also automatically updated your .env file with the new refresh token!"
except Exception as e:
env_msg = f"\nFailed to automatically update .env file: {e}"
return [TextContent(type="text", text=f"""
✅ Authorization successful!
You have successfully authenticated with Strava.
Your new Refresh Token is: `{refresh_token}`
{env_msg}
""")]