hnrx/strava-mcp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat: enhance OAuth flow with synchronous token exchange and automatic .env file updates
CI/CD Pipeline / Lint & Check (push) Failing after 9s
Details
CI/CD Pipeline / Build & Push Docker Image (push) Has been skipped
Details

Browse Source
This commit is contained in:
Matthias Hinrichs
2026-05-10 11:15:00 +02:00
parent fafda14fe9
commit 578c4b292a
3 changed files with 211 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2
View File
@@ -40,6 +40,8 @@ Repository = "https://git.hnrx.net/hnrx/strava-mcp-server"
[project.scripts] [project.scripts]
strava-mcp = "strava_mcp_server.main:main" strava-mcp = "strava_mcp_server.main:main"
strava-mcp-get-token = "strava_mcp_server.get_token:main" strava-mcp-get-token = "strava_mcp_server.get_token:main"
server = "strava_mcp_server.main:main"
auth = "strava_mcp_server.get_token:main"
[dependency-groups] [dependency-groups]
dev = [ dev = [
src/strava_mcp_server/get_token.py
+109 -32
View File
@@ -25,33 +25,95 @@ CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET")
REDIRECT_URI = "http://localhost:8765/callback" REDIRECT_URI = "http://localhost:8765/callback"
SCOPES = "profile:read_all,activity:read_all,activity:read,profile:write" SCOPES = "profile:read_all,activity:read_all,activity:read,profile:write"
# Global to capture the auth code from the callback
auth_code: str | None = None
class CallbackHandler(BaseHTTPRequestHandler): class CallbackHandler(BaseHTTPRequestHandler):
client_id: str = ""
client_secret: str = ""
tokens: dict = {}
error: str | None = None
def do_GET(self): def do_GET(self):
global auth_code
parsed = urlparse(self.path) parsed = urlparse(self.path)
params = parse_qs(parsed.query) params = parse_qs(parsed.query)
if "code" in params: if "code" in params:
auth_code = params["code"][0] code = params["code"][0]
try:
# Exchange code for token synchronously
response = httpx.post(
"https://www.strava.com/oauth/token",
data={
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"grant_type": "authorization_code",
},
)
response.raise_for_status()
self.tokens = response.json()
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "text/html") self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers() self.end_headers()
self.wfile.write(b"""
<html><body style="font-family:sans-serif;text-align:center;padding:40px"> refresh_token = self.tokens.get("refresh_token")
<h2>&#x2705; Authorization successful!</h2>
<p>You can close this window and return to your terminal.</p> self.wfile.write(f"""
</body></html> <html>
""") <head>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; line-height: 1.6; color: #333; max-width: 800px; margin: 40px auto; padding: 20px; text-align: left; }}
.card {{ background: #f4f7f6; border-radius: 8px; padding: 20px; border-left: 5px solid #fc4c02; margin-top: 20px; }}
pre {{ background: #222; color: #fff; padding: 15px; border-radius: 5px; overflow-x: auto; font-family: "SF Mono", Monaco, Consolas, monospace; font-size: 13px; }}
.success-header {{ text-align: center; margin-bottom: 40px; }}
.success-icon {{ color: #2ecc71; font-size: 48px; display: block; margin-bottom: 10px; }}
h2, h3 {{ color: #fc4c02; }}
.env-label {{ font-weight: bold; color: #666; font-size: 12px; text-transform: uppercase; margin-bottom: 5px; display: block; }}
.copy-hint {{ font-size: 12px; color: #666; font-style: italic; }}
code {{ background: #eee; padding: 2px 4px; border-radius: 4px; }}
</style>
</head>
<body>
<div class="success-header">
<span class="success-icon">&#x2705;</span>
<h2>Authorization successful!</h2>
<p>You have successfully authenticated with Strava. You can now close this window.</p>
</div>
<h3>1. Local Setup (.env)</h3>
<p>Copy the following block into your <code>.env</code> file in the project root:</p>
<div class="card">
<span class="env-label">Your .env content:</span>
<pre>STRAVA_CLIENT_ID={self.client_id}
STRAVA_CLIENT_SECRET={self.client_secret}
STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<h3>2. Kubernetes Setup (Secret)</h3>
<p>If you are deploying this server to Kubernetes, run the following command to create the required Secret:</p>
<div class="card">
<span class="env-label">Kubectl Command:</span>
<pre>kubectl create secret generic strava-mcp-server-secret \\
--from-literal=STRAVA_CLIENT_ID={self.client_id} \\
--from-literal=STRAVA_CLIENT_SECRET={self.client_secret} \\
--from-literal=STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<p style="margin-top: 40px; font-size: 14px; color: #666; text-align: center;">
&mdash; Strava MCP Server Authorization Helper &mdash;
</p>
</body>
</html>
""".encode("utf-8"))
except Exception as e:
self.error = str(e)
self.send_response(500)
self.end_headers()
self.wfile.write(f"Error exchanging token: {e}".encode())
else: else:
error = params.get("error", ["unknown"])[0] error_msg = params.get("error", ["unknown"])[0]
self.send_response(400) self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers() self.end_headers()
self.wfile.write(f"<html><body><h2>&#x274C; Error: {error}</h2></body></html>".encode()) self.wfile.write(f"Error: {error_msg}".encode())
def log_message(self, format, *args): def log_message(self, format, *args):
pass # Suppress server logs pass # Suppress server logs
@@ -71,6 +133,12 @@ def main():
f"&scope={SCOPES}" f"&scope={SCOPES}"
) )
# Configure handler
CallbackHandler.client_id = CLIENT_ID
CallbackHandler.client_secret = CLIENT_SECRET
CallbackHandler.tokens = {}
CallbackHandler.error = None
print("=" * 60) print("=" * 60)
print(" Strava OAuth2 Authorization") print(" Strava OAuth2 Authorization")
print("=" * 60) print("=" * 60)
@@ -85,26 +153,15 @@ def main():
server = HTTPServer(("localhost", 8765), CallbackHandler) server = HTTPServer(("localhost", 8765), CallbackHandler)
server.handle_request() # Handle exactly one request (the callback) server.handle_request() # Handle exactly one request (the callback)
if not auth_code: if CallbackHandler.error:
print("❌ No authorization code received.") print(f"❌ Token exchange failed: {CallbackHandler.error}")
return return
print("\nExchanging authorization code for tokens...") if not CallbackHandler.tokens:
response = httpx.post( print("❌ No tokens received.")
"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:
print(f"❌ Token exchange failed: {response.status_code} {response.text}")
return return
data = response.json() data = CallbackHandler.tokens
refresh_token = data["refresh_token"] refresh_token = data["refresh_token"]
athlete = data.get("athlete", {}) athlete = data.get("athlete", {})
@@ -120,6 +177,26 @@ def main():
print(f"STRAVA_REFRESH_TOKEN={refresh_token}") print(f"STRAVA_REFRESH_TOKEN={refresh_token}")
print("-" * 40) print("-" * 40)
# Optional: Automatically update .env if it exists
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:
found = False
for line in lines:
if line.startswith("STRAVA_REFRESH_TOKEN="):
f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n")
found = True
else:
f.write(line)
if not found:
f.write(f"\nSTRAVA_REFRESH_TOKEN={refresh_token}\n")
print("Successfully updated your .env file!")
except Exception as e:
print(f"Could not automatically update .env: {e}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()
src/strava_mcp_server/tools/auth.py
+95 -35
View File
@@ -7,31 +7,96 @@ from urllib.parse import urlparse, parse_qs
from mcp.server.fastmcp import FastMCP, Context from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent from mcp.types import TextContent
auth_code: str | None = None
class CallbackHandler(BaseHTTPRequestHandler): class CallbackHandler(BaseHTTPRequestHandler):
client_id: str = ""
client_secret: str = ""
tokens: dict = {}
error: str | None = None
def do_GET(self): def do_GET(self):
global auth_code
parsed = urlparse(self.path) parsed = urlparse(self.path)
params = parse_qs(parsed.query) params = parse_qs(parsed.query)
if "code" in params: if "code" in params:
auth_code = params["code"][0] code = params["code"][0]
try:
# Exchange code for token synchronously inside the handler
import httpx
response = httpx.post(
"https://www.strava.com/oauth/token",
data={
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"grant_type": "authorization_code",
},
)
response.raise_for_status()
self.tokens = response.json()
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "text/html") self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers() self.end_headers()
self.wfile.write(b"""
<html><body style="font-family:sans-serif;text-align:center;padding:40px"> refresh_token = self.tokens.get("refresh_token")
<h2>&#x2705; Authorization successful!</h2>
<p>You can close this window and return to your terminal/chat.</p> self.wfile.write(f"""
</body></html> <html>
""") <head>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; line-height: 1.6; color: #333; max-width: 800px; margin: 40px auto; padding: 20px; text-align: left; }}
.card {{ background: #f4f7f6; border-radius: 8px; padding: 20px; border-left: 5px solid #fc4c02; margin-top: 20px; }}
pre {{ background: #222; color: #fff; padding: 15px; border-radius: 5px; overflow-x: auto; font-family: "SF Mono", Monaco, Consolas, monospace; font-size: 13px; }}
.success-header {{ text-align: center; margin-bottom: 40px; }}
.success-icon {{ color: #2ecc71; font-size: 48px; display: block; margin-bottom: 10px; }}
h2, h3 {{ color: #fc4c02; }}
.env-label {{ font-weight: bold; color: #666; font-size: 12px; text-transform: uppercase; margin-bottom: 5px; display: block; }}
.copy-hint {{ font-size: 12px; color: #666; font-style: italic; }}
code {{ background: #eee; padding: 2px 4px; border-radius: 4px; }}
</style>
</head>
<body>
<div class="success-header">
<span class="success-icon">&#x2705;</span>
<h2>Authorization successful!</h2>
<p>You have successfully authenticated with Strava. You can now close this window.</p>
</div>
<h3>1. Local Setup (.env)</h3>
<p>Copy the following block into your <code>.env</code> file in the project root:</p>
<div class="card">
<span class="env-label">Your .env content:</span>
<pre>STRAVA_CLIENT_ID={self.client_id}
STRAVA_CLIENT_SECRET={self.client_secret}
STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<h3>2. Kubernetes Setup (Secret)</h3>
<p>If you are deploying this server to Kubernetes, run the following command to create the required Secret:</p>
<div class="card">
<span class="env-label">Kubectl Command:</span>
<pre>kubectl create secret generic strava-mcp-server-secret \\
--from-literal=STRAVA_CLIENT_ID={self.client_id} \\
--from-literal=STRAVA_CLIENT_SECRET={self.client_secret} \\
--from-literal=STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<p style="margin-top: 40px; font-size: 14px; color: #666; text-align: center;">
&mdash; Strava MCP Server Authorization Helper &mdash;
</p>
</body>
</html>
""".encode("utf-8"))
except Exception as e:
self.error = str(e)
self.send_response(500)
self.end_headers()
self.wfile.write(f"Error exchanging token: {e}".encode())
else: else:
error = params.get("error", ["unknown"])[0] error_msg = params.get("error", ["unknown"])[0]
self.send_response(400) self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers() self.end_headers()
self.wfile.write(f"<html><body><h2>&#x274C; Error: {error}</h2></body></html>".encode()) self.wfile.write(f"Error: {error_msg}".encode())
def log_message(self, format, *args): def log_message(self, format, *args):
pass # Suppress server logs pass # Suppress server logs
@@ -44,14 +109,17 @@ def register(mcp: FastMCP, strava) -> None:
This opens a browser window for the user to log in and authorize the app. 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. 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_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET") client_secret = os.getenv("STRAVA_CLIENT_SECRET")
if not client_id or not 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")] return [TextContent(type="text", text="Error: Missing STRAVA_CLIENT_ID or STRAVA_CLIENT_SECRET in .env")]
# Configure handler with credentials
CallbackHandler.client_id = client_id
CallbackHandler.client_secret = client_secret
CallbackHandler.tokens = {}
CallbackHandler.error = None
redirect_uri = "http://localhost:8765/callback" redirect_uri = "http://localhost:8765/callback"
scopes = "profile:read_all,activity:read_all,activity:read,profile:write" scopes = "profile:read_all,activity:read_all,activity:read,profile:write"
@@ -73,27 +141,15 @@ def register(mcp: FastMCP, strava) -> None:
# Run handle_request in a separate thread so it doesn't block the async event loop # Run handle_request in a separate thread so it doesn't block the async event loop
await asyncio.to_thread(server.handle_request) await asyncio.to_thread(server.handle_request)
if not auth_code: if CallbackHandler.error:
return [TextContent(type="text", text="Error: No authorization code received.")] return [TextContent(type="text", text=f"Error during token exchange: {CallbackHandler.error}")]
await ctx.info("Authorization code received. Exchanging for tokens...") if not CallbackHandler.tokens:
return [TextContent(type="text", text="Error: No tokens received.")]
async with httpx.AsyncClient() as client: await ctx.info("Tokens received successfully.")
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: refresh_token = CallbackHandler.tokens.get("refresh_token")
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 # Update the .env file if it exists
env_msg = "" env_msg = ""
@@ -103,11 +159,15 @@ def register(mcp: FastMCP, strava) -> None:
with open(env_path, "r") as f: with open(env_path, "r") as f:
lines = f.readlines() lines = f.readlines()
with open(env_path, "w") as f: with open(env_path, "w") as f:
found = False
for line in lines: for line in lines:
if line.startswith("STRAVA_REFRESH_TOKEN="): if line.startswith("STRAVA_REFRESH_TOKEN="):
f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n") f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n")
found = True
else: else:
f.write(line) f.write(line)
if not found:
f.write(f"\nSTRAVA_REFRESH_TOKEN={refresh_token}\n")
env_msg = "\nI have also automatically updated your .env file with the new refresh token!" env_msg = "\nI have also automatically updated your .env file with the new refresh token!"
except Exception as e: except Exception as e:
env_msg = f"\nFailed to automatically update .env file: {e}" env_msg = f"\nFailed to automatically update .env file: {e}"