Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4d127541c5 | |||
| 04663d7ba1 | |||
| 3164e91be9 | |||
| ef3b8305ab |
@@ -8,7 +8,6 @@ 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
|
||||
from mcp.types import TextContent, EmbeddedResource
|
||||
|
||||
|
||||
class MockContext(Context):
|
||||
@@ -60,6 +59,7 @@ async def generate():
|
||||
class MockMCP:
|
||||
def __init__(self):
|
||||
self.tools = {}
|
||||
self.resources = {}
|
||||
|
||||
def tool(self):
|
||||
def decorator(func):
|
||||
@@ -68,6 +68,13 @@ async def generate():
|
||||
|
||||
return decorator
|
||||
|
||||
def resource(self, path):
|
||||
def decorator(func):
|
||||
self.resources[path] = func
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
mcp = MockMCP()
|
||||
|
||||
# Mock StravaClient that returns our mock activity and streams
|
||||
@@ -106,34 +113,10 @@ async def generate():
|
||||
client = MockClient()
|
||||
register(mcp, client)
|
||||
|
||||
get_activity_map = mcp.tools["get_activity_map"]
|
||||
ctx = MockContext()
|
||||
|
||||
print("Generating mock map HTML...")
|
||||
result = await get_activity_map(ctx, 99999)
|
||||
|
||||
html_text = None
|
||||
for item in result:
|
||||
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(""", '"')
|
||||
break
|
||||
elif (
|
||||
isinstance(item, EmbeddedResource)
|
||||
and item.resource.mimeType == "text/html"
|
||||
):
|
||||
html_text = item.resource.text
|
||||
break
|
||||
print("Generating mock map HTML from resource...")
|
||||
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
|
||||
resource_result = await resource_handler(99999)
|
||||
html_text = resource_result.contents[0].text
|
||||
|
||||
if html_text:
|
||||
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
|
||||
@@ -172,6 +155,7 @@ async def generate():
|
||||
class MockMCP:
|
||||
def __init__(self):
|
||||
self.tools = {}
|
||||
self.resources = {}
|
||||
|
||||
def tool(self):
|
||||
def decorator(func):
|
||||
@@ -180,38 +164,20 @@ async def generate():
|
||||
|
||||
return decorator
|
||||
|
||||
def resource(self, path):
|
||||
def decorator(func):
|
||||
self.resources[path] = func
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
mcp = MockMCP()
|
||||
register(mcp, client)
|
||||
|
||||
get_activity_map = mcp.tools["get_activity_map"]
|
||||
ctx = MockContext()
|
||||
|
||||
print("Generating map HTML...")
|
||||
result = await get_activity_map(ctx, activity_id)
|
||||
|
||||
# Find the TextContent that holds the HTML
|
||||
html_text = None
|
||||
for item in result:
|
||||
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(""", '"')
|
||||
break
|
||||
elif (
|
||||
isinstance(item, EmbeddedResource)
|
||||
and item.resource.mimeType == "text/html"
|
||||
):
|
||||
html_text = item.resource.text
|
||||
break
|
||||
print("Generating map HTML from resource...")
|
||||
resource_handler = mcp.resources["strava://activity/{activity_id}/map"]
|
||||
resource_result = await resource_handler(activity_id)
|
||||
html_text = resource_result.contents[0].text
|
||||
|
||||
if html_text:
|
||||
output_path = os.path.join(os.path.dirname(__file__), "..", "test.html")
|
||||
@@ -221,11 +187,6 @@ async def generate():
|
||||
print(
|
||||
"👉 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:
|
||||
print(f"❌ Failed: {str(e)}")
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""Strava MCP Server"""
|
||||
|
||||
__version__ = "0.2.3"
|
||||
__version__ = "0.2.7"
|
||||
|
||||
@@ -1,6 +1,13 @@
|
||||
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,
|
||||
ReadResourceResult,
|
||||
)
|
||||
from strava_mcp_server.strava_client import StravaClient
|
||||
from strava_mcp_server.utils import (
|
||||
parse_iso_to_unix,
|
||||
@@ -377,15 +384,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
|
||||
except Exception as e:
|
||||
return f"Error fetching activity streams: {str(e)}"
|
||||
|
||||
@mcp.tool()
|
||||
async def get_activity_map(ctx: Context, activity_id: int):
|
||||
"""
|
||||
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 self-contained HTML page loading Leaflet.js and OpenStreetMap to render the route inside Open WebUI.
|
||||
"""
|
||||
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:
|
||||
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_map = activity.get("map", {})
|
||||
@@ -395,8 +402,9 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
|
||||
|
||||
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 [TextContent(type="text", text=warning_msg)]
|
||||
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")
|
||||
@@ -445,9 +453,10 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
|
||||
elif s.get("type") == "latlng":
|
||||
latlng_data = s.get("data", [])
|
||||
except Exception as e:
|
||||
await ctx.warning(
|
||||
f"Could not fetch streams for elevation profile: {str(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
|
||||
@@ -844,16 +853,101 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
|
||||
</script>
|
||||
</body>
|
||||
</html>"""
|
||||
escaped_html = html_content.strip().replace('"', """)
|
||||
iframe_content = f'<iframe srcdoc="{escaped_html}" style="width: 100%; height: 600px; border: none; border-radius: 12px;"></iframe>'
|
||||
return [
|
||||
TextContent(type="text", text=iframe_content),
|
||||
TextContent(
|
||||
type="text",
|
||||
text="[System Context: The interactive map has been generated successfully. You must output the exact <iframe> HTML block from the tool result verbatim in your chat response so that Open WebUI can render it inline. Do not modify the iframe, and do not explain that you cannot render it; just output the tag verbatim.]",
|
||||
),
|
||||
]
|
||||
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"
|
||||
return CallToolResult(
|
||||
content=[
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Die interaktive Karte wurde vorbereitet. Du findest sie unter der URI: {resource_uri}",
|
||||
),
|
||||
],
|
||||
_meta={"ui": {"resourceUri": resource_uri}},
|
||||
)
|
||||
except Exception as e:
|
||||
error_msg = f"Error preparing activity map: {str(e)}"
|
||||
await ctx.error(error_msg)
|
||||
return [TextContent(type="text", text=error_msg)]
|
||||
return CallToolResult(
|
||||
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>",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
+40
-20
@@ -15,6 +15,16 @@ class MockMCP:
|
||||
|
||||
return decorator
|
||||
|
||||
def resource(self, path):
|
||||
if not hasattr(self, "resources"):
|
||||
self.resources = {}
|
||||
|
||||
def decorator(func):
|
||||
self.resources[path] = func
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_activity_map_success():
|
||||
@@ -46,25 +56,34 @@ async def test_get_activity_map_success():
|
||||
get_activity_map = mcp.tools["get_activity_map"]
|
||||
result = await get_activity_map(ctx, 12345)
|
||||
|
||||
assert len(result) == 2
|
||||
assert result[0].type == "text"
|
||||
iframe_text = result[0].text
|
||||
assert iframe_text.startswith('<iframe srcdoc="')
|
||||
assert iframe_text.endswith('"></iframe>')
|
||||
assert result.isError is not True
|
||||
assert result.meta == {"ui": {"resourceUri": "strava://activity/12345/map"}}
|
||||
|
||||
assert result[1].type == "text"
|
||||
assert "System Context" in result[1].text
|
||||
content = result.content
|
||||
assert len(content) == 1
|
||||
assert content[0].type == "text"
|
||||
assert "Die interaktive Karte wurde vorbereitet" in content[0].text
|
||||
|
||||
decoded_html = iframe_text.replace(""", '"')
|
||||
assert "<!DOCTYPE html>" in decoded_html
|
||||
assert "unpkg.com/leaflet" in decoded_html
|
||||
assert "latlngData" in decoded_html
|
||||
assert "[47.1, 9.1]" in decoded_html
|
||||
assert 'sportEmoji = "🚴"' in decoded_html
|
||||
assert "L.control.layers" in decoded_html
|
||||
assert "satellite" in decoded_html
|
||||
assert "🟢" in decoded_html
|
||||
assert "🏁" in decoded_html
|
||||
# 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"]
|
||||
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 "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
|
||||
@@ -89,6 +108,7 @@ async def test_get_activity_map_no_polyline():
|
||||
get_activity_map = mcp.tools["get_activity_map"]
|
||||
result = await get_activity_map(ctx, 67890)
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].type == "text"
|
||||
assert "⚠️ Keine GPS-Routen-Daten" in result[0].text
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user