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, ) def _resource(uri: str, data) -> EmbeddedResource: """Helper: return an assistant-facing EmbeddedResource with application/json.""" return EmbeddedResource( type="resource", resource=TextResourceContents( uri=uri, mimeType="application/json", text=json.dumps(data, indent=2), ), 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"]) ) def register(mcp: FastMCP, strava: StravaClient) -> None: @mcp.tool() async def list_activities( ctx: Context, limit: int = 10, page: int = 1, before: str | None = None, after: str | None = None, ): """ 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: 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})...") activities = await strava.list_activities( limit=min(limit, 200), page=page, before=parse_iso_to_unix(before), after=parse_iso_to_unix(after), ) 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_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", "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." ) else: markdown_summary = f"### 🚴 Aktivitäten (Seite {page})\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 "-" ) markdown_summary += f"| {a['start_date_local']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n" return [ _user_text(markdown_summary.strip()), _resource("internal://activities/list", essential_data), ] except Exception as e: error_msg = f"Error listing activities: {str(e)}" await ctx.error(error_msg) return [TextContent(type="text", text=error_msg)] @mcp.tool() async def get_activity_details(activity_id: int): """ Get detailed information about a specific Strava activity by its ID. :param activity_id: The numeric ID of the Strava activity. Returns full activity data including segment efforts with string IDs safe for follow-up calls. """ try: activity = await strava.get_activity(activity_id) if "segment_efforts" in activity: for effort in activity["segment_efforts"]: if "id" in effort: effort["id"] = str(effort["id"]) segment = effort.get("segment") if segment and "id" in segment: segment["id"] = str(segment["id"]) if "id" in activity: activity["id"] = str(activity["id"]) name = activity.get("name", "N/A") sport = activity.get("sport_type") or activity.get("type", "N/A") date = format_date_human(activity.get("start_date")) 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" ) gear = activity.get("gear_id") or "N/A" n_efforts = len(activity.get("segment_efforts", [])) markdown_summary = f"""### 🏃 Aktivität: {name} | Feld | Wert | |------|------| | Sport | {sport} | | Datum | {date} | | Distanz | {dist} | | Zeit | {time} | | Höhenmeter | {elev} | | Ø Herzfrequenz | {avg_hr} | | Max Herzfrequenz | {max_hr} | | Ø Geschwindigkeit | {avg_spd} | | Ø Leistung | {avg_w} | | Ausrüstung | {gear} | | Segment-Efforts | {n_efforts} |""" return [ _user_text(markdown_summary.strip()), _resource(f"internal://activities/{activity_id}", activity), ] except Exception as 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): """ List comments on a specific Strava activity. :param activity_id: The numeric ID of the Strava activity. :param limit: Number of comments to return (default 30). """ try: comments = await strava.get_activity_comments(activity_id, per_page=limit) data = [ { "id": c.get("id"), "text": c.get("text"), "athlete": f"{c.get('athlete', {}).get('firstname')} {c.get('athlete', {}).get('lastname')}", "created_at": format_date_iso(c.get("created_at")), } for c in comments ] if not data: md = "### 💬 Keine Kommentare vorhanden." else: 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), ] except Exception as e: return [TextContent(type="text", text=f"Error fetching comments: {str(e)}")] @mcp.tool() async def get_activity_kudoers(activity_id: int, limit: int = 30): """ List athletes who gave kudos on a specific Strava activity. :param activity_id: The numeric ID of the Strava activity. :param limit: Number of kudoers to return (default 30). """ try: kudoers = await strava.get_activity_kudoers(activity_id, per_page=limit) data = [ { "id": k.get("id"), "name": f"{k.get('firstname')} {k.get('lastname')}", "city": k.get("city"), "country": k.get("country"), } for k in kudoers ] if not data: md = "### 👍 Noch keine Kudos." else: md = f"### 👍 Kudos ({len(data)})\n" md += "| Name | Ort |\n|------|-----|\n" 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), ] except Exception as e: return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")] @mcp.tool() async def get_laps_by_activity_id(activity_id: int): """ List all laps for a specific Strava activity. :param activity_id: The numeric ID of the Strava activity. Returns lap number, name, distance, moving time, average speed, and elevation gain. """ try: laps = await strava.get_activity_laps(activity_id) data = [ { "lap_index": lap.get("lap_index"), "name": lap.get("name"), "distance": f"{lap.get('distance', 0) / 1000:.3f} km", "moving_time": f"{lap.get('moving_time', 0) / 60:.1f} min", "elapsed_time": f"{lap.get('elapsed_time', 0) / 60:.1f} min", "average_speed": f"{lap.get('average_speed', 0) * 3.6:.2f} km/h", "max_speed": f"{lap.get('max_speed', 0) * 3.6:.2f} km/h", "elevation_gain": f"{lap.get('total_elevation_gain', 0):.0f} m", "average_heartrate": lap.get("average_heartrate"), "max_heartrate": lap.get("max_heartrate"), } for lap in laps ] if not data: md = "### 🔄 Keine Runden gefunden." else: md = f"### 🔄 Runden ({len(data)})\n" md += "| # | Distanz | Zeit | Ø Speed | Ø HR |\n" md += "|---|---------|------|---------|------|\n" for lap in data: 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), ] except Exception as e: return [TextContent(type="text", text=f"Error fetching laps: {str(e)}")] @mcp.tool() async def get_zones_by_activity_id(activity_id: int): """ Get heart rate and power zones distribution for a specific Strava activity. :param activity_id: The numeric ID of the Strava activity. Returns zone type, score, and time spent in each zone bucket. """ try: zones = await strava.get_activity_zones(activity_id) data = [] md = "### 💓 Zonen-Verteilung\n\n" for zone in zones: buckets = [ { "zone": i + 1, "min": b.get("min", 0), "max": b.get("max", -1), "time_in_zone": f"{b.get('time', 0) / 60:.1f} min", } for i, b in enumerate(zone.get("distribution_buckets", [])) ] zone_data = { "type": zone.get("type", "unknown"), "sensor_based": zone.get("sensor_based", False), "score": zone.get("score"), "custom_zones": zone.get("custom_zones", False), "points": zone.get("points"), "distribution_buckets": buckets, } data.append(zone_data) 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), ] except Exception as e: return [ TextContent( type="text", text=f"Error fetching activity zones: {str(e)}" ) ] @mcp.tool() async def get_activity_streams( activity_id: int, keys: str = "time,distance,heartrate,altitude,velocity_smooth,cadence,watts", ): """ Get raw data streams for a specific Strava activity. :param activity_id: The numeric ID of the Strava activity. :param keys: Comma-separated list of stream types to fetch. Available: time, distance, latlng, altitude, velocity_smooth, heartrate, cadence, watts, temp, moving, grade_smooth. Returns one data array per requested stream type. """ try: key_list = [k.strip() for k in keys.split(",") if k.strip()] return await strava.get_activity_streams(activity_id, key_list) except Exception as e: return f"Error fetching activity streams: {str(e)}"