Compare commits

23 Commits

Author SHA1 Message Date
matthias 598f3ac5da feat: parse --transport, --host, and --port flags in main.py to allow dynamic SSE/HTTP configurations, release 0.2.13
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m26s
2026-05-30 21:07:29 +02:00
matthias a30bef822b feat: add direct HTTP custom route /get-map/{activity_id} using FastMCP decorator and bump version to 0.2.12
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m30s
2026-05-30 20:56:07 +02:00
matthias 7d15789a0f bump: version to 0.2.11 for release
CI/CD Pipeline / Publish to PyPI (push) Successful in 14s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m35s
2026-05-30 20:39:00 +02:00
matthias 7d9f8daeaf fix: add padding newlines to activity map iframe HTML for improved formatting
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m24s
2026-05-30 20:37:33 +02:00
matthias 4c79d5179e feat: use highly strict pass-through LLM proxy instruction for map iframe tool output and release 0.2.10
CI/CD Pipeline / Publish to PyPI (push) Successful in 12s
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m29s
2026-05-30 20:28:34 +02:00
matthias 09af18ce7d feat: return base64-encoded HTML data-URI iframe inside get_activity_map tool to bypass gateway limits, keeping MCP App _meta block
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m28s
2026-05-30 20:09:46 +02:00
matthias f0b806a627 refactor: return str directly from resources to comply with FastMCP design and fix mimeType mismatch
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m33s
2026-05-30 19:38:22 +02:00
matthias 99ee6b7076 fix: enforce mime_type='text/html' inside FastMCP resource decorators and release 0.2.8 2026-05-30 19:34:45 +02:00
matthias 4d127541c5 refactor: remove old duplicate map resource declarations, clean generate_test_map and release version 0.2.7
CI/CD Pipeline / Publish to PyPI (push) Successful in 12s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m27s
2026-05-30 19:31:16 +02:00
matthias 04663d7ba1 feat: add static strava://activity/last/map resource to show up in MCP Inspector Resources tab
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m27s
2026-05-30 19:00:38 +02:00
matthias 3164e91be9 feat: expose activity map as a native MCP App resource with _meta ui resourceUri
CI/CD Pipeline / Publish to PyPI (push) Successful in 12s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m28s
2026-05-30 18:48:33 +02:00
matthias ef3b8305ab feat: output raw HTML in get_activity_map for native markdown block rendering
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m28s
2026-05-30 18:34:42 +02:00
matthias ab971b8307 feat: add LLM system context to get_activity_map to force verbatim iframe rendering
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m28s
2026-05-30 18:01:02 +02:00
matthias 62ab00bb23 feat: render activity map directly inline using iframe srcdoc
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m39s
2026-05-30 17:36:30 +02:00
matthias f231535f8e chore: ignore test.html in version control
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m32s
2026-05-30 13:21:39 +02:00
matthias badc2f870e feat(map): add start (🟢) and finish (🏁) route markers
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m32s
2026-05-30 13:20:17 +02:00
matthias d9523dd35b feat: add interactive activity map visualization with elevation charts and update test suite to support local map generation.
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Successful in 13s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m33s
2026-05-30 12:46:22 +02:00
matthias b4ab185240 feat: add get_activity_map tool for Leaflet-based route visualization and include corresponding unit tests
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m30s
2026-05-30 12:12:10 +02:00
matthias 39b182a87a feat: add emojis to statistics tables and use formatted markdown durations for activity times
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m33s
2026-05-30 11:56:08 +02:00
matthias b8c697fed3 feat: add speed and pace formatting utility and include in activity summary table
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m30s
2026-05-30 11:52:35 +02:00
matthias e7ec16abe5 feat: improve date formatting using local start times and add sport-specific emoji icons to activity summaries
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m30s
2026-05-30 11:47:08 +02:00
matthias 90e23c26b8 feat: implement format_duration_markdown utility and update tool outputs to use it for readable duration formatting
CI/CD Pipeline / Lint & Check (push) Successful in 11s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m34s
2026-05-30 11:40:51 +02:00
matthias e4f8a0c4e0 refactor: replace inline duration formatting with utility functions for consistent text and table output
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Publish to PyPI (push) Has been skipped
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m30s
2026-05-30 11:32:55 +02:00
12 changed files with 1347 additions and 47 deletions
+1
View File
@@ -28,3 +28,4 @@ build/
.Trashes
ehthumbs.db
Thumbs.db
test.html
+194
View File
@@ -0,0 +1,194 @@
import asyncio
import os
import sys
from dotenv import load_dotenv
# Add src directory to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src"))
from strava_mcp_server.strava_client import StravaClient
from mcp.server.fastmcp import Context
class MockContext(Context):
def __init__(self):
# Initialize parent with dummy request / info
pass
async def info(self, msg):
print(f"️ [INFO] {msg}")
async def warning(self, msg):
print(f"⚠️ [WARN] {msg}")
async def error(self, msg):
print(f"❌ [ERROR] {msg}")
async def generate():
from strava_mcp_server.config import get_config_file
# Load user config path
config_file = get_config_file()
if os.path.exists(config_file):
load_dotenv(dotenv_path=config_file)
# Fallback to local .env
load_dotenv()
client_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET")
refresh_token = os.getenv("STRAVA_REFRESH_TOKEN")
if not all([client_id, client_secret, refresh_token]):
print(
"️ Missing Strava credentials in .env file. Generating a beautiful MOCK route (Alpe d'Huez, France) for testing!"
)
# A real encoded polyline representing the famous Alpe d'Huez climb in France!
mock_polyline = (
"y~yxFn{_e@_AmAcAaBo@sAGo@Gs@H_ANc@jAyB^kALu@D_@Fk@KqBIkACe@DuABsAd@sBHi@@e@KoAKo@eAsC"
"e@oAMa@Bs@Jc@Nk@bAgBv@wAd@{AJc@Ds@Fk@GqAMmAEc@ByA@uAb@sBDi@@e@KoAKo@eAsCe@oAMa@Bs@Jc@Nk"
"@bAgBv@wAd@{AJc@Ds@Fk@GqAMmAEc@ByA@uAb@sBDi@@e@KoAKo@eAsCe@oAMa@Bs@Jc@Nk@bAgBv@wAd@{"
"AJc@Ds@Fk@GqAMmAEc@ByA@oB`@cCDq@@e@IoAKo@eAsCe@oAMi@FkADqAd@sBHi@@e@KoAKo@eAsCe@oAM"
)
# Import the tool function
from strava_mcp_server.tools.activities import register
class MockMCP:
def __init__(self):
self.tools = {}
self.resources = {}
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, path, **kwargs):
def decorator(func):
self.resources[path] = func
return func
return decorator
mcp = MockMCP()
# Mock StravaClient that returns our mock activity and streams
class MockClient:
async def get_activity(self, activity_id):
return {
"name": "🏔️ Alpe d'Huez Climb (Mock Test)",
"map": {"polyline": mock_polyline},
"distance": 14000,
"moving_time": 3600,
"total_elevation_gain": 1110,
"average_speed": 3.88,
"average_heartrate": 165,
"calories": 950,
}
async def get_activity_streams(self, activity_id, keys):
# Generate Alpe d'Huez climbing curve from 740m to 1850m
return [
{"type": "distance", "data": [i * 100 for i in range(141)]},
{
"type": "altitude",
"data": [
740 + int(1110 * (i / 140) ** 0.75) for i in range(141)
],
},
{
"type": "latlng",
"data": [
[45.1105 + (i * 0.0001), 6.0375 + (i * 0.0001)]
for i in range(141)
],
},
]
client = MockClient()
register(mcp, client)
print("Generating mock map HTML from resource...")
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
html_text = await resource_handler(99999)
if html_text:
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_text)
print(f"✅ Success! Mock map written to '{os.path.abspath(output_path)}'")
print(
"👉 Double-click 'test.html' or open it in your browser to view the interactive map!"
)
return
client = StravaClient()
print("Fetching recent activities to find one with GPS data...")
try:
activities = await client.list_activities(limit=20)
gps_activity = None
for a in activities:
map_data = a.get("map", {})
if map_data.get("summary_polyline") or map_data.get("polyline"):
gps_activity = a
break
if not gps_activity:
print(
"❌ No activity with GPS/polyline data found in your recent 20 activities."
)
return
activity_id = gps_activity["id"]
print(f"Found GPS Activity: '{gps_activity.get('name')}' (ID: {activity_id})")
# Import the tool function
from strava_mcp_server.tools.activities import register
class MockMCP:
def __init__(self):
self.tools = {}
self.resources = {}
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, path, **kwargs):
def decorator(func):
self.resources[path] = func
return func
return decorator
mcp = MockMCP()
register(mcp, client)
print("Generating map HTML from resource...")
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
html_text = await resource_handler(activity_id)
if html_text:
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_text)
print(f"✅ Success! Map written to '{os.path.abspath(output_path)}'")
print(
"👉 Double-click 'test.html' or open it in your browser to view the interactive map!"
)
except Exception as e:
print(f"❌ Failed: {str(e)}")
if __name__ == "__main__":
asyncio.run(generate())
+1 -1
View File
@@ -1,3 +1,3 @@
"""Strava MCP Server"""
__version__ = "0.1.0"
__version__ = "0.2.13"
+34 -5
View File
@@ -42,8 +42,32 @@ def main() -> None:
load_dotenv()
validate_credentials()
host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", 8000))
import argparse
parser = argparse.ArgumentParser(description="Strava MCP Server")
parser.add_argument(
"--transport",
choices=["stdio", "sse", "http"],
default=os.getenv("MCP_TRANSPORT", "stdio"),
help="Transport protocol (stdio, sse, or http/streamable-http)",
)
parser.add_argument(
"--host",
default=os.getenv("HOST", "0.0.0.0"),
help="Host address to bind to for HTTP / SSE",
)
parser.add_argument(
"--port",
type=int,
default=int(os.getenv("PORT", 8000)),
help="Port to listen on for HTTP / SSE",
)
args, unknown = parser.parse_known_args()
host = args.host
port = args.port
transport = args.transport
strava = StravaClient()
@@ -69,9 +93,6 @@ def main() -> None:
register_tools(mcp, strava)
# Check transport mode from environment (Default to stdio for local dev)
transport = os.getenv("MCP_TRANSPORT", "stdio")
if transport == "http":
# Run in Streamable HTTP mode (standard for Docker, K8s and OpenWebUI)
print(f"🚀 Starting Strava MCP Server on http://{host}:{port}")
@@ -80,6 +101,14 @@ def main() -> None:
mcp.run(transport="streamable-http")
except KeyboardInterrupt:
pass
elif transport == "sse":
# Run in Server-Sent Events (SSE) mode
print(f"🚀 Starting Strava MCP Server via SSE on http://{host}:{port}")
print(f" MCP endpoint: http://{host}:{port}/sse")
try:
mcp.run(transport="sse")
except KeyboardInterrupt:
pass
else:
# Run in STDIO mode (default for local testing and Claude Desktop)
mcp.run(transport="stdio")
+611 -16
View File
@@ -1,11 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from mcp.types import (
TextContent,
Annotations,
EmbeddedResource,
TextResourceContents,
CallToolResult,
)
from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import (
parse_iso_to_unix,
format_date_iso,
format_date_human,
format_duration_markdown,
get_sport_emoji,
format_speed_or_pace,
)
@@ -29,6 +38,19 @@ def _user_text(text: str) -> TextContent:
)
def _user_html_resource(uri: str, html: str) -> EmbeddedResource:
"""Helper: return a user-facing EmbeddedResource with text/html."""
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri,
mimeType="text/html",
text=html,
),
annotations=Annotations(audience=["user"]),
)
def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool()
async def list_activities(
@@ -58,17 +80,19 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
essential_data = []
for a in activities:
start_date_raw = a.get("start_date")
start_date_local_raw = a.get("start_date_local") or start_date_raw
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),
"start_date_local": format_date_human(start_date_local_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"),
"average_speed": a.get("average_speed"),
"gear_id": a.get("gear_id"),
}
)
@@ -79,19 +103,20 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
)
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:
markdown_summary += "| 📅 Datum | 🏷️ Name | 📐 Distanz | ⏱️ Zeit | ⚡ Ø Pace/Geschw. | 🏔️ Höhenmeter | ❤️ Ø HR |\n"
markdown_summary += "|----------|--------|-----------|--------|------------------|--------------|--------|\n"
for a, raw_act in zip(essential_data, activities):
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"
time_md = format_duration_markdown(raw_act.get("moving_time"))
sport_icon = get_sport_emoji(a["sport_type"])
speed_or_pace = format_speed_or_pace(
raw_act.get("average_speed"), a["sport_type"]
)
markdown_summary += f"| {a['start_date_local']} | {sport_icon} {a['name']} | {a['distance']} | {time_md} | {speed_or_pace} | {a['total_elevation_gain']} | {hr} |\n"
return [
_user_text(markdown_summary.strip()),
@@ -123,9 +148,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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"))
date = format_date_human(
activity.get("start_date_local") or activity.get("start_date")
)
dist = f"{activity.get('distance', 0) / 1000:.2f} km"
time = f"{activity.get('moving_time', 0) / 60:.1f} min"
time = format_duration_markdown(activity.get("moving_time"))
elev = f"{activity.get('total_elevation_gain', 0):.0f} m"
avg_hr = (
f"{activity.get('average_heartrate', 0):.0f} bpm"
@@ -270,13 +297,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md = f"### 🔄 Runden ({len(data)})\n"
md += "| # | Distanz | Zeit | Ø Speed | Ø HR |\n"
md += "|---|---------|------|---------|------|\n"
for lap in data:
for lap, raw_lap in zip(data, laps):
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"
time_md = format_duration_markdown(raw_lap.get("moving_time"))
md += f"| {lap['lap_index']} | {lap['distance']} | {time_md} | {lap['average_speed']} | {hr} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/laps", data),
@@ -320,9 +348,10 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
else "Leistung (Power)"
)
md += f"#### {label}\n| Zone | Bereich | Zeit |\n|------|---------|------|\n"
for b in buckets:
for b, raw_b in zip(buckets, zone.get("distribution_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"
time_md = format_duration_markdown(raw_b.get("time"))
md += f"| {b['zone']} | {b['min']} {max_val} | {time_md} |\n"
md += "\n"
return [
_user_text(md.strip()),
@@ -353,3 +382,569 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
return await strava.get_activity_streams(activity_id, key_list)
except Exception as e:
return f"Error fetching activity streams: {str(e)}"
async def _render_activity_map_html(
activity_id: int, ctx: Context | None = None
) -> str:
"""Helper: Fetches Strava GPS streams and builds the interactive Leaflet HTML template."""
try:
if ctx:
await ctx.info(
f"Fetching activity details for map (ID: {activity_id})..."
)
activity = await strava.get_activity(activity_id)
activity_map = activity.get("map", {})
polyline = activity_map.get("polyline") or activity_map.get(
"summary_polyline"
)
if not polyline:
warning_msg = f"⚠️ Keine GPS-Routen-Daten für diese Aktivität verfügbar (ID: {activity_id})."
if ctx:
await ctx.warning(warning_msg)
raise ValueError(warning_msg)
name = activity.get("name", "Aktivität")
sport_type = activity.get("sport_type") or activity.get("type", "Aktivität")
sport_emoji = get_sport_emoji(sport_type)
dist_val = f"{activity.get('distance', 0) / 1000:.2f} km"
time_val = format_duration_markdown(activity.get("moving_time"))
elev_val = f"{activity.get('total_elevation_gain', 0):.0f} m"
speed_val = format_speed_or_pace(activity.get("average_speed"), sport_type)
hr_val = (
f"{activity.get('average_heartrate', 0):.0f} bpm"
if activity.get("average_heartrate")
else "-"
)
cal_val = (
f"{activity.get('calories', 0):.0f} kcal"
if activity.get("calories")
else "-"
)
# Fetch streams for elevation chart and hover marker correlation
altitude_data = []
distance_data = []
latlng_data = []
try:
streams = await strava.get_activity_streams(
activity_id, ["distance", "altitude", "latlng"]
)
if isinstance(streams, dict):
altitude_stream = streams.get("altitude", {})
altitude_data = altitude_stream.get("data", [])
distance_stream = streams.get("distance", {})
distance_data = [
round(d / 1000, 2) for d in distance_stream.get("data", [])
]
latlng_stream = streams.get("latlng", {})
latlng_data = latlng_stream.get("data", [])
elif isinstance(streams, list):
for s in streams:
if isinstance(s, dict):
if s.get("type") == "altitude":
altitude_data = s.get("data", [])
elif s.get("type") == "distance":
distance_data = [
round(d / 1000, 2) for d in s.get("data", [])
]
elif s.get("type") == "latlng":
latlng_data = s.get("data", [])
except Exception as e:
if ctx:
await ctx.warning(
f"Could not fetch streams for elevation profile: {str(e)}"
)
# Downsample to max 150 points for smooth performance
max_points = 150
if len(altitude_data) > max_points:
step = len(altitude_data) // max_points
altitude_data = altitude_data[::step][:max_points]
distance_data = distance_data[::step][:max_points]
if latlng_data:
latlng_data = latlng_data[::step][:max_points]
has_elevation_chart = (
"true" if len(altitude_data) > 0 and len(distance_data) > 0 else "false"
)
html_content = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>{name}</title>
<link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css" />
<script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
html, body {{
margin: 0;
padding: 0;
height: 100%;
width: 100%;
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
background-color: #f8f9fa;
display: flex;
flex-direction: column;
overflow: hidden;
}}
.map-container {{
position: relative;
flex: 1;
width: 100%;
display: flex;
min-height: 350px;
}}
#map {{
flex: 1;
}}
#recenter-btn {{
position: absolute;
top: 12px;
right: 12px;
z-index: 1000;
background: #fc4c02;
color: white;
border: none;
padding: 8px 14px;
border-radius: 6px;
font-weight: 600;
font-size: 12px;
cursor: pointer;
box-shadow: 0 2px 4px rgba(0,0,0,0.15);
transition: background-color 0.2s, transform 0.1s;
display: flex;
align-items: center;
gap: 6px;
}}
#recenter-btn:hover {{
background: #e24300;
}}
#recenter-btn:active {{
transform: scale(0.97);
}}
.chart-container {{
background: #ffffff;
padding: 12px 20px;
border-top: 1px solid #e5e7eb;
box-shadow: 0 -4px 6px -1px rgba(0, 0, 0, 0.05);
height: 150px;
display: {"block" if has_elevation_chart == "true" else "none"};
}}
.stats-container {{
background: #ffffff;
padding: 16px 20px;
border-top: 1px solid #e5e7eb;
display: grid;
grid-template-columns: repeat(auto-fit, minmax(130px, 1fr));
gap: 12px;
}}
.stat-card {{
background: #f3f4f6;
padding: 10px 12px;
border-radius: 8px;
text-align: center;
display: flex;
flex-direction: column;
justify-content: center;
border: 1px solid #e5e7eb;
transition: transform 0.2s, box-shadow 0.2s;
}}
.stat-card:hover {{
transform: translateY(-2px);
box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.1);
}}
.stat-label {{
font-size: 10px;
font-weight: 600;
color: #6b7280;
text-transform: uppercase;
letter-spacing: 0.05em;
margin-bottom: 4px;
display: flex;
align-items: center;
justify-content: center;
gap: 4px;
}}
.stat-value {{
font-size: 15px;
font-weight: 700;
color: #111827;
}}
</style>
</head>
<body>
<div class="map-container">
<div id="map"></div>
<button id="recenter-btn" onclick="recenterMap()">🗺️ Route zentrieren</button>
</div>
<div class="chart-container">
<canvas id="elevationChart"></canvas>
</div>
<div class="stats-container">
<div class="stat-card">
<span class="stat-label">📐 Distanz</span>
<span class="stat-value">{dist_val}</span>
</div>
<div class="stat-card">
<span class="stat-label">⏱️ Zeit</span>
<span class="stat-value">{time_val}</span>
</div>
<div class="stat-card">
<span class="stat-label">⚡ Ø Pace/Geschw.</span>
<span class="stat-value">{speed_val}</span>
</div>
<div class="stat-card">
<span class="stat-label">🏔️ Höhenmeter</span>
<span class="stat-value">{elev_val}</span>
</div>
<div class="stat-card">
<span class="stat-label">❤️ Ø Herzfrequenz</span>
<span class="stat-value">{hr_val}</span>
</div>
<div class="stat-card">
<span class="stat-label">🔥 Kalorien</span>
<span class="stat-value">{cal_val}</span>
</div>
</div>
<script>
var osm = L.tileLayer('https://{{s}}.tile.openstreetmap.org/{{z}}/{{x}}/{{y}}.png', {{
attribution: '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> contributors'
}});
var satellite = L.tileLayer('https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{{z}}/{{y}}/{{x}}', {{
attribution: 'Tiles &copy; Esri &mdash; Source: Esri, USDA, USGS, AEX, GeoEye, IGN, and the GIS User Community'
}});
var map = L.map('map', {{
layers: [osm]
}});
var baseMaps = {{
"🗺️ Standard": osm,
"🛰️ Satellit": satellite
}};
L.control.layers(baseMaps, null, {{ position: 'topleft' }}).addTo(map);
var distanceData = {json.dumps(distance_data)};
var altitudeData = {json.dumps(altitude_data)};
var latlngData = {json.dumps(latlng_data)};
var track;
if (latlngData.length > 0) {{
track = L.polyline(latlngData, {{color: '#fc4c02', weight: 5, opacity: 0.8}}).addTo(map);
map.fitBounds(track.getBounds());
// Start- und End-Marker zeichnen
var startIcon = L.divIcon({{
html: '<span style="font-size: 18px; filter: drop-shadow(0 2px 4px rgba(0,0,0,0.3)); line-height: 18px; display: block; text-align: center;">🟢</span>',
iconSize: [18, 18],
iconAnchor: [9, 9],
className: 'start-marker-emoji'
}});
L.marker(latlngData[0], {{ icon: startIcon, zIndexOffset: 500 }}).addTo(map);
var endIcon = L.divIcon({{
html: '<span style="font-size: 22px; filter: drop-shadow(0 2px 4px rgba(0,0,0,0.3)); line-height: 22px; display: block; text-align: center;">🏁</span>',
iconSize: [22, 22],
iconAnchor: [11, 20],
className: 'end-marker-emoji'
}});
L.marker(latlngData[latlngData.length - 1], {{ icon: endIcon, zIndexOffset: 500 }}).addTo(map);
}} else {{
map.setView([0, 0], 2);
}}
function recenterMap() {{
if (track) {{
map.fitBounds(track.getBounds(), {{ animate: true, duration: 0.5 }});
}}
}}
// Render Elevation Chart if data is available
if ({has_elevation_chart}) {{
var ctx = document.getElementById('elevationChart').getContext('2d');
var sportEmoji = "{sport_emoji}";
var hoverIcon = L.divIcon({{
html: '<span style="font-size: 24px; filter: drop-shadow(0 2px 4px rgba(0,0,0,0.35)); line-height: 24px; display: block; text-align: center;">' + sportEmoji + '</span>',
iconSize: [24, 24],
iconAnchor: [12, 12],
className: 'hover-marker-emoji'
}});
var hoverMarker = L.marker([0, 0], {{ icon: hoverIcon }});
var hoverLinePlugin = {{
id: 'hoverLine',
afterDraw: function(chart) {{
if (chart.tooltip && chart.tooltip._active && chart.tooltip._active.length > 0) {{
var activePoint = chart.tooltip._active[0];
var ctx = chart.ctx;
var x = activePoint.element.x;
var topY = chart.scales.y.top;
var bottomY = chart.scales.y.bottom;
ctx.save();
ctx.beginPath();
ctx.setLineDash([5, 5]); // Gestrichelt
ctx.moveTo(x, topY);
ctx.lineTo(x, bottomY);
ctx.lineWidth = 1.5;
ctx.strokeStyle = 'rgba(252, 76, 2, 0.5)'; // Strava-Orange transparent
ctx.stroke();
ctx.restore();
}}
}}
}};
var elevationChartInstance = new Chart(ctx, {{
type: 'line',
data: {{
labels: distanceData,
datasets: [{{
label: 'Höhe (m)',
data: altitudeData,
borderColor: '#fc4c02',
backgroundColor: 'rgba(252, 76, 2, 0.08)',
borderWidth: 2.5,
fill: true,
tension: 0.4,
pointRadius: 0,
pointHoverRadius: 4,
pointHoverBackgroundColor: '#fc4c02'
}}]
}},
options: {{
responsive: true,
maintainAspectRatio: false,
interaction: {{
mode: 'index',
intersect: false
}},
onHover: function(event, activeElements) {{
if (activeElements && activeElements.length > 0 && latlngData && latlngData.length > 0) {{
var index = activeElements[0].index;
var coord = latlngData[index];
if (coord && coord.length === 2) {{
hoverMarker.setLatLng(coord);
if (!map.hasLayer(hoverMarker)) {{
hoverMarker.addTo(map);
}}
}}
}} else {{
if (map.hasLayer(hoverMarker)) {{
map.removeLayer(hoverMarker);
}}
}}
}},
plugins: {{
legend: {{ display: false }},
tooltip: {{
mode: 'index',
intersect: false,
callbacks: {{
title: function(context) {{
return 'Distanz: ' + context[0].label + ' km';
}},
label: function(context) {{
return 'Höhe: ' + context.raw + ' m';
}}
}}
}}
}},
scales: {{
x: {{
grid: {{ display: false }},
ticks: {{
maxTicksLimit: 10,
callback: function(value, index, values) {{
return distanceData[index] + ' km';
}}
}}
}},
y: {{
grid: {{ color: '#f3f4f6' }},
ticks: {{ maxTicksLimit: 4 }}
}}
}}
}},
plugins: [hoverLinePlugin]
}});
// Snapping coordinates from map hover to chart
map.on('mousemove', function(e) {{
if (!latlngData || latlngData.length === 0 || !elevationChartInstance) return;
var mouseLatLng = e.latlng;
var closestIndex = -1;
var minDistance = Infinity;
for (var i = 0; i < latlngData.length; i++) {{
var pt = L.latLng(latlngData[i]);
var dist = mouseLatLng.distanceTo(pt);
if (dist < minDistance) {{
minDistance = dist;
closestIndex = i;
}}
}}
// Threshold: 300 meters, dynamically adjusted based on zoom so it snaps comfortably
var zoom = map.getZoom();
var threshold = Math.max(100, Math.min(1000, 4000 / Math.pow(1.5, zoom - 12)));
if (closestIndex !== -1 && minDistance < threshold) {{
// Snap marker to the closest track coordinate
var snapCoord = latlngData[closestIndex];
hoverMarker.setLatLng(snapCoord);
if (!map.hasLayer(hoverMarker)) {{
hoverMarker.addTo(map);
}}
// Programmatically highlight point in Chart.js using precise pre-calculated element coordinates
var meta = elevationChartInstance.getDatasetMeta(0);
var element = meta.data[closestIndex];
if (element) {{
elevationChartInstance.setActiveElements([{{ datasetIndex: 0, index: closestIndex }}]);
elevationChartInstance.tooltip.setActiveElements([{{ datasetIndex: 0, index: closestIndex }}], {{
x: element.x,
y: element.y
}});
elevationChartInstance.update('none'); // Update without animation for instant snap performance
}}
}} else {{
// Hide marker and clear chart highlight if too far away
if (map.hasLayer(hoverMarker)) {{
map.removeLayer(hoverMarker);
}}
elevationChartInstance.setActiveElements([]);
elevationChartInstance.tooltip.setActiveElements([], {{ x: 0, y: 0 }});
elevationChartInstance.update('none');
}}
}});
// Also hide marker if mouse leaves the map container or canvas entirely
map.on('mouseout', function() {{
if (map.hasLayer(hoverMarker)) {{
map.removeLayer(hoverMarker);
}}
if (elevationChartInstance) {{
elevationChartInstance.setActiveElements([]);
elevationChartInstance.tooltip.setActiveElements([], {{ x: 0, y: 0 }});
elevationChartInstance.update('none');
}}
}});
document.getElementById('elevationChart').addEventListener('mouseleave', function() {{
if (map.hasLayer(hoverMarker)) {{
map.removeLayer(hoverMarker);
}}
if (elevationChartInstance) {{
elevationChartInstance.setActiveElements([]);
elevationChartInstance.tooltip.setActiveElements([], {{ x: 0, y: 0 }});
elevationChartInstance.update('none');
}}
}});
}}
</script>
</body>
</html>"""
return html_content.strip()
except Exception as e:
error_msg = f"Error rendering activity map: {str(e)}"
if ctx:
await ctx.error(error_msg)
raise RuntimeError(error_msg)
@mcp.tool()
async def get_activity_map(ctx: Context, activity_id: int) -> CallToolResult:
"""
Get an interactive map visualization for a specific Strava activity by its ID.
:param activity_id: The numeric ID of the Strava activity.
Returns a Base64 encoded interactive iframe and the dynamic resource URI.
"""
try:
# Check early that activity has GPS data so we can warn the user/LLM
await ctx.info(f"Checking GPS data for activity (ID: {activity_id})...")
activity = await strava.get_activity(activity_id)
activity_map = activity.get("map", {})
polyline = activity_map.get("polyline") or activity_map.get(
"summary_polyline"
)
if not polyline:
warning_msg = f"⚠️ Keine GPS-Routen-Daten für diese Aktivität verfügbar (ID: {activity_id})."
await ctx.warning(warning_msg)
return CallToolResult(
content=[TextContent(type="text", text=warning_msg)],
isError=True,
)
await ctx.info(f"Generating map HTML for activity (ID: {activity_id})...")
html_content = await _render_activity_map_html(activity_id, ctx)
import base64
b64_html = base64.b64encode(html_content.encode("utf-8")).decode("utf-8")
iframe_tag = f'\n\n<iframe src="data:text/html;base64,{b64_html}" style="width: 100%; height: 600px; border: none; border-radius: 12px;"></iframe>\n\n'
resource_uri = f"strava://activity/{activity_id}/map"
return CallToolResult(
content=[
TextContent(type="text", text=iframe_tag),
TextContent(
type="text",
text=(
"CRITICAL INSTRUCTION FOR THE LLM: You are a strict pass-through proxy. "
"Your ONLY task is to copy the <iframe> tag from the tool result above and print it "
"into the chat. Do NOT think about your capabilities. Do NOT explain that you are a text model. "
"Do NOT write any introduction or conclusion. Do NOT use markdown code blocks. "
"Print ONLY the raw <iframe> tag. If you write anything else, the UI will break."
),
),
],
_meta={"ui": {"resourceUri": resource_uri}},
)
except Exception as e:
error_msg = f"Error preparing activity map: {str(e)}"
await ctx.error(error_msg)
return CallToolResult(
content=[TextContent(type="text", text=error_msg)],
isError=True,
)
@mcp.resource("strava://activity/{activity_id}/map", mime_type="text/html")
async def get_activity_map_resource(activity_id: int) -> str:
"""HTML map resource for a specific Strava activity."""
return await _render_activity_map_html(activity_id)
@mcp.resource("strava://activity/last/map", mime_type="text/html")
async def get_last_activity_map_resource() -> str:
"""HTML map resource for the most recent Strava activity."""
try:
activities = await strava.list_activities(limit=1)
if not activities:
return "<html><body><p>Keine Strava-Aktivitäten gefunden.</p></body></html>"
last_id = activities[0]["id"]
return await _render_activity_map_html(last_id)
except Exception as e:
return f"<html><body><p>Error loading last activity: {str(e)}</p></body></html>"
from starlette.requests import Request
from starlette.responses import HTMLResponse
@mcp.custom_route("/get-map/{activity_id}", methods=["GET"])
async def get_web_map(request: Request) -> HTMLResponse:
"""HTTP Endpoint to directly serve the interactive Leaflet HTML map."""
activity_id_str = request.path_params.get("activity_id")
try:
activity_id = int(activity_id_str)
html_content = await _render_activity_map_html(activity_id)
return HTMLResponse(content=html_content)
except Exception as e:
return HTMLResponse(
content=f"<html><body><h3>Error loading map:</h3><p>{str(e)}</p></body></html>",
status_code=400,
)
+15 -14
View File
@@ -2,7 +2,7 @@ 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
from strava_mcp_server.utils import format_date_human, format_duration_markdown
def register(mcp: FastMCP, strava: StravaClient) -> None:
@@ -135,36 +135,37 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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",
"moving_time_raw": s.get("moving_time", 0),
"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", {})),
"🏃 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", {})),
"🏃 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", {})),
"🏃 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"
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"
time_md = format_duration_markdown(s["moving_time_raw"])
tbl += f"| {sport} | {s['count']} | {s['distance']} | {time_md} | {s['elevation_gain']} |\n"
return tbl + "\n"
markdown_summary += create_table("Letzte 4 Wochen", recent)
+4 -2
View File
@@ -2,6 +2,7 @@ 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_duration_markdown
def _resource(uri: str, data) -> EmbeddedResource:
@@ -91,8 +92,9 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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"
for a, raw_act in zip(data, activities):
time_md = format_duration_markdown(raw_act.get("moving_time"))
md += f"| {a['athlete']} | {a['sport_type']} | {a['name']} | {a['distance']} | {time_md} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://clubs/{club_id}/activities", data),
+8 -3
View File
@@ -2,6 +2,7 @@ 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_duration_markdown
def _resource(uri: str, data) -> EmbeddedResource:
@@ -55,9 +56,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md = f"### 🗺️ Routen ({len(data)})\n"
md += "| Name | Typ | Distanz | Höhenmeter | Dauer |\n"
md += "|------|-----|---------|------------|-------|\n"
for r in data:
for r, raw_r in zip(data, routes):
star = "" if r["starred"] else ""
md += f"| {star}{r['name']} | {r['type']} | {r['distance']} | {r['elevation_gain']} | {r['estimated_moving_time']} |\n"
time_md = format_duration_markdown(
raw_r.get("estimated_moving_time")
)
md += f"| {star}{r['name']} | {r['type']} | {r['distance']} | {r['elevation_gain']} | {time_md} |\n"
return [_user_text(md.strip()), _resource("internal://routes/list", data)]
except Exception as e:
return [TextContent(type="text", text=f"Error fetching routes: {str(e)}")]
@@ -96,13 +100,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
],
}
n_seg = len(data["segments"])
time_md = format_duration_markdown(r.get("estimated_moving_time"))
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"]} |
| Geschätzte Dauer | {time_md} |
| Segmente | {n_seg} |
| Favorit | {"⭐ Ja" if data["starred"] else "Nein"} |
| Privat | {"🔒 Ja" if data["private"] else "Nein"} |"""
+20 -6
View File
@@ -2,7 +2,11 @@ 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
from strava_mcp_server.utils import (
format_date_iso,
format_date_human,
format_duration_markdown,
)
def _resource(uri: str, data) -> EmbeddedResource:
@@ -53,13 +57,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
if data["average_heartrate"]
else "N/A"
)
time_elapsed = format_duration_markdown(e.get("elapsed_time"))
time_moving = format_duration_markdown(e.get("moving_time"))
date_local = format_date_human(
e.get("start_date_local") or e.get("start_date")
)
md = f"""### 🏅 Segment-Effort: {data["name"]}
| Feld | Wert |
|------|------|
| Datum | {format_date_human(data["start_date"])} |
| Datum | {date_local} |
| Distanz | {data["distance"]} |
| Zeit (gesamt) | {data["elapsed_time"]} |
| Fahrzeit | {data["moving_time"]} |
| Zeit (gesamt) | {time_elapsed} |
| Fahrzeit | {time_moving} |
| Ø Leistung | {w} |
| Ø Herzfrequenz | {hr} |
| PR-Rang | {pr} |
@@ -114,9 +123,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
md = f"### 🏅 Segment-Efforts ({len(data)})\n"
md += "| Datum | Zeit | Fahrzeit | PR-Rang |\n"
md += "|-------|------|----------|--------|\n"
for effort in data:
for effort, raw_eff in zip(data, efforts):
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"
elapsed_md = format_duration_markdown(raw_eff.get("elapsed_time"))
moving_md = format_duration_markdown(raw_eff.get("moving_time"))
date_local = format_date_human(
raw_eff.get("start_date_local") or raw_eff.get("start_date")
)
md += f"| {date_local} | {elapsed_md} | {moving_md} | {pr} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://segments/{segment_id}/efforts", data),
+138
View File
@@ -61,3 +61,141 @@ def format_date_human(date_input: Optional[str | datetime]) -> str:
return dt.strftime("%d.%m.%Y %H:%M")
except Exception:
return str(date_input)
def format_duration_text(seconds_input: Optional[float | int]) -> str:
"""
Formats a duration in seconds to a human-readable text string (Stunden und Minuten).
e.g., 9240 -> '2 Stunden 34 Minuten', 45 -> '1 Minute' (since 45s rounds to 1m), 0 -> '0 Minuten'.
"""
if seconds_input is None:
return "N/A"
try:
# Use round-half-up logic so 30 seconds correctly rounds up to 1 minute
total_minutes = int(float(seconds_input) / 60 + 0.5)
hours = total_minutes // 60
minutes = total_minutes % 60
parts = []
if hours > 0:
parts.append(f"{hours} {'Stunde' if hours == 1 else 'Stunden'}")
if minutes > 0 or not parts:
parts.append(f"{minutes} {'Minute' if minutes == 1 else 'Minuten'}")
return " ".join(parts)
except Exception:
return str(seconds_input)
def format_duration_table(seconds_input: Optional[float | int]) -> str:
"""
Formats a duration in seconds to a short table format (h:mm).
e.g., 9240 -> '2:34', 2700 -> '0:45'.
"""
if seconds_input is None:
return "N/A"
try:
# Use round-half-up logic so 30 seconds correctly rounds up to 1 minute
total_minutes = int(float(seconds_input) / 60 + 0.5)
hours = total_minutes // 60
minutes = total_minutes % 60
return f"{hours}:{minutes:02d}"
except Exception:
return str(seconds_input)
def format_duration_markdown(seconds_input: Optional[float | int]) -> str:
"""
Formats a duration in seconds to a readable markdown format: '1h 23min' or '45min'.
e.g., 9240 -> '2h 34min', 2700 -> '45min'.
"""
if seconds_input is None:
return "N/A"
try:
total_minutes = int(float(seconds_input) / 60 + 0.5)
hours = total_minutes // 60
minutes = total_minutes % 60
if hours > 0:
return f"{hours}h {minutes}min"
return f"{minutes}min"
except Exception:
return str(seconds_input)
def get_sport_emoji(sport_type: Optional[str]) -> str:
"""
Returns an emoji corresponding to the given sport type.
"""
if not sport_type:
return ""
sport_lower = sport_type.lower()
if "run" in sport_lower:
return "🏃"
elif "ride" in sport_lower or "bike" in sport_lower:
return "🚴"
elif "swim" in sport_lower:
return "🏊"
elif "walk" in sport_lower:
return "🚶"
elif "hike" in sport_lower:
return "🥾"
elif "ski" in sport_lower or "snowboard" in sport_lower:
return "🎿"
elif "row" in sport_lower or "canoe" in sport_lower or "kayak" in sport_lower:
return "🚣"
elif (
"weight" in sport_lower or "workout" in sport_lower or "crossfit" in sport_lower
):
return "🏋️"
elif "yoga" in sport_lower:
return "🧘"
elif "surf" in sport_lower:
return "🏄"
elif "skate" in sport_lower:
return "🛹"
else:
return "🏃"
def format_speed_or_pace(
average_speed_mps: Optional[float | int], sport_type: Optional[str]
) -> str:
"""
Formats speed/pace based on sport type.
- Runs, walks, hikes -> Pace (min/km, e.g. '5:30/km')
- Cycling and other sports -> Speed (km/h, e.g. '25.4 km/h')
"""
if not average_speed_mps or float(average_speed_mps) <= 0:
return "-"
try:
mps = float(average_speed_mps)
sport_lower = (sport_type or "").lower()
# Runs, walks, hikes get pace
if "run" in sport_lower or "walk" in sport_lower or "hike" in sport_lower:
pace_seconds_per_km = 1000 / mps
pace_minutes = int(pace_seconds_per_km // 60)
pace_seconds = int(round(pace_seconds_per_km % 60))
if pace_seconds == 60:
pace_minutes += 1
pace_seconds = 0
# Guard against crazy paces (e.g. standing still)
if pace_minutes > 59:
return "-"
return f"{pace_minutes}:{pace_seconds:02d}/km"
else:
# Cycling and other sports get speed in km/h
speed_kph = mps * 3.6
return f"{speed_kph:.1f} km/h"
except Exception:
return "-"
+163
View File
@@ -0,0 +1,163 @@
from unittest.mock import AsyncMock, MagicMock
import pytest
from mcp.server.fastmcp import Context
from strava_mcp_server.tools.activities import register
class MockMCP:
def __init__(self):
self.tools = {}
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, path, **kwargs):
if not hasattr(self, "resources"):
self.resources = {}
def decorator(func):
self.resources[path] = func
return func
return decorator
def custom_route(self, path, methods, **kwargs):
if not hasattr(self, "routes"):
self.routes = {}
def decorator(func):
self.routes[path] = func
return func
return decorator
@pytest.mark.asyncio
async def test_get_activity_map_success():
mcp = MockMCP()
strava = MagicMock()
# Mock detailed activity with a valid polyline
strava.get_activity = AsyncMock(
return_value={
"name": "Evening Ride",
"map": {"polyline": "a~l~FqzbwOq}@it@..."},
"sport_type": "Ride",
}
)
strava.get_activity_streams = AsyncMock(
return_value={
"altitude": {"data": [100.0, 101.0, 102.0]},
"distance": {"data": [0.0, 100.0, 200.0]},
"latlng": {"data": [[47.1, 9.1], [47.2, 9.2], [47.3, 9.3]]},
}
)
ctx = AsyncMock(spec=Context)
# Register tools
register(mcp, strava)
# Retrieve and call get_activity_map tool
get_activity_map = mcp.tools["get_activity_map"]
result = await get_activity_map(ctx, 12345)
assert result.isError is not True
assert result.meta == {"ui": {"resourceUri": "strava://activity/12345/map"}}
content = result.content
assert len(content) == 2
assert content[0].type == "text"
assert 'iframe src="data:text/html;base64,' in content[0].text
assert content[1].type == "text"
assert "CRITICAL INSTRUCTION FOR THE LLM" in content[1].text
# Now fetch and test the actual HTML from the registered resource handler
assert "strava://activity/{activity_id}/map" in mcp.resources
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
html_text = await resource_handler(12345)
assert isinstance(html_text, str)
assert "<!DOCTYPE html>" in html_text
assert "unpkg.com/leaflet" in html_text
assert "latlngData" in html_text
assert "[47.1, 9.1]" in html_text
assert 'sportEmoji = "🚴"' in html_text
assert "L.control.layers" in html_text
assert "satellite" in html_text
assert "🟢" in html_text
assert "🏁" in html_text
@pytest.mark.asyncio
async def test_get_activity_map_no_polyline():
mcp = MockMCP()
strava = MagicMock()
# Mock activity without GPS map info
strava.get_activity = AsyncMock(
return_value={
"name": "Indoor Trainer",
"map": {"polyline": None, "summary_polyline": None},
}
)
ctx = AsyncMock(spec=Context)
# Register tools
register(mcp, strava)
# Retrieve and call get_activity_map tool
get_activity_map = mcp.tools["get_activity_map"]
result = await get_activity_map(ctx, 67890)
assert result.isError is True
assert len(result.content) == 1
assert result.content[0].type == "text"
assert "⚠️ Keine GPS-Routen-Daten" in result.content[0].text
@pytest.mark.asyncio
async def test_get_web_map_route_success():
mcp = MockMCP()
strava = MagicMock()
# Mock detailed activity with a valid polyline
strava.get_activity = AsyncMock(
return_value={
"name": "Evening Ride",
"map": {"polyline": "a~l~FqzbwOq}@it@..."},
"sport_type": "Ride",
}
)
strava.get_activity_streams = AsyncMock(
return_value={
"altitude": {"data": [100.0, 101.0, 102.0]},
"distance": {"data": [0.0, 100.0, 200.0]},
"latlng": {"data": [[47.1, 9.1], [47.2, 9.2], [47.3, 9.3]]},
}
)
register(mcp, strava)
assert "/get-map/{activity_id}" in mcp.routes
route_handler = mcp.routes["/get-map/{activity_id}"]
# Mock Starlette Request
request = MagicMock()
request.path_params = {"activity_id": "12345"}
response = await route_handler(request)
from starlette.responses import HTMLResponse
assert isinstance(response, HTMLResponse)
assert response.status_code == 200
html_body = response.body.decode("utf-8")
assert "<!DOCTYPE html>" in html_body
assert "Evening Ride" in html_body
+158
View File
@@ -6,6 +6,11 @@ from strava_mcp_server.utils import (
parse_iso_to_unix,
format_date_iso,
format_date_human,
format_duration_text,
format_duration_table,
format_duration_markdown,
get_sport_emoji,
format_speed_or_pace,
)
@@ -96,3 +101,156 @@ class TestFormatDateHuman:
import re
assert re.match(r"\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}", result)
class TestFormatDurationText:
def test_none_returns_na(self):
assert format_duration_text(None) == "N/A"
def test_zero_seconds(self):
assert format_duration_text(0) == "0 Minuten"
def test_under_one_minute_rounds_to_zero_but_shows_zero(self):
# 20 seconds rounds to 0 minutes
assert format_duration_text(20) == "0 Minuten"
def test_half_minute_rounds_up(self):
# 30 seconds rounds up to 1 minute
assert format_duration_text(30) == "1 Minute"
def test_singular_minute(self):
# 60 seconds = 1 minute
assert format_duration_text(60) == "1 Minute"
def test_plural_minutes(self):
# 120 seconds = 2 minutes
assert format_duration_text(120) == "2 Minuten"
def test_singular_hour_and_plural_minutes(self):
# 3600 + 120 = 3720 seconds = 1 hour 2 minutes
assert format_duration_text(3720) == "1 Stunde 2 Minuten"
def test_plural_hours_and_singular_minute(self):
# 7200 + 60 = 7260 seconds = 2 hours 1 minute
assert format_duration_text(7260) == "2 Stunden 1 Minute"
def test_plural_hours_and_plural_minutes(self):
# 9240 seconds = 2 hours 34 minutes
assert format_duration_text(9240) == "2 Stunden 34 Minuten"
def test_only_hours_no_minutes(self):
# 7200 seconds = 2 hours 0 minutes
assert format_duration_text(7200) == "2 Stunden"
def test_fractional_input(self):
assert format_duration_text(9240.4) == "2 Stunden 34 Minuten"
class TestFormatDurationTable:
def test_none_returns_na(self):
assert format_duration_table(None) == "N/A"
def test_zero_seconds(self):
assert format_duration_table(0) == "0:00"
def test_under_one_hour(self):
# 2700 seconds = 45 minutes
assert format_duration_table(2700) == "0:45"
def test_over_one_hour(self):
# 9240 seconds = 2h 34m
assert format_duration_table(9240) == "2:34"
def test_hours_with_leading_zero_minute(self):
# 7260 seconds = 2h 1m -> '2:01'
assert format_duration_table(7260) == "2:01"
def test_fractional_input(self):
assert format_duration_table(2700.2) == "0:45"
class TestFormatDurationMarkdown:
def test_none_returns_na(self):
assert format_duration_markdown(None) == "N/A"
def test_zero_seconds(self):
assert format_duration_markdown(0) == "0min"
def test_under_one_hour(self):
# 2700 seconds = 45 minutes
assert format_duration_markdown(2700) == "45min"
def test_over_one_hour(self):
# 9240 seconds = 2h 34m
assert format_duration_markdown(9240) == "2h 34min"
def test_hours_with_zero_minutes(self):
# 7200 seconds = 2h 0m
assert format_duration_markdown(7200) == "2h 0min"
def test_fractional_input(self):
assert format_duration_markdown(2700.2) == "45min"
class TestGetSportEmoji:
def test_none_returns_question_mark(self):
assert get_sport_emoji(None) == ""
def test_empty_string_returns_question_mark(self):
assert get_sport_emoji("") == ""
def test_running(self):
assert get_sport_emoji("Run") == "🏃"
assert get_sport_emoji("VirtualRun") == "🏃"
def test_riding(self):
assert get_sport_emoji("Ride") == "🚴"
assert get_sport_emoji("VirtualRide") == "🚴"
assert get_sport_emoji("MountainBikeRide") == "🚴"
def test_swimming(self):
assert get_sport_emoji("Swim") == "🏊"
def test_walking_and_hiking(self):
assert get_sport_emoji("Walk") == "🚶"
assert get_sport_emoji("Hike") == "🥾"
def test_workout(self):
assert get_sport_emoji("Workout") == "🏋️"
assert get_sport_emoji("WeightTraining") == "🏋️"
def test_yoga(self):
assert get_sport_emoji("Yoga") == "🧘"
def test_unknown_returns_default(self):
assert get_sport_emoji("SomeOddSport") == "🏃"
class TestFormatSpeedOrPace:
def test_invalid_input_returns_hyphen(self):
assert format_speed_or_pace(None, "Run") == "-"
assert format_speed_or_pace(0, "Run") == "-"
assert format_speed_or_pace(-1, "Run") == "-"
def test_running_pace_exact(self):
# 3.0303 m/s = ~5:30 min/km
# 1000 / 3.0303 = 330 seconds = 5m 30s
assert format_speed_or_pace(3.0303, "Run") == "5:30/km"
def test_running_pace_rounds_correctly(self):
# 3.0211 m/s = 330.99 seconds -> rounds to 331s -> 5:31/km
assert format_speed_or_pace(3.0211, "VirtualRun") == "5:31/km"
def test_cycling_speed(self):
# 10 m/s = 36 km/h
assert format_speed_or_pace(10, "Ride") == "36.0 km/h"
# 7.5 m/s = 27 km/h
assert format_speed_or_pace(7.5, "MountainBikeRide") == "27.0 km/h"
def test_hike_gets_pace(self):
# 2 m/s = 500s/km = 8:20/km
assert format_speed_or_pace(2, "Hike") == "8:20/km"
def test_other_sports_get_speed(self):
# 10 m/s = 36 km/h
assert format_speed_or_pace(10, "Swim") == "36.0 km/h"