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 format_date_human def register(mcp: FastMCP, strava: StravaClient) -> None: @mcp.tool() async def get_athlete_profile(ctx: Context): """ Get the authenticated Strava athlete's profile. 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 = ", ".join(location_parts) if location_parts else "N/A" full_name = ( f"{athlete.get('firstname', '')} {athlete.get('lastname', '')}".strip() ) markdown_summary = f""" 👤 **Profile for {full_name}** (ID: {athlete.get("id")}) - Username: {athlete.get("username") or "N/A"} - Location: {location} - Sex: {athlete.get("sex") or "N/A"} - Weight: {athlete.get("weight") or "N/A"} kg - Measurement Units: {athlete.get("measurement_preference") or "N/A"} - Strava Summit Member: {"Yes" if athlete.get("premium") else "No"} - Profile Image (Medium): {athlete.get("profile_medium") or "N/A"} - Joined Strava: {format_date_human(athlete.get("created_at"))} - Last Updated: {format_date_human(athlete.get("updated_at"))} """.strip() return [ TextContent( type="text", text=markdown_summary, annotations=Annotations(audience=["user"]), ), EmbeddedResource( type="resource", resource=TextResourceContents( uri="internal://athlete/profile", mimeType="application/json", text=json.dumps(athlete, indent=2), ), annotations=Annotations(audience=["assistant"]), ), ] except Exception as e: error_msg = f"Error fetching athlete profile: {str(e)}" await ctx.error(error_msg) return [TextContent(type="text", text=error_msg)] @mcp.tool() async def get_athlete_zones(ctx: Context): """ 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: 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"]), ), EmbeddedResource( type="resource", resource=TextResourceContents( uri="internal://athlete/zones", mimeType="application/json", text=json.dumps(zones, indent=2), ), annotations=Annotations(audience=["assistant"]), ), ] except Exception as 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(ctx: Context): """ 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. """ try: await ctx.info("Fetching athlete statistics...") stats = await strava.get_athlete_stats() def fmt_sport(s: dict) -> dict: return { "count": s.get("count", 0), "distance": f"{s.get('distance', 0) / 1000:.1f} km", "moving_time": f"{s.get('moving_time', 0) / 3600:.1f} h", "elevation_gain": f"{s.get('elevation_gain', 0):.0f} m", } # 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", {})), } 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"]), ), EmbeddedResource( type="resource", resource=TextResourceContents( uri="internal://athlete/stats", mimeType="application/json", text=json.dumps(stats, indent=2), ), annotations=Annotations(audience=["assistant"]), ), ] except Exception as e: error_msg = f"Error fetching athlete stats: {str(e)}" await ctx.error(error_msg) return [TextContent(type="text", text=error_msg)]