200 lines
8.7 KiB
Python
200 lines
8.7 KiB
Python
import json
|
|
from mcp.server.fastmcp import FastMCP, Context
|
|
from mcp.types import ContentBlock, TextContent, Annotations, EmbeddedResource, TextResourceContents
|
|
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):
|
|
"""
|
|
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"
|
|
|
|
essential_data = {
|
|
"id": athlete.get("id"),
|
|
"username": athlete.get("username"),
|
|
"name": f"{athlete.get('firstname')} {athlete.get('lastname')}".strip(),
|
|
"location": location,
|
|
"sex": athlete.get("sex"),
|
|
"weight": athlete.get("weight"),
|
|
"measurement_units": athlete.get("measurement_preference"),
|
|
"is_premium": athlete.get("premium", False),
|
|
"profile_medium": athlete.get("profile_medium"),
|
|
"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"),
|
|
}
|
|
|
|
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'])}
|
|
""".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(essential_data, 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)]
|