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

refactor: enhance tool outputs by returning formatted markdown and JSON resources for structured display
CI/CD Pipeline / Lint & Check (push) Failing after 10s
Details
CI/CD Pipeline / Build & Push Docker Image (push) Has been skipped
Details

Browse Source
This commit is contained in:
Matthias Hinrichs
2026-05-12 23:36:37 +02:00
parent 7c089d90c5
commit 40b5d004b1
6 changed files with 319 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/strava_mcp_server/tools/activities.py
+98 -28
View File
@@ -4,6 +4,22 @@ from mcp.types import ContentBlock, TextContent, Annotations, EmbeddedResource,
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(
@@ -29,7 +45,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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")
@@ -57,20 +73,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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(),
annotations=Annotations(audience=["user"])
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://activities/list",
mimeType="application/json",
text=json.dumps(essential_data, indent=2)
),
annotations=Annotations(audience=["assistant"])
)
_user_text(markdown_summary.strip()),
_resource("internal://activities/list", essential_data),
]
except Exception as e:
error_msg = f"Error listing activities: {str(e)}"
@@ -95,9 +99,41 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
segment["id"] = str(segment["id"])
if "id" in activity:
activity["id"] = str(activity["id"])
return activity
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 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):
@@ -108,7 +144,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
comments = await strava.get_activity_comments(activity_id, per_page=limit)
return [
data = [
{
"id": c.get("id"),
"text": c.get("text"),
@@ -117,8 +153,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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 f"Error fetching comments: {str(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):
@@ -129,7 +172,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
kudoers = await strava.get_activity_kudoers(activity_id, per_page=limit)
return [
data = [
{
"id": k.get("id"),
"name": f"{k.get('firstname')} {k.get('lastname')}",
@@ -138,8 +181,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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 f"Error fetching kudoers: {str(e)}"
return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")]
@mcp.tool()
async def get_laps_by_activity_id(activity_id: int):
@@ -150,7 +202,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
laps = await strava.get_activity_laps(activity_id)
return [
data = [
{
"lap_index": lap.get("lap_index"),
"name": lap.get("name"),
@@ -165,8 +217,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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 f"Error fetching laps: {str(e)}"
return [TextContent(type="text", text=f"Error fetching laps: {str(e)}")]
@mcp.tool()
async def get_zones_by_activity_id(activity_id: int):
@@ -177,7 +239,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
zones = await strava.get_activity_zones(activity_id)
result = []
data = []
md = "### 💓 Zonen-Verteilung\n\n"
for zone in zones:
buckets = [
{
@@ -188,17 +251,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for i, b in enumerate(zone.get("distribution_buckets", []))
]
result.append({
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,
})
return result
}
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 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/clubs.py
+48 -6
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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)),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def list_athlete_clubs():
@@ -10,7 +24,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
clubs = await strava.get_athlete_clubs()
return [
data = [
{
"id": c.get("id"),
"name": c.get("name"),
@@ -22,8 +36,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for c in clubs
]
if not data:
md = "### 🏘️ Keine Clubs gefunden."
else:
md = f"### 🏘️ Clubs ({len(data)})\n"
md += "| Name | Sport | Mitglieder | Ort |\n"
md += "|------|-------|------------|-----|\n"
for c in data:
loc = ", ".join(filter(None, [c["city"], c["country"]])) or "N/A"
md += f"| {c['name']} | {c['sport_type'] or 'N/A'} | {c['member_count']} | {loc} |\n"
return [_user_text(md.strip()), _resource("internal://clubs/list", data)]
except Exception as e:
return f"Error fetching clubs: {str(e)}"
return [TextContent(type="text", text=f"Error fetching clubs: {str(e)}")]
@mcp.tool()
async def get_club(club_id: int):
@@ -45,7 +69,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
activities = await strava.get_club_activities(club_id, per_page=limit)
return [
data = [
{
"name": a.get("name"),
"sport_type": a.get("sport_type") or a.get("type"),
@@ -56,8 +80,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for a in activities
]
if not data:
md = "### 🚴 Keine Club-Aktivitäten gefunden."
else:
md = f"### 🚴 Club-Aktivitäten ({len(data)})\n"
md += "| Athlet | Sport | Name | Distanz | Zeit |\n"
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)]
except Exception as e:
return 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):
@@ -68,7 +101,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
members = await strava.get_club_members(club_id, per_page=limit)
return [
data = [
{
"id": m.get("id"),
"name": f"{m.get('firstname')} {m.get('lastname')}",
@@ -77,5 +110,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for m in members
]
if not data:
md = "### 👥 Keine Mitglieder gefunden."
else:
md = f"### 👥 Mitglieder ({len(data)})\n"
md += "| Name | Ort |\n|------|-----|\n"
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)]
except Exception as e:
return 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
+29 -2
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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)),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def get_gear_by_id(gear_id: str):
@@ -12,7 +26,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
g = await strava.get_gear_by_id(gear_id)
return {
data = {
"id": g.get("id"),
"name": g.get("name"),
"nickname": g.get("nickname"),
@@ -24,5 +38,18 @@ 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}
| 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'} |"""
if data["description"]:
md += f"\n\n_{data['description']}_"
return [_user_text(md.strip()), _resource(f"internal://gear/{gear_id}", data)]
except Exception as e:
return f"Error fetching gear: {str(e)}"
return [TextContent(type="text", text=f"Error fetching gear: {str(e)}")]
src/strava_mcp_server/tools/routes.py
+42 -4
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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)),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def get_routes_by_athlete_id(limit: int = 30):
@@ -11,7 +25,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
routes = await strava.get_routes_by_athlete(per_page=min(limit, 200))
return [
data = [
{
"id": str(r.get("id")),
"name": r.get("name"),
@@ -26,8 +40,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for r in routes
]
if not data:
md = "### 🗺️ Keine Routen gefunden."
else:
md = f"### 🗺️ Routen ({len(data)})\n"
md += "| Name | Typ | Distanz | Höhenmeter | Dauer |\n"
md += "|------|-----|---------|------------|-------|\n"
for r in data:
star = "⭐ " if r["starred"] else ""
md += f"| {star}{r['name']} | {r['type']} | {r['distance']} | {r['elevation_gain']} | {r['estimated_moving_time']} |\n"
return [_user_text(md.strip()), _resource("internal://routes/list", data)]
except Exception as e:
return f"Error fetching routes: {str(e)}"
return [TextContent(type="text", text=f"Error fetching routes: {str(e)}")]
@mcp.tool()
async def get_route_by_id(route_id: str):
@@ -38,7 +62,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
r = await strava.get_route_by_id(route_id)
return {
data = {
"id": str(r.get("id")),
"name": r.get("name"),
"description": r.get("description") or "",
@@ -58,8 +82,22 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
for s in r.get("segments", [])
],
}
n_seg = len(data["segments"])
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']} |
| Segmente | {n_seg} |
| 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)]
except Exception as e:
return f"Error fetching route: {str(e)}"
return [TextContent(type="text", text=f"Error fetching route: {str(e)}")]
@mcp.tool()
async def get_route_streams(route_id: str):
src/strava_mcp_server/tools/segment_efforts.py
+47 -6
View File
@@ -1,5 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import 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 _resource(uri: str, data) -> EmbeddedResource:
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:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
@@ -12,12 +27,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
e = await strava.get_segment_effort(effort_id)
return {
data = {
"id": str(e.get("id")),
"name": e.get("name"),
"elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min",
"moving_time": f"{e.get('moving_time', 0) / 60:.1f} min",
"start_date": e.get("start_date"),
"start_date": format_date_iso(e.get("start_date")),
"distance": f"{e.get('distance', 0) / 1000:.2f} km",
"average_watts": e.get("average_watts"),
"average_heartrate": e.get("average_heartrate"),
@@ -25,8 +40,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"pr_rank": e.get("pr_rank"),
"kom_rank": e.get("kom_rank"),
}
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']}
| Feld | Wert |
|------|------|
| 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)]
except Exception as e:
return 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(
@@ -49,20 +80,30 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
end_date_local=end_date_local,
per_page=limit,
)
return [
data = [
{
"id": str(e.get("id")),
"elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min",
"moving_time": f"{e.get('moving_time', 0) / 60:.1f} min",
"start_date": e.get("start_date"),
"start_date": format_date_iso(e.get("start_date")),
"average_watts": e.get("average_watts"),
"average_heartrate": e.get("average_heartrate"),
"pr_rank": e.get("pr_rank"),
}
for e in efforts
]
if not data:
md = "### 🏅 Keine Efforts für dieses Segment gefunden."
else:
md = f"### 🏅 Segment-Efforts ({len(data)})\n"
md += "| Datum | Zeit | Fahrzeit | PR-Rang |\n"
md += "|-------|------|----------|--------|\n"
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)]
except Exception as e:
return 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
+55 -6
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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)),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(type="text", text=text, annotations=Annotations(audience=["user"]))
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def get_segment(segment_id: int):
@@ -11,7 +25,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
s = await strava.get_segment(segment_id)
return {
data = {
"id": s.get("id"),
"name": s.get("name"),
"activity_type": s.get("activity_type"),
@@ -28,8 +42,25 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"city": s.get("city"),
"country": s.get("country"),
}
loc = ", ".join(filter(None, [data["city"], data["country"]])) or "N/A"
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'} |
| Ort | {loc} |
| Favorit | {'⭐ Ja' if data['starred'] else 'Nein'} |"""
return [_user_text(md.strip()), _resource(f"internal://segments/{segment_id}", data)]
except Exception as e:
return f"Error fetching segment: {str(e)}"
return [TextContent(type="text", text=f"Error fetching segment: {str(e)}")]
@mcp.tool()
async def list_starred_segments(limit: int = 30):
@@ -39,7 +70,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
segments = await strava.get_starred_segments(per_page=limit)
return [
data = [
{
"id": s.get("id"),
"name": s.get("name"),
@@ -51,8 +82,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for s in segments
]
if not data:
md = "### ⭐ Keine favorisierten Segmente."
else:
md = f"### ⭐ Favorisierte Segmente ({len(data)})\n"
md += "| Name | Sport | Distanz | Ø Steigung | Versuche |\n"
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)]
except Exception as e:
return 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(
@@ -68,7 +108,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
result = await strava.explore_segments(bounds, activity_type)
return [
data = [
{
"id": s.get("id"),
"name": s.get("name"),
@@ -79,8 +119,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for s in result.get("segments", [])
]
if not data:
md = "### 🗺️ Keine Segmente in diesem Bereich gefunden."
else:
md = f"### 🗺️ Segmente in der Region ({len(data)})\n"
md += "| Name | Distanz | Ø Steigung | Höhendiff | Kategorie |\n"
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)]
except Exception as e:
return f"Error exploring segments: {str(e)}"
return [TextContent(type="text", text=f"Error exploring segments: {str(e)}")]
@mcp.tool()
async def get_segment_streams(