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

style: refactor codebase to adhere to PEP 8 formatting standards throughout all source files

Browse Source
This commit is contained in:
Matthias Hinrichs
2026-05-12 23:55:58 +02:00
parent bcc11cb07e
commit 8e9e4c01d4
17 changed files with 443 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+22
View File
@@ -32,6 +32,8 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
- [Project Structure](#project-structure)
- [Design Decisions](#design-decisions)
- [CI/CD (Gitea Actions)](#cicd-gitea-actions)
- [Development & Testing](#️-development--testing)
- [Git Hooks](#git-hooks)
- [Known Strava API Limitations](#known-strava-api-limitations)
- [Troubleshooting](#troubleshooting)
@@ -226,6 +228,26 @@ docker buildx build --platform linux/amd64,linux/arm64 -t strava-mcp-server:test
---
### 5. Git Hooks
A `pre-commit` hook is provided to automatically run `ruff check` on staged Python files before every commit, catching linting errors before they enter the history.
The hook is stored in the repository under `scripts/hooks/` for easy installation.
**Install the hook:**
```bash
cp scripts/hooks/pre-commit .git/hooks/pre-commit
chmod +x .git/hooks/pre-commit
```
**Behaviour:**
- ✅ Commit proceeds if `ruff check` passes on all staged `.py` files.
- ❌ Commit is aborted if any linting errors are found.
- 🔧 Auto-fix lint issues with `uv run ruff check --fix`.
- ⚡ Bypass for emergencies: `git commit --no-verify`.
---
## License
MIT
scripts/hooks/pre-commit
Executable
+26
View File
@@ -0,0 +1,26 @@
#!/bin/sh
# pre-commit hook: runs ruff check on staged Python files before every commit
# To bypass: git commit --no-verify
# Get list of staged Python files
STAGED_FILES=$(git diff --cached --name-only --diff-filter=ACM | grep '\.py$')
if [ -z "$STAGED_FILES" ]; then
exit 0
fi
echo "🔍 Running ruff check on staged files..."
uv run ruff check $STAGED_FILES
exit_code=$?
if [ $exit_code -ne 0 ]; then
echo ""
echo "❌ ruff check failed. Commit aborted."
echo " Fix the issues above or run: uv run ruff check --fix"
echo " To bypass: git commit --no-verify"
exit 1
fi
echo "✅ ruff check passed."
exit 0
src/strava_mcp_server/__init__.py
+1
View File
@@ -1,2 +1,3 @@
"""Strava MCP Server"""
__version__ = "0.1.0"
src/strava_mcp_server/get_token.py
+6 -2
View File
@@ -11,6 +11,7 @@ Required scopes:
Usage: uv run get_token.py
"""
import os
import webbrowser
import httpx
@@ -25,6 +26,7 @@ CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET")
REDIRECT_URI = "http://localhost:8765/callback"
SCOPES = "profile:read_all,activity:read_all,activity:read,profile:write"
class CallbackHandler(BaseHTTPRequestHandler):
client_id: str = ""
client_secret: str = ""
@@ -57,7 +59,8 @@ class CallbackHandler(BaseHTTPRequestHandler):
refresh_token = self.tokens.get("refresh_token")
self.wfile.write(f"""
self.wfile.write(
f"""
<html>
<head>
<style>
@@ -103,7 +106,8 @@ STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</p>
</body>
</html>
""".encode("utf-8"))
""".encode("utf-8")
)
except Exception as e:
self.error = str(e)
self.send_response(500)
src/strava_mcp_server/main.py
+3 -1
View File
@@ -26,7 +26,9 @@ def validate_credentials() -> None:
sys.exit(1)
if not os.getenv("STRAVA_REFRESH_TOKEN"):
print("ℹ️ No STRAVA_REFRESH_TOKEN found. Server starting in unauthenticated mode.")
print(
"ℹ️ No STRAVA_REFRESH_TOKEN found. Server starting in unauthenticated mode."
)
print(" Run 'uv run auth' on your local machine to authenticate.")
src/strava_mcp_server/strava_client.py
+60 -20
View File
@@ -37,7 +37,9 @@ class StravaClient:
async def get_valid_token(self) -> str:
"""Returns a valid access token, refreshing it if necessary."""
if not self.refresh_token:
raise ValueError("No Strava refresh token found. Please run 'uv run auth' on your local machine to authenticate first.")
raise ValueError(
"No Strava refresh token found. Please run 'uv run auth' on your local machine to authenticate first."
)
if not self.access_token or time.time() > self.expires_at - 60:
await self._refresh_access_token()
@@ -99,15 +101,27 @@ class StravaClient:
"""Gets the heart rate and power zones for a specific activity."""
return await self._get(f"activities/{activity_id}/zones")
async def get_activity_comments(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_activity_comments(
self, activity_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets comments on a specific activity."""
return await self._get(f"activities/{activity_id}/comments", params={"page": page, "per_page": per_page})
return await self._get(
f"activities/{activity_id}/comments",
params={"page": page, "per_page": per_page},
)
async def get_activity_kudoers(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_activity_kudoers(
self, activity_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets athletes who kudoed a specific activity."""
return await self._get(f"activities/{activity_id}/kudos", params={"page": page, "per_page": per_page})
return await self._get(
f"activities/{activity_id}/kudos",
params={"page": page, "per_page": per_page},
)
async def get_activity_streams(self, activity_id: int, keys: List[str]) -> Dict[str, Any]:
async def get_activity_streams(
self, activity_id: int, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific activity."""
return await self._get(
f"activities/{activity_id}/streams",
@@ -116,28 +130,45 @@ class StravaClient:
# ── Clubs ─────────────────────────────────────────────────────────────────
async def get_athlete_clubs(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_athlete_clubs(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets clubs the current athlete belongs to."""
return await self._get("athlete/clubs", params={"page": page, "per_page": per_page})
return await self._get(
"athlete/clubs", params={"page": page, "per_page": per_page}
)
async def get_club(self, club_id: int) -> Dict[str, Any]:
"""Gets a specific club by ID."""
return await self._get(f"clubs/{club_id}")
async def get_club_activities(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_club_activities(
self, club_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets recent activities from a club."""
return await self._get(f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page})
return await self._get(
f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page}
)
async def get_club_members(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_club_members(
self, club_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets members of a club."""
return await self._get(f"clubs/{club_id}/members", params={"page": page, "per_page": per_page})
return await self._get(
f"clubs/{club_id}/members", params={"page": page, "per_page": per_page}
)
# ── Routes ────────────────────────────────────────────────────────────────
async def get_routes_by_athlete(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_routes_by_athlete(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets routes created by the current athlete."""
athlete = await self.get_athlete()
return await self._get(f"athletes/{athlete['id']}/routes", params={"page": page, "per_page": per_page})
return await self._get(
f"athletes/{athlete['id']}/routes",
params={"page": page, "per_page": per_page},
)
async def get_route_by_id(self, route_id: str) -> Dict[str, Any]:
"""Gets a specific route by its ID."""
@@ -153,19 +184,26 @@ class StravaClient:
"""Gets a specific segment by ID."""
return await self._get(f"segments/{segment_id}")
async def get_starred_segments(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]:
async def get_starred_segments(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets segments starred by the current athlete."""
return await self._get("segments/starred", params={"page": page, "per_page": per_page})
return await self._get(
"segments/starred", params={"page": page, "per_page": per_page}
)
async def explore_segments(self, bounds: str, activity_type: str = "riding") -> Dict[str, Any]:
async def explore_segments(
self, bounds: str, activity_type: str = "riding"
) -> Dict[str, Any]:
"""Explores segments within a given bounding box."""
return await self._get(
"segments/explore",
params={"bounds": bounds, "activity_type": activity_type},
)
async def get_segment_streams(self, segment_id: int, keys: List[str]) -> Dict[str, Any]:
async def get_segment_streams(
self, segment_id: int, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific segment."""
return await self._get(
f"segments/{segment_id}/streams",
@@ -193,7 +231,9 @@ class StravaClient:
params["end_date_local"] = end_date_local
return await self._get("segment_efforts", params=params)
async def get_segment_effort_streams(self, effort_id: str, keys: List[str]) -> Dict[str, Any]:
async def get_segment_effort_streams(
self, effort_id: str, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific segment effort."""
return await self._get(
f"segment_efforts/{effort_id}/streams",
src/strava_mcp_server/tools/__init__.py
+2
View File
@@ -3,6 +3,7 @@ MCP Tool definitions for the Strava MCP Server.
Register all tools by calling register_tools(mcp, strava).
"""
from mcp.server.fastmcp import FastMCP
from strava_mcp_server.strava_client import StravaClient
@@ -15,6 +16,7 @@ from . import segment_efforts
from . import gear
from . import prompts
def register_tools(mcp: FastMCP, strava: StravaClient) -> None:
"""Register all available tools and prompts."""
athlete.register(mcp, strava)
src/strava_mcp_server/tools/activities.py
+85 -20
View File
@@ -2,7 +2,12 @@ import json
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import parse_iso_to_unix, format_date_iso, format_date_human
from strava_mcp_server.utils import (
parse_iso_to_unix,
format_date_iso,
format_date_human,
)
def _resource(uri: str, data) -> EmbeddedResource:
"""Helper: return an assistant-facing EmbeddedResource with application/json."""
@@ -16,9 +21,13 @@ def _resource(uri: str, data) -> EmbeddedResource:
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
"""Helper: return a user-facing TextContent."""
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
@@ -49,7 +58,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
essential_data = []
for a in activities:
start_date_raw = a.get("start_date")
essential_data.append({
essential_data.append(
{
"id": a["id"],
"name": a["name"],
"sport_type": a.get("sport_type") or a.get("type"),
@@ -60,16 +70,27 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"total_elevation_gain": f"{a.get('total_elevation_gain', 0):.0f} m",
"average_heartrate": a.get("average_heartrate"),
"gear_id": a.get("gear_id"),
})
}
)
if not essential_data:
markdown_summary = "### 📭 Keine Aktivitäten in diesem Zeitraum gefunden."
markdown_summary = (
"### 📭 Keine Aktivitäten in diesem Zeitraum gefunden."
)
else:
markdown_summary = f"### 🚴 Aktivitäten (Seite {page})\n"
markdown_summary += "| Datum | Sport | Name | Distanz | Zeit | Höhenmeter | Ø HR |\n"
markdown_summary += "|-------|-------|------|---------|------|------------|------|\n"
markdown_summary += (
"| Datum | Sport | Name | Distanz | Zeit | Höhenmeter | Ø HR |\n"
)
markdown_summary += (
"|-------|-------|------|---------|------|------------|------|\n"
)
for a in essential_data:
hr = f"{a['average_heartrate']:.0f} bpm" if a['average_heartrate'] else "-"
hr = (
f"{a['average_heartrate']:.0f} bpm"
if a["average_heartrate"]
else "-"
)
markdown_summary += f"| {a['start_date_local']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
return [
@@ -106,10 +127,26 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
dist = f"{activity.get('distance', 0) / 1000:.2f} km"
time = f"{activity.get('moving_time', 0) / 60:.1f} min"
elev = f"{activity.get('total_elevation_gain', 0):.0f} m"
avg_hr = f"{activity.get('average_heartrate', 0):.0f} bpm" if activity.get("average_heartrate") else "N/A"
max_hr = f"{activity.get('max_heartrate', 0):.0f} bpm" if activity.get("max_heartrate") else "N/A"
avg_spd = f"{activity.get('average_speed', 0) * 3.6:.1f} km/h" if activity.get("average_speed") else "N/A"
avg_w = f"{activity.get('average_watts', 0):.0f} W" if activity.get("average_watts") else "N/A"
avg_hr = (
f"{activity.get('average_heartrate', 0):.0f} bpm"
if activity.get("average_heartrate")
else "N/A"
)
max_hr = (
f"{activity.get('max_heartrate', 0):.0f} bpm"
if activity.get("max_heartrate")
else "N/A"
)
avg_spd = (
f"{activity.get('average_speed', 0) * 3.6:.1f} km/h"
if activity.get("average_speed")
else "N/A"
)
avg_w = (
f"{activity.get('average_watts', 0):.0f} W"
if activity.get("average_watts")
else "N/A"
)
gear = activity.get("gear_id") or "N/A"
n_efforts = len(activity.get("segment_efforts", []))
@@ -133,7 +170,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
_resource(f"internal://activities/{activity_id}", activity),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching activity details: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching activity details: {str(e)}"
)
]
@mcp.tool()
async def get_activity_comments(activity_id: int, limit: int = 30):
@@ -159,7 +200,10 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md = f"### 💬 Kommentare ({len(data)})\n"
for c in data:
md += f"- **{c['athlete']}** ({format_date_human(c['created_at'])}): {c['text']}\n"
return [_user_text(md.strip()), _resource(f"internal://activities/{activity_id}/comments", data)]
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/comments", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching comments: {str(e)}")]
@@ -189,7 +233,10 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
for k in data:
loc = ", ".join(filter(None, [k["city"], k["country"]])) or "N/A"
md += f"| {k['name']} | {loc} |\n"
return [_user_text(md.strip()), _resource(f"internal://activities/{activity_id}/kudoers", data)]
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/kudoers", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")]
@@ -224,9 +271,16 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md += "| # | Distanz | Zeit | Ø Speed | Ø HR |\n"
md += "|---|---------|------|---------|------|\n"
for lap in data:
hr = f"{lap['average_heartrate']:.0f} bpm" if lap['average_heartrate'] else "-"
hr = (
f"{lap['average_heartrate']:.0f} bpm"
if lap["average_heartrate"]
else "-"
)
md += f"| {lap['lap_index']} | {lap['distance']} | {lap['moving_time']} | {lap['average_speed']} | {hr} |\n"
return [_user_text(md.strip()), _resource(f"internal://activities/{activity_id}/laps", data)]
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/laps", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching laps: {str(e)}")]
@@ -260,15 +314,26 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"distribution_buckets": buckets,
}
data.append(zone_data)
label = "Herzfrequenz" if zone_data["type"] == "heartrate" else "Leistung (Power)"
label = (
"Herzfrequenz"
if zone_data["type"] == "heartrate"
else "Leistung (Power)"
)
md += f"#### {label}\n| Zone | Bereich | Zeit |\n|------|---------|------|\n"
for b in buckets:
max_val = "max" if b["max"] == -1 else str(b["max"])
md += f"| {b['zone']} | {b['min']} – {max_val} | {b['time_in_zone']} |\n"
md += "\n"
return [_user_text(md.strip()), _resource(f"internal://activities/{activity_id}/zones", data)]
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/zones", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching activity zones: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching activity zones: {str(e)}"
)
]
@mcp.tool()
async def get_activity_streams(
src/strava_mcp_server/tools/athlete.py
+32 -24
View File
@@ -4,6 +4,7 @@ from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceCo
from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import format_date_iso, format_date_human
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def get_athlete_profile(ctx: Context):
@@ -12,12 +13,19 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
Returns name, city, country, follower count, and other profile details.
"""
try:
await ctx.info("Fetching athlete profile...")
athlete = await strava.get_athlete()
location_parts = [p for p in (athlete.get('city'), athlete.get('state'), athlete.get('country')) if p]
location_parts = [
p
for p in (
athlete.get("city"),
athlete.get("state"),
athlete.get("country"),
)
if p
]
location = ", ".join(location_parts) if location_parts else "N/A"
essential_data = {
@@ -38,33 +46,33 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
markdown_summary = f"""
👤 **Profile for {essential_data['name']}** (ID: {essential_data['id']})
- Username: {essential_data['username'] or 'N/A'}
- Location: {essential_data['location']}
- Sex: {essential_data['sex'] or 'N/A'}
- Weight: {essential_data['weight'] or 'N/A'} kg
- Measurement Units: {essential_data['measurement_units'] or 'N/A'}
- Strava Summit Member: {'Yes' if essential_data['is_premium'] else 'No'}
- Profile Image (Medium): {essential_data['profile_medium'] or 'N/A'}
- Joined Strava: {format_date_human(essential_data['created_at'])}
- Last Updated: {format_date_human(essential_data['updated_at'])}
👤 **Profile for {essential_data["name"]}** (ID: {essential_data["id"]})
- Username: {essential_data["username"] or "N/A"}
- Location: {essential_data["location"]}
- Sex: {essential_data["sex"] or "N/A"}
- Weight: {essential_data["weight"] or "N/A"} kg
- Measurement Units: {essential_data["measurement_units"] or "N/A"}
- Strava Summit Member: {"Yes" if essential_data["is_premium"] else "No"}
- Profile Image (Medium): {essential_data["profile_medium"] or "N/A"}
- Joined Strava: {format_date_human(essential_data["created_at"])}
- Last Updated: {format_date_human(essential_data["updated_at"])}
""".strip()
return [
TextContent(
type="text",
text=markdown_summary,
annotations=Annotations(audience=["user"])
annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/profile",
mimeType="application/json",
text=json.dumps(essential_data, indent=2)
text=json.dumps(essential_data, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
@@ -110,17 +118,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"])
annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/zones",
mimeType="application/json",
text=json.dumps(zones, indent=2)
text=json.dumps(zones, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
error_msg = f"Error fetching athlete zones: {str(e)}"
@@ -181,17 +189,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"])
annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/stats",
mimeType="application/json",
text=json.dumps(stats, indent=2)
text=json.dumps(stats, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
error_msg = f"Error fetching athlete stats: {str(e)}"
src/strava_mcp_server/tools/clubs.py
+23 -6
View File
@@ -7,12 +7,17 @@ from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)),
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -88,9 +93,16 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md += "|--------|-------|------|---------|------|\n"
for a in data:
md += f"| {a['athlete']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} |\n"
return [_user_text(md.strip()), _resource(f"internal://clubs/{club_id}/activities", data)]
return [
_user_text(md.strip()),
_resource(f"internal://clubs/{club_id}/activities", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching club activities: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching club activities: {str(e)}"
)
]
@mcp.tool()
async def get_club_members(club_id: int, limit: int = 30):
@@ -118,6 +130,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
for m in data:
loc = ", ".join(filter(None, [m["city"], m["country"]])) or "N/A"
md += f"| {m['name']} | {loc} |\n"
return [_user_text(md.strip()), _resource(f"internal://clubs/{club_id}/members", data)]
return [
_user_text(md.strip()),
_resource(f"internal://clubs/{club_id}/members", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching club members: {str(e)}")]
return [
TextContent(type="text", text=f"Error fetching club members: {str(e)}")
]
src/strava_mcp_server/tools/gear.py
+21 -10
View File
@@ -7,12 +7,17 @@ from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)),
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -38,18 +43,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"primary": g.get("primary", False),
"retired": g.get("retired", False),
}
brand_model = " ".join(filter(None, [data["brand_name"], data["model_name"]])) or "N/A"
md = f"""### 🚲 Ausrüstung: {data['name'] or gear_id}
brand_model = (
" ".join(filter(None, [data["brand_name"], data["model_name"]]))
or "N/A"
)
md = f"""### 🚲 Ausrüstung: {data["name"] or gear_id}
| Feld | Wert |
|------|------|
| Marke / Modell | {brand_model} |
| Spitzname | {data['nickname'] or 'N/A'} |
| Typ | {data['type'] or 'N/A'} |
| Gesamt-Distanz | {data['distance']} |
| Primär | {'✅ Ja' if data['primary'] else 'Nein'} |
| Im Ruhestand | {'🛑 Ja' if data['retired'] else 'Nein'} |"""
| Spitzname | {data["nickname"] or "N/A"} |
| Typ | {data["type"] or "N/A"} |
| Gesamt-Distanz | {data["distance"]} |
| Primär | {"✅ Ja" if data["primary"] else "Nein"} |
| Im Ruhestand | {"🛑 Ja" if data["retired"] else "Nein"} |"""
if data["description"]:
md += f"\n\n_{data['description']}_"
return [_user_text(md.strip()), _resource(f"internal://gear/{gear_id}", data)]
return [
_user_text(md.strip()),
_resource(f"internal://gear/{gear_id}", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching gear: {str(e)}")]
src/strava_mcp_server/tools/prompts.py
+6 -2
View File
@@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP
from strava_mcp_server.strava_client import StravaClient
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.prompt()
def analyze_activity(activity_id: str) -> str:
@@ -30,8 +31,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
Fetches recent activities and athlete stats to produce a summary report.
"""
from datetime import datetime, timedelta, timezone
after_dt = (datetime.now(timezone.utc) - timedelta(weeks=weeks)).replace(hour=0, minute=0, second=0, microsecond=0)
after_iso = after_dt.isoformat().replace('+00:00', 'Z')
after_dt = (datetime.now(timezone.utc) - timedelta(weeks=weeks)).replace(
hour=0, minute=0, second=0, microsecond=0
)
after_iso = after_dt.isoformat().replace("+00:00", "Z")
return (
f"Please summarize my Strava training for the last {weeks} weeks.\n\n"
f"Use list_activities with after='{after_iso}' (ISO 8601) "
src/strava_mcp_server/tools/routes.py
+28 -12
View File
@@ -7,12 +7,17 @@ from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)),
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -30,7 +35,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"id": str(r.get("id")),
"name": r.get("name"),
"description": r.get("description") or "",
"type": "Run" if r.get("type") == 1 else "Ride" if r.get("type") == 2 else "Other",
"type": "Run"
if r.get("type") == 1
else "Ride"
if r.get("type") == 2
else "Other",
"sub_type": r.get("sub_type"),
"distance": f"{r.get('distance', 0) / 1000:.2f} km",
"elevation_gain": f"{r.get('elevation_gain', 0):.0f} m",
@@ -66,7 +75,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"id": str(r.get("id")),
"name": r.get("name"),
"description": r.get("description") or "",
"type": "Run" if r.get("type") == 1 else "Ride" if r.get("type") == 2 else "Other",
"type": "Run"
if r.get("type") == 1
else "Ride"
if r.get("type") == 2
else "Other",
"distance": f"{r.get('distance', 0) / 1000:.2f} km",
"elevation_gain": f"{r.get('elevation_gain', 0):.0f} m",
"estimated_moving_time": f"{r.get('estimated_moving_time', 0) / 60:.0f} min",
@@ -83,19 +96,22 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
],
}
n_seg = len(data["segments"])
md = f"""### 🗺️ Route: {data['name']}
md = f"""### 🗺️ Route: {data["name"]}
| Feld | Wert |
|------|------|
| Typ | {data['type']} |
| Distanz | {data['distance']} |
| Höhenmeter | {data['elevation_gain']} |
| Geschätzte Dauer | {data['estimated_moving_time']} |
| Typ | {data["type"]} |
| Distanz | {data["distance"]} |
| Höhenmeter | {data["elevation_gain"]} |
| Geschätzte Dauer | {data["estimated_moving_time"]} |
| Segmente | {n_seg} |
| Favorit | {'⭐ Ja' if data['starred'] else 'Nein'} |
| Privat | {'🔒 Ja' if data['private'] else 'Nein'} |"""
| Favorit | {"⭐ Ja" if data["starred"] else "Nein"} |
| Privat | {"🔒 Ja" if data["private"] else "Nein"} |"""
if data["description"]:
md += f"\n\n_{data['description']}_"
return [_user_text(md.strip()), _resource(f"internal://routes/{route_id}", data)]
return [
_user_text(md.strip()),
_resource(f"internal://routes/{route_id}", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching route: {str(e)}")]
src/strava_mcp_server/tools/segment_efforts.py
+35 -12
View File
@@ -8,12 +8,17 @@ from strava_mcp_server.utils import format_date_iso, format_date_human
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)),
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -43,21 +48,32 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
pr = f"#{data['pr_rank']}" if data["pr_rank"] else "N/A"
kom = f"#{data['kom_rank']}" if data["kom_rank"] else "N/A"
w = f"{data['average_watts']:.0f} W" if data["average_watts"] else "N/A"
hr = f"{data['average_heartrate']:.0f} bpm" if data["average_heartrate"] else "N/A"
md = f"""### 🏅 Segment-Effort: {data['name']}
hr = (
f"{data['average_heartrate']:.0f} bpm"
if data["average_heartrate"]
else "N/A"
)
md = f"""### 🏅 Segment-Effort: {data["name"]}
| Feld | Wert |
|------|------|
| Datum | {format_date_human(data['start_date'])} |
| Distanz | {data['distance']} |
| Zeit (gesamt) | {data['elapsed_time']} |
| Fahrzeit | {data['moving_time']} |
| Datum | {format_date_human(data["start_date"])} |
| Distanz | {data["distance"]} |
| Zeit (gesamt) | {data["elapsed_time"]} |
| Fahrzeit | {data["moving_time"]} |
| Ø Leistung | {w} |
| Ø Herzfrequenz | {hr} |
| PR-Rang | {pr} |
| KOM-Rang | {kom} |"""
return [_user_text(md.strip()), _resource(f"internal://segment_efforts/{effort_id}", data)]
return [
_user_text(md.strip()),
_resource(f"internal://segment_efforts/{effort_id}", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching segment effort: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching segment effort: {str(e)}"
)
]
@mcp.tool()
async def list_segment_efforts(
@@ -101,9 +117,16 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
for effort in data:
pr = f"#{effort['pr_rank']}" if effort["pr_rank"] else "-"
md += f"| {format_date_human(effort['start_date'])} | {effort['elapsed_time']} | {effort['moving_time']} | {pr} |\n"
return [_user_text(md.strip()), _resource(f"internal://segments/{segment_id}/efforts", data)]
return [
_user_text(md.strip()),
_resource(f"internal://segments/{segment_id}/efforts", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching segment efforts: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching segment efforts: {str(e)}"
)
]
@mcp.tool()
async def get_segment_effort_streams(
src/strava_mcp_server/tools/segments.py
+39 -19
View File
@@ -7,12 +7,17 @@ from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)),
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -43,22 +48,25 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"country": s.get("country"),
}
loc = ", ".join(filter(None, [data["city"], data["country"]])) or "N/A"
md = f"""### 📍 Segment: {data['name']}
md = f"""### 📍 Segment: {data["name"]}
| Feld | Wert |
|------|------|
| Sport | {data['activity_type']} |
| Distanz | {data['distance']} |
| Ø Steigung | {data['average_grade']} |
| Max Steigung | {data['maximum_grade']} |
| Höhe (hoch) | {data['elevation_high']} |
| Höhe (tief) | {data['elevation_low']} |
| Höhenmeter | {data['total_elevation_gain']} |
| Versuche | {data['effort_count']} |
| Athleten | {data['athlete_count']} |
| KOM/QOM | {data['kom'] or 'N/A'} |
| Sport | {data["activity_type"]} |
| Distanz | {data["distance"]} |
| Ø Steigung | {data["average_grade"]} |
| Max Steigung | {data["maximum_grade"]} |
| Höhe (hoch) | {data["elevation_high"]} |
| Höhe (tief) | {data["elevation_low"]} |
| Höhenmeter | {data["total_elevation_gain"]} |
| Versuche | {data["effort_count"]} |
| Athleten | {data["athlete_count"]} |
| KOM/QOM | {data["kom"] or "N/A"} |
| Ort | {loc} |
| Favorit | {'⭐ Ja' if data['starred'] else 'Nein'} |"""
return [_user_text(md.strip()), _resource(f"internal://segments/{segment_id}", data)]
| Favorit | {"⭐ Ja" if data["starred"] else "Nein"} |"""
return [
_user_text(md.strip()),
_resource(f"internal://segments/{segment_id}", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching segment: {str(e)}")]
@@ -90,9 +98,16 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md += "|------|-------|---------|------------|----------|\n"
for s in data:
md += f"| {s['name']} | {s['activity_type']} | {s['distance']} | {s['average_grade']} | {s['effort_count']} |\n"
return [_user_text(md.strip()), _resource("internal://segments/starred", data)]
return [
_user_text(md.strip()),
_resource("internal://segments/starred", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching starred segments: {str(e)}")]
return [
TextContent(
type="text", text=f"Error fetching starred segments: {str(e)}"
)
]
@mcp.tool()
async def explore_segments(
@@ -127,9 +142,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md += "|------|---------|------------|-----------|----------|\n"
for s in data:
md += f"| {s['name']} | {s['distance']} | {s['average_grade']} | {s['elevation_difference']} | {s['climb_category']} |\n"
return [_user_text(md.strip()), _resource("internal://segments/explore", data)]
return [
_user_text(md.strip()),
_resource("internal://segments/explore", data),
]
except Exception as e:
return [TextContent(type="text", text=f"Error exploring segments: {str(e)}")]
return [
TextContent(type="text", text=f"Error exploring segments: {str(e)}")
]
@mcp.tool()
async def get_segment_streams(
src/strava_mcp_server/utils.py
+7 -4
View File
@@ -1,6 +1,7 @@
from datetime import datetime, timezone
from typing import Optional
def parse_iso_to_unix(iso_str: Optional[str]) -> Optional[int]:
"""
Parses an ISO 8601 string into a Unix timestamp.
@@ -12,7 +13,7 @@ def parse_iso_to_unix(iso_str: Optional[str]) -> Optional[int]:
# Remove 'Z' and replace with +00:00 for fromisoformat compatibility if needed
# but fromisoformat in Python 3.11+ handles Z correctly.
# For older versions or varied formats, we use a slightly more robust approach.
clean_iso = iso_str.replace('Z', '+00:00')
clean_iso = iso_str.replace("Z", "+00:00")
dt = datetime.fromisoformat(clean_iso)
# Ensure it has a timezone; default to UTC if missing
if dt.tzinfo is None:
@@ -21,6 +22,7 @@ def parse_iso_to_unix(iso_str: Optional[str]) -> Optional[int]:
except Exception:
return None
def format_date_iso(date_input: Optional[str | datetime]) -> str:
"""
Standardizes a date string or datetime object to ISO 8601 (UTC).
@@ -31,17 +33,18 @@ def format_date_iso(date_input: Optional[str | datetime]) -> str:
try:
if isinstance(date_input, str):
# Strava dates are often '2024-01-01T12:00:00Z'
dt = datetime.fromisoformat(date_input.replace('Z', '+00:00'))
dt = datetime.fromisoformat(date_input.replace("Z", "+00:00"))
else:
dt = date_input
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat().replace('+00:00', 'Z')
return dt.isoformat().replace("+00:00", "Z")
except Exception:
return str(date_input)
def format_date_human(date_input: Optional[str | datetime]) -> str:
"""
Returns a human-readable date/time string (DD.MM.YYYY HH:MM).
@@ -51,7 +54,7 @@ def format_date_human(date_input: Optional[str | datetime]) -> str:
try:
if isinstance(date_input, str):
dt = datetime.fromisoformat(date_input.replace('Z', '+00:00'))
dt = datetime.fromisoformat(date_input.replace("Z", "+00:00"))
else:
dt = date_input
tests/test_strava.py
+11 -3
View File
@@ -3,6 +3,7 @@ import os
from strava_client import StravaClient
from dotenv import load_dotenv
async def test_connection():
load_dotenv()
@@ -12,7 +13,9 @@ async def test_connection():
if not all([client_id, client_secret, refresh_token]):
print("❌ Error: Missing Strava credentials in .env file.")
print("Please ensure STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, and STRAVA_REFRESH_TOKEN are set.")
print(
"Please ensure STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, and STRAVA_REFRESH_TOKEN are set."
)
return
client = StravaClient()
@@ -20,7 +23,9 @@ async def test_connection():
print("Testing Strava connection...")
try:
athlete = await client.get_athlete()
print(f"✅ Success! Connected as {athlete.get('firstname')} {athlete.get('lastname')}")
print(
f"✅ Success! Connected as {athlete.get('firstname')} {athlete.get('lastname')}"
)
print(f"Athlete ID: {athlete.get('id')}")
print("\nFetching recent activities...")
@@ -32,8 +37,11 @@ async def test_connection():
except Exception as e:
print(f"❌ Connection failed: {str(e)}")
if "401" in str(e):
print("Hint: This is likely a scope issue. Your token needs 'activity:read' permission.")
print(
"Hint: This is likely a scope issue. Your token needs 'activity:read' permission."
)
print("Run: uv run get_token.py to re-authorize with the correct scopes.")
if __name__ == "__main__":
asyncio.run(test_connection())