feat: standardize on ISO 8601 for dates, add utility functions, and document design decisions.
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m21s

This commit is contained in:
2026-05-12 23:09:00 +02:00
parent c56f7ad7b4
commit 3805ca3274
7 changed files with 250 additions and 68 deletions
+20 -23
View File
@@ -1,7 +1,8 @@
import json
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent
from mcp.types import TextContent, Annotations
from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import parse_iso_to_unix, format_date_iso, format_date_human
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
@@ -9,16 +10,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
ctx: Context,
limit: int = 10,
page: int = 1,
before: int | None = None,
after: int | None = None,
before: str | None = None,
after: str | None = None,
) -> list[TextContent]:
"""
List recent Strava activities for the authenticated user.
:param limit: Number of activities to return per page (default 10, max 200).
:param page: Page number for pagination (default 1).
:param before: Unix timestamp — only return activities before this time.
:param after: Unix timestamp — only return activities after this time.
Example: 1704067200 = 2024-01-01 00:00:00 UTC.
:param before: ISO 8601 date string (e.g., '2024-05-01T00:00:00Z') — only return activities before this time.
:param after: ISO 8601 date string (e.g., '2024-04-01T00:00:00Z') — only return activities after this time.
"""
try:
await ctx.info(f"Fetching activities (Page {page}, Limit {limit})...")
@@ -26,27 +26,19 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
activities = await strava.list_activities(
limit=min(limit, 200),
page=page,
before=before,
after=after,
before=parse_iso_to_unix(before),
after=parse_iso_to_unix(after),
)
from datetime import datetime
def format_date(d_str):
if not d_str:
return "N/A"
try:
d = datetime.fromisoformat(d_str.replace('Z', '+00:00'))
return f"{d.day:02d}.{d.month:02d}.{d.year} {d.hour:02d}:{d.minute:02d}"
except Exception:
return d_str
essential_data = []
for a in activities:
start_date_raw = a.get("start_date")
essential_data.append({
"id": a["id"],
"name": a["name"],
"sport_type": a.get("sport_type") or a.get("type"),
"start_date": format_date(a.get("start_date")),
"start_date": format_date_iso(start_date_raw),
"start_date_local": format_date_human(start_date_raw),
"distance": f"{a.get('distance', 0) / 1000:.2f} km",
"moving_time": f"{a.get('moving_time', 0) / 60:.1f} min",
"total_elevation_gain": f"{a.get('total_elevation_gain', 0):.0f} m",
@@ -62,13 +54,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
markdown_summary += "|-------|-------|------|---------|------|------------|------|\n"
for a in essential_data:
hr = f"{a['average_heartrate']:.0f} bpm" if a['average_heartrate'] else "-"
markdown_summary += f"| {a['start_date']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
markdown_summary += f"| {a['start_date_local']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
return [
TextContent(type="text", text=markdown_summary.strip()),
TextContent(
type="text",
text=f"Raw Data for Analysis: {json.dumps(essential_data, indent=2)}"
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"])
),
TextContent(
type="text",
text=json.dumps(essential_data, indent=2),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
@@ -112,7 +109,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"id": c.get("id"),
"text": c.get("text"),
"athlete": f"{c.get('athlete', {}).get('firstname')} {c.get('athlete', {}).get('lastname')}",
"created_at": c.get("created_at"),
"created_at": format_date_iso(c.get("created_at")),
}
for c in comments
]
+104 -41
View File
@@ -1,7 +1,8 @@
import json
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent
from mcp.types import TextContent, Annotations
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()
@@ -16,17 +17,6 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
athlete = await strava.get_athlete()
from datetime import datetime
def format_date(d_str):
if not d_str:
return "N/A"
try:
d = datetime.fromisoformat(d_str.replace('Z', '+00:00'))
return f"{d.day}.{d.month}.{d.year}"
except Exception:
return d_str
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"
@@ -40,8 +30,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"measurement_units": athlete.get("measurement_preference"),
"is_premium": athlete.get("premium", False),
"profile_medium": athlete.get("profile_medium"),
"created_at": athlete.get("created_at"),
"updated_at": athlete.get("updated_at"),
"created_at": format_date_iso(athlete.get("created_at")),
"updated_at": format_date_iso(athlete.get("updated_at")),
"bio": athlete.get("bio"),
"follower_count": athlete.get("follower_count"),
"friend_count": athlete.get("friend_count"),
@@ -56,15 +46,20 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
- 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(essential_data['created_at'])}
- Last Updated: {format_date(essential_data['updated_at'])}
- 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),
TextContent(
type="text",
text=f"Raw Data for Analysis: {json.dumps(essential_data, indent=2)}"
text=markdown_summary,
annotations=Annotations(audience=["user"])
),
TextContent(
type="text",
text=json.dumps(essential_data, indent=2),
annotations=Annotations(audience=["assistant"])
)
]
@@ -74,24 +69,64 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
return [TextContent(type="text", text=error_msg)]
@mcp.tool()
async def get_athlete_zones():
async def get_athlete_zones(ctx: Context) -> list[TextContent]:
"""
Get the heart rate and power zones configured for the authenticated athlete.
Returns zone boundaries for both heart rate and power (if a power meter is configured).
"""
try:
return await strava.get_athlete_zones()
await ctx.info("Fetching athlete zones...")
zones = await strava.get_athlete_zones()
markdown_summary = "### 💓 Trainingszonen\n\n"
# Heart Rate Zones
hr_zones = zones.get("heart_rate", {}).get("zones", [])
if hr_zones:
markdown_summary += "#### Herzfrequenz-Zonen\n"
markdown_summary += "| Zone | Bereich (bpm) |\n"
markdown_summary += "|------|---------------|\n"
for i, z in enumerate(hr_zones):
markdown_summary += f"| {i+1} | {z.get('min')} - {z.get('max') if z.get('max') != -1 else 'max'} |\n"
markdown_summary += "\n"
# Power Zones
power_zones = zones.get("power", {}).get("zones", [])
if power_zones:
markdown_summary += "#### Leistungs-Zonen (Power)\n"
markdown_summary += "| Zone | Bereich (W) |\n"
markdown_summary += "|------|-------------|\n"
for i, z in enumerate(power_zones):
markdown_summary += f"| {i+1} | {z.get('min')} - {z.get('max') if z.get('max') != -1 else 'max'} |\n"
if not hr_zones and not power_zones:
markdown_summary = "⚠️ Keine Trainingszonen konfiguriert."
return [
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"])
),
TextContent(
type="text",
text=json.dumps(zones, indent=2),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
return f"Error fetching athlete zones: {str(e)}"
error_msg = f"Error fetching athlete zones: {str(e)}"
await ctx.error(error_msg)
return [TextContent(type="text", text=error_msg)]
@mcp.tool()
async def get_athlete_stats():
async def get_athlete_stats(ctx: Context) -> list[TextContent]:
"""
Get cumulative training statistics for the authenticated Strava athlete.
Includes all-time, year-to-date, and recent (4-week) totals for runs, rides, and swims.
"""
import json
try:
await ctx.info("Fetching athlete statistics...")
stats = await strava.get_athlete_stats()
def fmt_sport(s: dict) -> dict:
@@ -102,23 +137,51 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"elevation_gain": f"{s.get('elevation_gain', 0):.0f} m",
}
result = {
"all_time": {
"runs": fmt_sport(stats.get("all_run_totals", {})),
"rides": fmt_sport(stats.get("all_ride_totals", {})),
"swims": fmt_sport(stats.get("all_swim_totals", {})),
},
"ytd": {
"runs": fmt_sport(stats.get("ytd_run_totals", {})),
"rides": fmt_sport(stats.get("ytd_ride_totals", {})),
"swims": fmt_sport(stats.get("ytd_swim_totals", {})),
},
"recent_4_weeks": {
"runs": fmt_sport(stats.get("recent_run_totals", {})),
"rides": fmt_sport(stats.get("recent_ride_totals", {})),
"swims": fmt_sport(stats.get("recent_swim_totals", {})),
},
# Prepare structured data for Markdown
all_time = {
"Laufen": fmt_sport(stats.get("all_run_totals", {})),
"Radfahren": fmt_sport(stats.get("all_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("all_swim_totals", {})),
}
return json.dumps(result, indent=2)
ytd = {
"Laufen": fmt_sport(stats.get("ytd_run_totals", {})),
"Radfahren": fmt_sport(stats.get("ytd_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("ytd_swim_totals", {})),
}
recent = {
"Laufen": fmt_sport(stats.get("recent_run_totals", {})),
"Radfahren": fmt_sport(stats.get("recent_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("recent_swim_totals", {})),
}
markdown_summary = "### 📈 Trainingsstatistiken\n\n"
def create_table(title: str, data: dict):
tbl = f"#### {title}\n"
tbl += "| Sport | Aktivitäten | Distanz | Zeit | Höhenmeter |\n"
tbl += "|-------|-------------|---------|------|------------|\n"
for sport, s in data.items():
if s["count"] > 0:
tbl += f"| {sport} | {s['count']} | {s['distance']} | {s['moving_time']} | {s['elevation_gain']} |\n"
return tbl + "\n"
markdown_summary += create_table("Letzte 4 Wochen", recent)
markdown_summary += create_table("Dieses Jahr (YTD)", ytd)
markdown_summary += create_table("Gesamt", all_time)
return [
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"])
),
TextContent(
type="text",
text=json.dumps(stats, indent=2),
annotations=Annotations(audience=["assistant"])
)
]
except Exception as e:
return f"Error fetching athlete stats: {str(e)}"
error_msg = f"Error fetching athlete stats: {str(e)}"
await ctx.error(error_msg)
return [TextContent(type="text", text=error_msg)]
+4 -3
View File
@@ -29,11 +29,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
:param weeks: Number of weeks to cover (default 4).
Fetches recent activities and athlete stats to produce a summary report.
"""
import time
after_ts = int(time.time()) - weeks * 7 * 24 * 3600
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')
return (
f"Please summarize my Strava training for the last {weeks} weeks.\n\n"
f"Use list_activities with after={after_ts} (Unix timestamp = {weeks} weeks ago) "
f"Use list_activities with after='{after_iso}' (ISO 8601) "
"and a high limit to fetch all recent activities. "
"Also read the get_athlete_stats tool for overall totals.\n\n"
"Structure the report as follows:\n"