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

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
Details
CI/CD Pipeline / Lint & Check (push) Successful in 10s
Details
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m27s
Details

Browse Source
This commit is contained in:
Matthias Hinrichs
2026-05-30 19:31:16 +02:00
parent 04663d7ba1
commit 4d127541c5
4 changed files with 142 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_test_map.py
+24 -65
View File
@@ -8,7 +8,6 @@ sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src"))
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
from mcp.server.fastmcp import Context from mcp.server.fastmcp import Context
from mcp.types import TextContent, EmbeddedResource
class MockContext(Context): class MockContext(Context):
@@ -60,6 +59,7 @@ async def generate():
class MockMCP: class MockMCP:
def __init__(self): def __init__(self):
self.tools = {} self.tools = {}
self.resources = {}
def tool(self): def tool(self):
def decorator(func): def decorator(func):
@@ -68,6 +68,13 @@ async def generate():
return decorator return decorator
def resource(self, path):
def decorator(func):
self.resources[path] = func
return func
return decorator
mcp = MockMCP() mcp = MockMCP()
# Mock StravaClient that returns our mock activity and streams # Mock StravaClient that returns our mock activity and streams
@@ -106,35 +113,10 @@ async def generate():
client = MockClient() client = MockClient()
register(mcp, client) register(mcp, client)
get_activity_map = mcp.tools["get_activity_map"] print("Generating mock map HTML from resource...")
ctx = MockContext() resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
resource_result = await resource_handler(99999)
print("Generating mock map HTML...") html_text = resource_result.contents[0].text
result = await get_activity_map(ctx, 99999)
html_text = None
content_list = result.content if hasattr(result, "content") else result
for item in content_list:
if isinstance(item, TextContent):
if "<!DOCTYPE html>" in item.text:
html_text = item.text
break
elif '<iframe srcdoc="' in item.text:
prefix = '<iframe srcdoc="'
suffix = '" style="'
start = item.text.find(prefix)
if start != -1:
start += len(prefix)
end = item.text.find(suffix, start)
if end != -1:
html_text = item.text[start:end].replace("&quot;", '"')
break
elif (
isinstance(item, EmbeddedResource)
and item.resource.mimeType == "text/html"
):
html_text = item.resource.text
break
if html_text: if html_text:
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html") output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
@@ -173,6 +155,7 @@ async def generate():
class MockMCP: class MockMCP:
def __init__(self): def __init__(self):
self.tools = {} self.tools = {}
self.resources = {}
def tool(self): def tool(self):
def decorator(func): def decorator(func):
@@ -181,39 +164,20 @@ async def generate():
return decorator return decorator
def resource(self, path):
def decorator(func):
self.resources[path] = func
return func
return decorator
mcp = MockMCP() mcp = MockMCP()
register(mcp, client) register(mcp, client)
get_activity_map = mcp.tools["get_activity_map"] print("Generating map HTML from resource...")
ctx = MockContext() resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
resource_result = await resource_handler(activity_id)
print("Generating map HTML...") html_text = resource_result.contents[0].text
result = await get_activity_map(ctx, activity_id)
# Find the TextContent that holds the HTML
html_text = None
content_list = result.content if hasattr(result, "content") else result
for item in content_list:
if isinstance(item, TextContent):
if "<!DOCTYPE html>" in item.text:
html_text = item.text
break
elif '<iframe srcdoc="' in item.text:
prefix = '<iframe srcdoc="'
suffix = '" style="'
start = item.text.find(prefix)
if start != -1:
start += len(prefix)
end = item.text.find(suffix, start)
if end != -1:
html_text = item.text[start:end].replace("&quot;", '"')
break
elif (
isinstance(item, EmbeddedResource)
and item.resource.mimeType == "text/html"
):
html_text = item.resource.text
break
if html_text: if html_text:
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html") output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
@@ -223,11 +187,6 @@ async def generate():
print( print(
"👉 Double-click 'test.html' or open it in your browser to view the interactive map!" "👉 Double-click 'test.html' or open it in your browser to view the interactive map!"
) )
else:
# Maybe it returned a warning message instead of HTML
for item in result:
if isinstance(item, TextContent):
print(f"Returned message: {item.text}")
except Exception as e: except Exception as e:
print(f"❌ Failed: {str(e)}") print(f"❌ Failed: {str(e)}")
src/strava_mcp_server/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""Strava MCP Server""" """Strava MCP Server"""
__version__ = "0.2.6" __version__ = "0.2.7"
src/strava_mcp_server/tools/activities.py
+101 -67
View File
@@ -6,6 +6,7 @@ from mcp.types import (
EmbeddedResource, EmbeddedResource,
TextResourceContents, TextResourceContents,
CallToolResult, CallToolResult,
ReadResourceResult,
) )
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import ( from strava_mcp_server.utils import (
@@ -267,53 +268,6 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
except Exception as e: except Exception as e:
return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")] return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")]
@mcp.resource("strava://activity/{activity_id}/map")
async def get_activity_map_resource(activity_id: int) -> str:
"""HTML map resource for a specific Strava activity."""
class DummyContext:
async def info(self, msg):
pass
async def warning(self, msg):
pass
async def error(self, msg):
pass
ctx = DummyContext()
result = await get_activity_map(ctx, activity_id)
if result and result.content:
return result.content[0].text
return "<html><body><p>Error loading map</p></body></html>"
@mcp.resource("strava://activity/last/map")
async def get_last_activity_map_resource() -> str:
"""HTML map resource for the most recent Strava activity."""
class DummyContext:
async def info(self, msg):
pass
async def warning(self, msg):
pass
async def error(self, msg):
pass
ctx = DummyContext()
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"]
result = await get_activity_map(ctx, last_id)
if result and result.content:
return result.content[0].text
except Exception as e:
return f"<html><body><p>Error loading last activity: {str(e)}</p></body></html>"
return "<html><body><p>Error loading map</p></body></html>"
@mcp.tool() @mcp.tool()
async def get_laps_by_activity_id(activity_id: int): async def get_laps_by_activity_id(activity_id: int):
""" """
@@ -430,15 +384,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
except Exception as e: except Exception as e:
return f"Error fetching activity streams: {str(e)}" return f"Error fetching activity streams: {str(e)}"
@mcp.tool() async def _render_activity_map_html(
async def get_activity_map(ctx: Context, activity_id: int): activity_id: int, ctx: Context | None = None
""" ) -> str:
Get an interactive map visualization for a specific Strava activity by its ID. """Helper: Fetches Strava GPS streams and builds the interactive Leaflet HTML template."""
:param activity_id: The numeric ID of the Strava activity.
Returns a self-contained HTML page loading Leaflet.js and OpenStreetMap to render the route inside Open WebUI.
"""
try: try:
await ctx.info(f"Fetching activity details for map (ID: {activity_id})...") if ctx:
await ctx.info(
f"Fetching activity details for map (ID: {activity_id})..."
)
activity = await strava.get_activity(activity_id) activity = await strava.get_activity(activity_id)
activity_map = activity.get("map", {}) activity_map = activity.get("map", {})
@@ -448,11 +402,9 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
if not polyline: if not polyline:
warning_msg = f"⚠️ Keine GPS-Routen-Daten für diese Aktivität verfügbar (ID: {activity_id})." warning_msg = f"⚠️ Keine GPS-Routen-Daten für diese Aktivität verfügbar (ID: {activity_id})."
await ctx.warning(warning_msg) if ctx:
return CallToolResult( await ctx.warning(warning_msg)
content=[TextContent(type="text", text=warning_msg)], raise ValueError(warning_msg)
isError=True,
)
name = activity.get("name", "Aktivität") name = activity.get("name", "Aktivität")
sport_type = activity.get("sport_type") or activity.get("type", "Aktivität") sport_type = activity.get("sport_type") or activity.get("type", "Aktivität")
@@ -501,9 +453,10 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
elif s.get("type") == "latlng": elif s.get("type") == "latlng":
latlng_data = s.get("data", []) latlng_data = s.get("data", [])
except Exception as e: except Exception as e:
await ctx.warning( if ctx:
f"Could not fetch streams for elevation profile: {str(e)}" await ctx.warning(
) f"Could not fetch streams for elevation profile: {str(e)}"
)
# Downsample to max 150 points for smooth performance # Downsample to max 150 points for smooth performance
max_points = 150 max_points = 150
@@ -900,20 +853,101 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
</script> </script>
</body> </body>
</html>""" </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 the Resource URI where the interactive app is served.
"""
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,
)
resource_uri = f"strava://activity/{activity_id}/map" resource_uri = f"strava://activity/{activity_id}/map"
return CallToolResult( return CallToolResult(
content=[ content=[
TextContent(type="text", text=html_content.strip()),
TextContent( TextContent(
type="text", type="text",
text="[System: Render the above content directly as an HTML block so the UI can iframe it.]", text=f"Die interaktive Karte wurde vorbereitet. Du findest sie unter der URI: {resource_uri}",
), ),
], ],
_meta={"ui": {"resourceUri": resource_uri}}, _meta={"ui": {"resourceUri": resource_uri}},
) )
except Exception as e: except Exception as e:
error_msg = f"Error rendering activity map: {str(e)}" error_msg = f"Error preparing activity map: {str(e)}"
await ctx.error(error_msg) await ctx.error(error_msg)
return CallToolResult( return CallToolResult(
content=[TextContent(type="text", text=error_msg)], isError=True content=[TextContent(type="text", text=error_msg)],
isError=True,
)
@mcp.resource("strava://activity/{activity_id}/map")
async def get_activity_map_resource(activity_id: int) -> ReadResourceResult:
"""HTML map resource for a specific Strava activity."""
html_content = await _render_activity_map_html(activity_id)
return ReadResourceResult(
contents=[
TextResourceContents(
uri=f"strava://activity/{activity_id}/map",
mimeType="text/html",
text=html_content,
)
]
)
@mcp.resource("strava://activity/last/map")
async def get_last_activity_map_resource() -> ReadResourceResult:
"""HTML map resource for the most recent Strava activity."""
try:
activities = await strava.list_activities(limit=1)
if not activities:
return ReadResourceResult(
contents=[
TextResourceContents(
uri="strava://activity/last/map",
mimeType="text/html",
text="<html><body><p>Keine Strava-Aktivitäten gefunden.</p></body></html>",
)
]
)
last_id = activities[0]["id"]
html_content = await _render_activity_map_html(last_id)
return ReadResourceResult(
contents=[
TextResourceContents(
uri="strava://activity/last/map",
mimeType="text/html",
text=html_content,
)
]
)
except Exception as e:
return ReadResourceResult(
contents=[
TextResourceContents(
uri="strava://activity/last/map",
mimeType="text/html",
text=f"<html><body><p>Error loading last activity: {str(e)}</p></body></html>",
)
]
) )
tests/unit/test_map.py
+16 -4
View File
@@ -16,7 +16,11 @@ class MockMCP:
return decorator return decorator
def resource(self, path): def resource(self, path):
if not hasattr(self, "resources"):
self.resources = {}
def decorator(func): def decorator(func):
self.resources[path] = func
return func return func
return decorator return decorator
@@ -56,13 +60,21 @@ async def test_get_activity_map_success():
assert result.meta == {"ui": {"resourceUri": "strava://activity/12345/map"}} assert result.meta == {"ui": {"resourceUri": "strava://activity/12345/map"}}
content = result.content content = result.content
assert len(content) == 2 assert len(content) == 1
assert content[0].type == "text" assert content[0].type == "text"
html_text = content[0].text assert "Die interaktive Karte wurde vorbereitet" in content[0].text
assert content[1].type == "text" # Now fetch and test the actual HTML from the registered resource handler
assert "Render the above content" in content[1].text assert "strava://activity/{activity_id}/map" in mcp.resources
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
resource_result = await resource_handler(12345)
assert len(resource_result.contents) == 1
resource_content = resource_result.contents[0]
assert str(resource_content.uri) == "strava://activity/12345/map"
assert resource_content.mimeType == "text/html"
html_text = resource_content.text
assert "<!DOCTYPE html>" in html_text assert "<!DOCTYPE html>" in html_text
assert "unpkg.com/leaflet" in html_text assert "unpkg.com/leaflet" in html_text
assert "latlngData" in html_text assert "latlngData" in html_text