sstent/AICycling_mcp
1
0
Fork 0
mirror of https://github.com/sstent/AICycling_mcp.git synced 2026-01-25 08:35:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

added custom MCP

Browse Source
This commit is contained in:
sstent
2025-09-25 07:46:57 -07:00
parent ab7414b788
commit e06683b875
10 changed files with 1699 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

525
custom_garth_mcp.py Normal file
View File

@@ -0,0 +1,525 @@
#!/usr/bin/env python3
"""
Custom Garth MCP Implementation
Direct wrapper around the Garth module with MCP-like interface
"""
import asyncio
import json
import logging
import os
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional, Union
from pathlib import Path
try:
import garth
GARTH_AVAILABLE = True
except ImportError:
GARTH_AVAILABLE = False
garth = None
from cache_manager import CacheManager
logger = logging.getLogger(__name__)
class GarthTool:
"""Represents a single Garth-based tool"""
def __init__(self, name: str, description: str, parameters: Dict[str, Any] = None):
self.name = name
self.description = description
self.parameters = parameters or {}
def __repr__(self):
return f"GarthTool(name='{self.name}')"
class CustomGarthMCP:
"""
Custom MCP-like interface for Garth
Provides tools for accessing Garmin Connect data with local caching
"""
def __init__(self, garth_token: str = None, cache_ttl: int = 300):
self.garth_token = garth_token or os.getenv("GARTH_TOKEN")
self.cache = CacheManager(default_ttl=cache_ttl)
self.session = None
self._tools = []
self._setup_tools()
if not GARTH_AVAILABLE:
logger.error("Garth module not available")
raise ImportError("Garth module not available. Install with: pip install garth")
def _setup_tools(self):
"""Setup available tools based on working endpoints"""
self._tools = [
GarthTool(
name="user_profile",
description="Get user profile information (social profile)",
parameters={}
),
GarthTool(
name="user_settings",
description="Get user statistics/settings information",
parameters={}
),
GarthTool(
name="get_activities",
description="Get list of activities from Garmin Connect",
parameters={
"start_date": {"type": "string", "description": "Start date (YYYY-MM-DD)"},
"limit": {"type": "integer", "description": "Maximum number of activities (default: 20)"}
}
),
GarthTool(
name="get_activities_by_date",
description="Get activities for a specific date",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)", "required": True}
}
),
GarthTool(
name="get_activity_details",
description="Get detailed information for a specific activity",
parameters={
"activity_id": {"type": "string", "description": "Activity ID", "required": True}
}
),
GarthTool(
name="daily_steps",
description="Get daily summary data (may include steps if available)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"},
"days": {"type": "integer", "description": "Number of days"}
}
),
GarthTool(
name="weekly_steps",
description="Get multi-day summary data (weekly equivalent)",
parameters={
"date": {"type": "string", "description": "Start date (YYYY-MM-DD)"},
"weeks": {"type": "integer", "description": "Number of weeks"}
}
),
GarthTool(
name="get_body_composition",
description="Get body composition/weight data (may have parameter requirements)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"}
}
),
GarthTool(
name="snapshot",
description="Get comprehensive data snapshot using working endpoints only",
parameters={
"start_date": {"type": "string", "description": "Start date (YYYY-MM-DD)"},
"end_date": {"type": "string", "description": "End date (YYYY-MM-DD)"}
}
),
# Note: These tools are available but will return helpful error messages
# since the endpoints don't work
GarthTool(
name="daily_sleep",
description="Get daily sleep data (currently unavailable - endpoint not working)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"},
"days": {"type": "integer", "description": "Number of days"}
}
),
GarthTool(
name="daily_stress",
description="Get daily stress data (currently unavailable - endpoint not working)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"},
"days": {"type": "integer", "description": "Number of days"}
}
),
GarthTool(
name="daily_body_battery",
description="Get daily body battery data (currently unavailable - endpoint not working)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"},
"days": {"type": "integer", "description": "Number of days"}
}
),
GarthTool(
name="daily_hrv",
description="Get daily heart rate variability data (currently unavailable - endpoint not working)",
parameters={
"date": {"type": "string", "description": "Date (YYYY-MM-DD)"},
"days": {"type": "integer", "description": "Number of days"}
}
),
GarthTool(
name="get_devices",
description="Get connected devices info (limited - returns user profile as fallback)",
parameters={}
)
]
async def initialize(self):
"""Initialize Garth session"""
if not self.garth_token:
logger.error("No GARTH_TOKEN provided")
raise ValueError("GARTH_TOKEN is required")
try:
# Configure Garth
garth.configure()
# Try to use saved session first
session_path = Path.home() / ".garth"
if session_path.exists():
try:
garth.resume(str(session_path))
# Test the session
await self._test_session()
logger.info("Resumed existing Garth session")
except Exception as e:
logger.warning(f"Could not resume session: {e}")
await self._create_new_session()
else:
await self._create_new_session()
logger.info("Garth session initialized successfully")
except Exception as e:
logger.error(f"Failed to initialize Garth: {e}")
raise
async def _create_new_session(self):
"""Create new Garth session using token"""
try:
# Use the token to login
# Note: You might need to adjust this based on your token format
garth.login_oauth(token=self.garth_token)
# Save session
session_path = Path.home() / ".garth"
garth.save(str(session_path))
except Exception as e:
logger.error(f"Failed to create Garth session: {e}")
raise
async def _test_session(self):
"""Test if current session is valid"""
# Try multiple endpoints to find one that works
test_endpoints = [
"/userprofile-service/socialProfile",
"/user-service/users/settings",
"/modern/currentuser-service/user/profile",
"/userstats-service/statistics"
]
for endpoint in test_endpoints:
try:
garth.connectapi(endpoint)
logger.debug(f"Session test successful with {endpoint}")
return True
except Exception as e:
logger.debug(f"Session test failed for {endpoint}: {e}")
continue
logger.debug("All session tests failed")
raise Exception("No working endpoints found")
async def cleanup(self):
"""Cleanup resources"""
# Garth doesn't require explicit cleanup
pass
def list_tools(self) -> List[GarthTool]:
"""List available tools"""
return self._tools
async def has_tool(self, tool_name: str) -> bool:
"""Check if tool is available"""
return any(tool.name == tool_name for tool in self._tools)
async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
"""Call a specific tool"""
# Check cache first
cache_key = f"{tool_name}:{hash(str(sorted(parameters.items())))}"
cached_result = self.cache.get(cache_key)
if cached_result is not None:
logger.debug(f"Cache hit for {tool_name}")
return cached_result
try:
result = await self._execute_tool(tool_name, parameters)
# Cache result with appropriate TTL
ttl = self._get_cache_ttl(tool_name)
self.cache.set(cache_key, result, ttl=ttl)
logger.debug(f"Tool {tool_name} executed successfully")
return result
except Exception as e:
logger.error(f"Error calling tool {tool_name}: {e}")
raise
def _get_cache_ttl(self, tool_name: str) -> int:
"""Get appropriate cache TTL for different tools"""
ttl_map = {
"user_profile": 3600, # 1 hour - rarely changes
"user_settings": 3600, # 1 hour - rarely changes
"get_devices": 1800, # 30 minutes - occasionally changes
"get_activities": 300, # 5 minutes - changes frequently
"daily_steps": 3600, # 1 hour - daily data
"daily_sleep": 3600, # 1 hour - daily data
"daily_stress": 3600, # 1 hour - daily data
"daily_body_battery": 3600, # 1 hour - daily data
"daily_hrv": 3600, # 1 hour - daily data
"weekly_steps": 1800, # 30 minutes - weekly data
"get_body_composition": 1800, # 30 minutes
"snapshot": 600, # 10 minutes - comprehensive data
}
return ttl_map.get(tool_name, 300) # Default 5 minutes
async def _execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
"""Execute the actual tool call"""
# User profile and settings - use working endpoints
if tool_name == "user_profile":
# Use the working social profile endpoint
return garth.connectapi("/userprofile-service/socialProfile")
elif tool_name == "user_settings":
# Try user stats as fallback since user settings doesn't work
try:
return garth.connectapi("/userstats-service/statistics")
except Exception:
# If stats don't work, return the social profile as user info
return garth.connectapi("/userprofile-service/socialProfile")
# Activities - use working endpoint
elif tool_name == "get_activities":
start_date = parameters.get("start_date")
limit = parameters.get("limit", 20)
params = {"limit": limit}
if start_date:
params["startDate"] = start_date
return garth.connectapi("/activitylist-service/activities/search/activities", params=params)
elif tool_name == "get_activities_by_date":
date = parameters["date"]
start = f"{date}T00:00:00.000Z"
end = f"{date}T23:59:59.999Z"
return garth.connectapi("/activitylist-service/activities/search/activities", params={
"startDate": start,
"endDate": end,
"limit": 100
})
elif tool_name == "get_activity_details":
activity_id = parameters["activity_id"]
return garth.connectapi(f"/activity-service/activity/{activity_id}")
# Daily metrics - many don't work, so provide fallbacks
elif tool_name == "daily_steps":
date = parameters.get("date", datetime.now().strftime("%Y-%m-%d"))
days = parameters.get("days", 1)
# Try usersummary first, if that fails, try to get from activities
try:
return garth.connectapi("/usersummary-service/usersummary/daily", params={
"startDate": date,
"numOfDays": days
})
except Exception as e:
logger.warning(f"Daily steps via usersummary failed: {e}")
# Fallback: get activities for the date and sum steps if available
try:
activities = garth.connectapi("/activitylist-service/activities/search/activities", params={
"startDate": f"{date}T00:00:00.000Z",
"endDate": f"{date}T23:59:59.999Z",
"limit": 50
})
return {"fallback_activities": activities, "date": date, "message": "Daily steps not available, showing activities instead"}
except Exception as e2:
raise Exception(f"Both daily steps and activities failed: {e}, {e2}")
elif tool_name == "daily_sleep":
# Sleep endpoint doesn't work, return error with helpful message
raise Exception("Daily sleep endpoint not available. Sleep data may be accessible through other means.")
elif tool_name == "daily_stress":
# Stress endpoint doesn't work
raise Exception("Daily stress endpoint not available.")
elif tool_name == "daily_body_battery":
# Body battery endpoint doesn't work
raise Exception("Daily body battery endpoint not available.")
elif tool_name == "daily_hrv":
# HRV endpoint doesn't work
raise Exception("HRV endpoint not available.")
# Weekly metrics
elif tool_name == "weekly_steps":
# Weekly endpoint doesn't work, try to get multiple days instead
date = parameters.get("date", datetime.now().strftime("%Y-%m-%d"))
weeks = parameters.get("weeks", 1)
days = weeks * 7
try:
return garth.connectapi("/usersummary-service/usersummary/daily", params={
"startDate": date,
"numOfDays": days
})
except Exception as e:
raise Exception(f"Weekly steps not available: {e}")
# Device info
elif tool_name == "get_devices":
# Device registration doesn't work, return user profile as fallback
profile = garth.connectapi("/userprofile-service/socialProfile")
return {"message": "Device registration endpoint not available", "user_profile": profile}
# Body composition
elif tool_name == "get_body_composition":
# Weight service gives 400 error, likely needs parameters
date = parameters.get("date", datetime.now().strftime("%Y-%m-%d"))
try:
return garth.connectapi("/weight-service/weight/dateRange", params={
"startDate": date,
"endDate": date
})
except Exception as e:
raise Exception(f"Body composition endpoint error: {e}")
# Comprehensive snapshot - use working endpoints only
elif tool_name == "snapshot":
start_date = parameters.get("start_date", datetime.now().strftime("%Y-%m-%d"))
end_date = parameters.get("end_date", start_date)
snapshot = {}
try:
# User profile - we know this works
snapshot["user_profile"] = garth.connectapi("/userprofile-service/socialProfile")
except Exception as e:
logger.warning(f"Could not get user profile: {e}")
try:
# Activities for date range - we know this works
snapshot["activities"] = garth.connectapi("/activitylist-service/activities/search/activities", params={
"startDate": f"{start_date}T00:00:00.000Z",
"endDate": f"{end_date}T23:59:59.999Z",
"limit": 100
})
except Exception as e:
logger.warning(f"Could not get activities: {e}")
try:
# User stats - we know this works
snapshot["user_stats"] = garth.connectapi("/userstats-service/statistics")
except Exception as e:
logger.warning(f"Could not get user stats: {e}")
# Try some daily data (even though many endpoints don't work)
try:
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
end_dt = datetime.strptime(end_date, "%Y-%m-%d")
days = (end_dt - start_dt).days + 1
snapshot["daily_summary"] = garth.connectapi("/usersummary-service/usersummary/daily", params={
"startDate": start_date,
"numOfDays": days
})
except Exception as e:
logger.warning(f"Could not get daily summary: {e}")
if not snapshot:
raise Exception("Could not retrieve any data for snapshot")
return snapshot
else:
raise ValueError(f"Unknown tool: {tool_name}")
def print_tools(self):
"""Pretty print available tools"""
print(f"\n{'='*60}")
print("AVAILABLE GARTH TOOLS")
print(f"{'='*60}")
for i, tool in enumerate(self._tools, 1):
print(f"\n{i}. {tool.name}")
print(f" Description: {tool.description}")
if tool.parameters:
print(" Parameters:")
for param, info in tool.parameters.items():
param_type = info.get("type", "string")
param_desc = info.get("description", "")
required = info.get("required", False)
req_str = " (required)" if required else " (optional)"
print(f" - {param} ({param_type}){req_str}: {param_desc}")
print(f"\n{'='*60}")
@property
def is_available(self) -> bool:
"""Check if MCP server is available"""
return GARTH_AVAILABLE and self.garth_token is not None
def get_cache_stats(self) -> Dict[str, Any]:
"""Get cache statistics"""
return self.cache.get_stats()
def clear_cache(self):
"""Clear all cached data"""
self.cache.clear()
# Utility functions
async def create_garth_mcp(garth_token: str = None, cache_ttl: int = 300) -> CustomGarthMCP:
"""Create and initialize a CustomGarthMCP instance"""
mcp = CustomGarthMCP(garth_token=garth_token, cache_ttl=cache_ttl)
await mcp.initialize()
return mcp
async def test_garth_mcp():
"""Test the custom Garth MCP implementation"""
print("🚀 Testing Custom Garth MCP Implementation")
print("=" * 50)
try:
# Initialize
mcp = await create_garth_mcp()
# List tools
tools = mcp.list_tools()
print(f"✅ Found {len(tools)} tools")
# Test user profile
profile = await mcp.call_tool("user_profile", {})
print("✅ Got user profile")
print(f" User: {profile.get('displayName', 'Unknown')}")
# Test activities
activities = await mcp.call_tool("get_activities", {"limit": 5})
print(f"✅ Got {len(activities)} activities")
# Test cache
cached_profile = await mcp.call_tool("user_profile", {}) # Should hit cache
print("✅ Cache working")
# Show cache stats
cache_stats = mcp.get_cache_stats()
print(f"📊 Cache has {cache_stats['total_entries']} entries")
print("\n🎉 All tests passed!")
except Exception as e:
print(f"❌ Test failed: {e}")
raise
if __name__ == "__main__":
asyncio.run(test_garth_mcp())

169
garth_diagnostics.py Normal file
View File

@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""
Garth API Diagnostics
Test various Garmin Connect API endpoints to see what's available
"""
import json
from pathlib import Path
try:
import garth
print("✅ Garth module available")
except ImportError:
print("❌ Garth module not found")
exit(1)
def test_endpoints():
"""Test various Garmin Connect API endpoints"""
print("🔍 Testing Garmin Connect API Endpoints")
print("=" * 50)
# Load session
session_path = Path.home() / ".garth"
if not session_path.exists():
print("❌ No Garth session found. Run setup_garth.py first.")
return
try:
garth.resume(str(session_path))
print("✅ Session loaded")
except Exception as e:
print(f"❌ Could not load session: {e}")
return
# List of endpoints to test
endpoints_to_test = [
# User/Profile endpoints
("/userprofile-service/socialProfile", "Social Profile"),
("/user-service/user", "User Service"),
("/user-service/users/settings", "User Settings"),
("/modern/currentuser-service/user/profile", "Modern User Profile"),
("/userprofile-service/userprofile", "User Profile Alt"),
# Activity endpoints
("/activitylist-service/activities/search/activities?limit=1", "Recent Activities"),
("/activity-service/activity", "Activity Service"),
# Wellness endpoints
("/wellness-service/wellness", "Wellness Service"),
("/wellness-service/wellness/dailySleep", "Daily Sleep"),
("/wellness-service/wellness/dailyStress", "Daily Stress"),
("/wellness-service/wellness/dailyBodyBattery", "Daily Body Battery"),
# Stats endpoints
("/userstats-service/statistics", "User Statistics"),
("/usersummary-service/usersummary/daily", "Daily Summary"),
("/usersummary-service/usersummary/weekly", "Weekly Summary"),
# Device endpoints
("/device-service/deviceRegistration", "Device Registration"),
("/device-service/deviceService/app-info", "Device App Info"),
# HRV and health
("/hrv-service/hrv", "HRV Service"),
("/weight-service/weight/dateRange", "Weight Service"),
# Other services
("/badge-service/badge", "Badge Service"),
("/golf-service/golf", "Golf Service"),
("/content-service/content", "Content Service"),
]
working_endpoints = []
failed_endpoints = []
for endpoint, name in endpoints_to_test:
try:
print(f"\n📡 Testing: {name}")
print(f" Endpoint: {endpoint}")
# Extract base endpoint and parameters
if "?" in endpoint:
base_endpoint, params_str = endpoint.split("?", 1)
# Parse simple parameters
params = {}
if params_str:
for param in params_str.split("&"):
if "=" in param:
key, value = param.split("=", 1)
params[key] = value
result = garth.connectapi(base_endpoint, params=params)
else:
result = garth.connectapi(endpoint)
if result is not None:
if isinstance(result, dict):
keys_info = f"Dict with {len(result)} keys: {list(result.keys())[:5]}"
elif isinstance(result, list):
keys_info = f"List with {len(result)} items"
else:
keys_info = f"{type(result).__name__}: {str(result)[:50]}"
print(f" ✅ SUCCESS: {keys_info}")
working_endpoints.append((endpoint, name, result))
else:
print(f" ⚠️ SUCCESS but empty response")
working_endpoints.append((endpoint, name, None))
except Exception as e:
error_str = str(e)
if "404" in error_str:
print(f" ❌ NOT FOUND (404)")
elif "403" in error_str:
print(f" 🔒 FORBIDDEN (403) - May need different auth")
elif "401" in error_str:
print(f" 🚫 UNAUTHORIZED (401) - Auth expired?")
elif "500" in error_str:
print(f" 💥 SERVER ERROR (500)")
else:
print(f" ❌ ERROR: {error_str[:100]}")
failed_endpoints.append((endpoint, name, str(e)))
# Summary
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
print(f"\n✅ Working Endpoints ({len(working_endpoints)}):")
for endpoint, name, _ in working_endpoints:
print(f" {name}: {endpoint}")
print(f"\n❌ Failed Endpoints ({len(failed_endpoints)}):")
for endpoint, name, error in failed_endpoints:
short_error = error.split('\n')[0][:50]
print(f" {name}: {short_error}")
# Show sample data from best working endpoint
if working_endpoints:
print(f"\n📋 SAMPLE DATA from first working endpoint:")
endpoint, name, data = working_endpoints[0]
print(f"Endpoint: {name} ({endpoint})")
if data:
if isinstance(data, dict):
# Show first few key-value pairs
for i, (key, value) in enumerate(data.items()):
if i >= 10: # Limit to first 10 items
print(f" ... and {len(data) - 10} more fields")
break
value_str = str(value)[:100]
print(f" {key}: {value_str}")
elif isinstance(data, list) and len(data) > 0:
print(f" List with {len(data)} items")
if isinstance(data[0], dict):
print(f" First item keys: {list(data[0].keys())}")
else:
print(f" First item: {str(data[0])[:100]}")
else:
print(f" Data: {str(data)[:200]}")
print(f"\n💡 Recommendation:")
if working_endpoints:
best_endpoint = working_endpoints[0]
print(f"Use '{best_endpoint[0]}' for user profile data")
else:
print("No working endpoints found. Check authentication.")
if __name__ == "__main__":
test_endpoints()

260
mcp_client.py
View File

@@ -1,47 +1,103 @@
#!/usr/bin/env python3
"""
MCP Client - Handles MCP server connections and tool management
Updated MCP Client - Uses our custom Garth implementation
Fallback to standard MCP if needed, but prefer custom implementation
"""
import os
import shutil
import logging
import asyncio
from typing import List, Dict, Any, Optional
from config import Config
# Try to import both implementations
try:
from custom_garth_mcp import CustomGarthMCP, GarthTool
CUSTOM_GARTH_AVAILABLE = True
except ImportError:
CUSTOM_GARTH_AVAILABLE = False
CustomGarthMCP = None
GarthTool = None
try:
from pydantic_ai.mcp import MCPServerStdio
MCP_AVAILABLE = True
import shutil
import os
STANDARD_MCP_AVAILABLE = True
except ImportError:
MCP_AVAILABLE = False
STANDARD_MCP_AVAILABLE = False
MCPServerStdio = None
from config import Config
logger = logging.getLogger(__name__)
class MCPClient:
"""Manages MCP server connection and tool interactions"""
"""
Enhanced MCP Client that prefers custom Garth implementation
Falls back to standard MCP if needed
"""
def __init__(self, config: Config):
self.config = config
self.mcp_server = None
self.garth_mcp = None
self.standard_mcp = None
self.available_tools = []
self._initialized = False
self._use_custom = True
if not MCP_AVAILABLE:
logger.warning("MCP not available. Tool functionality will be limited.")
return
# Set up MCP server
self._setup_mcp_server()
# Decide which implementation to use
if CUSTOM_GARTH_AVAILABLE and config.garth_token:
logger.info("Using custom Garth MCP implementation")
self._use_custom = True
elif STANDARD_MCP_AVAILABLE and config.garth_token:
logger.info("Falling back to standard MCP implementation")
self._use_custom = False
else:
logger.warning("No MCP implementation available")
def _setup_mcp_server(self):
"""Set up MCP server connection"""
async def initialize(self):
"""Initialize the preferred MCP implementation"""
if not self.config.garth_token:
logger.warning("No GARTH_TOKEN provided. MCP tools will be unavailable.")
return
try:
if self._use_custom and CUSTOM_GARTH_AVAILABLE:
await self._initialize_custom_garth()
elif STANDARD_MCP_AVAILABLE:
await self._initialize_standard_mcp()
else:
logger.error("No MCP implementation available")
return
self._initialized = True
logger.info("MCP client initialized successfully")
except Exception as e:
logger.error(f"MCP initialization failed: {e}")
# Try fallback if custom failed
if self._use_custom and STANDARD_MCP_AVAILABLE:
logger.info("Trying fallback to standard MCP")
try:
self._use_custom = False
await self._initialize_standard_mcp()
self._initialized = True
logger.info("Fallback MCP initialization successful")
except Exception as fallback_error:
logger.error(f"Fallback MCP initialization also failed: {fallback_error}")
async def _initialize_custom_garth(self):
"""Initialize custom Garth MCP"""
self.garth_mcp = CustomGarthMCP(
garth_token=self.config.garth_token,
cache_ttl=self.config.cache_ttl
)
await self.garth_mcp.initialize()
logger.info("Custom Garth MCP initialized")
async def _initialize_standard_mcp(self):
"""Initialize standard MCP (fallback)"""
if not self.config.garth_token:
raise ValueError("GARTH_TOKEN required for standard MCP")
# Set up environment
os.environ["GARTH_TOKEN"] = self.config.garth_token
env = os.environ.copy()
@@ -50,85 +106,116 @@ class MCPClient:
# Find server executable
server_executable = shutil.which(self.config.garth_mcp_server_path)
if not server_executable:
logger.error(f"'{self.config.garth_mcp_server_path}' not found in PATH")
return
raise FileNotFoundError(f"'{self.config.garth_mcp_server_path}' not found in PATH")
self.mcp_server = MCPServerStdio(
self.standard_mcp = MCPServerStdio(
command=server_executable,
args=["garth-mcp-server"],
env=env,
)
async def initialize(self):
"""Initialize MCP server connection"""
if not self.mcp_server:
logger.warning("MCP server not configured")
return
try:
logger.info("Initializing MCP server connection...")
# The MCP server will be initialized when used by the agent
# For now, we'll try to list tools to verify connection
await asyncio.sleep(0.1) # Give it a moment
logger.info("MCP server connection established")
self._initialized = True
except Exception as e:
logger.error(f"MCP server initialization failed: {e}")
self.mcp_server = None
logger.info("Standard MCP initialized")
async def cleanup(self):
"""Cleanup MCP server connection"""
if self.mcp_server:
# MCP server cleanup is handled by the agent
pass
"""Cleanup MCP resources"""
if self.garth_mcp:
await self.garth_mcp.cleanup()
# Standard MCP cleanup is handled by the agent
async def list_tools(self) -> List[Any]:
"""List available MCP tools"""
if not self.mcp_server:
if not self._initialized:
return []
try:
if not self.available_tools:
self.available_tools = await self.mcp_server.list_tools()
return self.available_tools
if self._use_custom and self.garth_mcp:
if not self.available_tools:
self.available_tools = self.garth_mcp.list_tools()
return self.available_tools
elif self.standard_mcp:
if not self.available_tools:
self.available_tools = await self.standard_mcp.list_tools()
return self.available_tools
except Exception as e:
logger.error(f"Error listing tools: {e}")
return []
return []
def list_tools_sync(self) -> List[Any]:
"""Synchronous version for compatibility"""
if self._use_custom and self.garth_mcp:
return self.garth_mcp.list_tools()
return []
async def has_tool(self, tool_name: str) -> bool:
"""Check if a specific tool is available"""
tools = await self.list_tools()
return any(tool.name == tool_name for tool in tools)
async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
"""Call a specific MCP tool directly"""
if not self.mcp_server:
raise RuntimeError("MCP server not available")
if not self._initialized:
return False
try:
result = await self.mcp_server.direct_call_tool(tool_name, parameters)
return result.output if hasattr(result, 'output') else result
if self._use_custom and self.garth_mcp:
return await self.garth_mcp.has_tool(tool_name)
elif self.standard_mcp:
tools = await self.list_tools()
return any(tool.name == tool_name for tool in tools)
except Exception as e:
logger.error(f"Error checking tool {tool_name}: {e}")
return False
return False
async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
"""Call a specific MCP tool"""
if not self._initialized:
raise RuntimeError("MCP client not initialized")
try:
if self._use_custom and self.garth_mcp:
result = await self.garth_mcp.call_tool(tool_name, parameters)
logger.debug(f"Custom MCP tool {tool_name} called successfully")
return result
elif self.standard_mcp:
result = await self.standard_mcp.direct_call_tool(tool_name, parameters)
# Handle different result formats
if hasattr(result, 'output'):
return result.output
elif hasattr(result, 'content'):
return result.content
else:
return result
except Exception as e:
logger.error(f"Error calling tool {tool_name}: {e}")
raise
def get_tool_info(self, tool_name: str) -> Optional[Dict[str, Any]]:
"""Get information about a specific tool"""
for tool in self.available_tools:
tools = self.list_tools_sync() if self._use_custom else []
for tool in tools:
if tool.name == tool_name:
return {
"name": tool.name,
"description": getattr(tool, 'description', ''),
"parameters": getattr(tool, 'inputSchema', {}).get('properties', {})
}
if self._use_custom and isinstance(tool, GarthTool):
return {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
else:
# Standard MCP tool
return {
"name": tool.name,
"description": getattr(tool, 'description', ''),
"parameters": getattr(tool, 'inputSchema', {}).get('properties', {})
}
return None
def print_tools(self):
"""Pretty print available tools"""
if not self.available_tools:
if self._use_custom and self.garth_mcp:
self.garth_mcp.print_tools()
return
tools = self.list_tools_sync()
if not tools:
print("No MCP tools available")
return
@@ -136,7 +223,7 @@ class MCPClient:
print("AVAILABLE MCP TOOLS")
print(f"{'='*60}")
for i, tool in enumerate(self.available_tools, 1):
for i, tool in enumerate(tools, 1):
print(f"\n{i}. {tool.name}")
if hasattr(tool, 'description') and tool.description:
print(f" Description: {tool.description}")
@@ -156,5 +243,42 @@ class MCPClient:
@property
def is_available(self) -> bool:
"""Check if MCP server is available"""
return self.mcp_server is not None and self._initialized
"""Check if MCP is available and initialized"""
return self._initialized and (
(self._use_custom and self.garth_mcp is not None) or
(not self._use_custom and self.standard_mcp is not None)
)
@property
def implementation_type(self) -> str:
"""Get the type of MCP implementation being used"""
if self._use_custom:
return "custom_garth"
else:
return "standard_mcp"
def get_cache_stats(self) -> Dict[str, Any]:
"""Get cache statistics (only available with custom implementation)"""
if self._use_custom and self.garth_mcp:
return self.garth_mcp.get_cache_stats()
else:
return {"message": "Cache stats only available with custom implementation"}
def clear_cache(self):
"""Clear cache (only available with custom implementation)"""
if self._use_custom and self.garth_mcp:
self.garth_mcp.clear_cache()
logger.info("Cache cleared")
else:
logger.warning("Cache clearing only available with custom implementation")
# Compatibility methods for existing code
@property
def mcp_server(self):
"""Compatibility property for existing code using agent integration"""
if self._use_custom:
# For custom implementation, we can't provide direct agent integration
# Return None to indicate tools should be called directly
return None
else:
return self.standard_mcp

167
minimal_mcp_test (1).py
View File

@@ -1,167 +0,0 @@
#!/usr/bin/env python3
"""
Minimal MCP Test - Just test MCP connection and user profile
"""
import asyncio
import json
import logging
import os
# Minimal imports - just what we need
try:
from pydantic_ai.mcp import MCPServerStdio
import shutil
MCP_AVAILABLE = True
except ImportError:
print("❌ pydantic-ai MCP not available")
print("Install with: pip install pydantic-ai")
exit(1)
# Simple logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MinimalMCPTest:
"""Minimal MCP test class"""
def __init__(self, garth_token: str, server_path: str = "uvx"):
self.garth_token = garth_token
self.server_path = server_path
self.mcp_server = None
self.cached_profile = None
def setup_mcp_server(self):
"""Setup MCP server connection"""
# Set environment
os.environ["GARTH_TOKEN"] = self.garth_token
env = os.environ.copy()
# Find server executable
server_executable = shutil.which(self.server_path)
if not server_executable:
raise FileNotFoundError(f"'{self.server_path}' not found in PATH")
self.mcp_server = MCPServerStdio(
command=server_executable,
args=["garth-mcp-server"],
env=env,
)
print("✅ MCP server configured")
async def test_connection(self):
"""Test basic MCP connection"""
if not self.mcp_server:
raise RuntimeError("MCP server not configured")
try:
# Try to list tools
tools = await self.mcp_server.list_tools()
print(f"✅ MCP connected - found {len(tools)} tools")
# Show tools
for tool in tools:
print(f" 📋 {tool.name}: {getattr(tool, 'description', 'No description')}")
return tools
except Exception as e:
print(f"❌ MCP connection failed: {e}")
raise
async def get_user_profile(self):
"""Get and cache user profile"""
try:
print("📞 Calling user_profile tool...")
# Direct tool call
result = await self.mcp_server.direct_call_tool("user_profile", {})
profile_data = result.output if hasattr(result, 'output') else result
# Cache it
self.cached_profile = profile_data
print("✅ User profile retrieved and cached")
return profile_data
except Exception as e:
print(f"❌ Failed to get user profile: {e}")
raise
def print_profile(self):
"""Print cached profile"""
if not self.cached_profile:
print("❌ No cached profile")
return
print("\n" + "="*50)
print("USER PROFILE")
print("="*50)
print(json.dumps(self.cached_profile, indent=2, default=str))
print("="*50)
async def run_test(self):
"""Run the complete test"""
print("🚀 Starting Minimal MCP Test\n")
# Setup
self.setup_mcp_server()
# Test connection
tools = await self.test_connection()
# Check if user_profile tool exists
user_profile_tool = next((t for t in tools if t.name == "user_profile"), None)
if not user_profile_tool:
print("❌ user_profile tool not found")
return False
# Get user profile
await self.get_user_profile()
# Show results
self.print_profile()
print("\n🎉 Test completed successfully!")
return True
def get_config():
"""Get configuration from environment or user input"""
garth_token = os.getenv("GARTH_TOKEN")
if not garth_token:
print("GARTH_TOKEN not found in environment")
print("Please run 'uvx garth login' first to authenticate")
garth_token = input("Enter your GARTH_TOKEN: ").strip()
if not garth_token:
raise ValueError("GARTH_TOKEN is required")
server_path = os.getenv("GARTH_MCP_SERVER_PATH", "uvx")
return garth_token, server_path
async def main():
"""Main entry point"""
try:
# Get config
garth_token, server_path = get_config()
# Run test
test = MinimalMCPTest(garth_token, server_path)
success = await test.run_test()
if success:
print("\n✅ All tests passed!")
else:
print("\n❌ Tests failed!")
except KeyboardInterrupt:
print("\n👋 Interrupted by user")
except Exception as e:
print(f"\n💥 Error: {e}")
logger.error("Test error", exc_info=True)
if __name__ == "__main__":
asyncio.run(main())

0
minimal_mcp_test (2).py → minimal_mcp_test.py
View File

20
requirements.txt Normal file
View File

@@ -0,0 +1,20 @@
# Core dependencies
pydantic-ai>=0.0.1
pyyaml>=6.0
aiohttp>=3.8.0
# LLM dependencies
pydantic>=2.0.0
openai>=1.0.0
# Garmin Connect data access
garth>=0.4.0
# Optional MCP dependencies (fallback)
# Install garth-mcp-server separately if needed:
# npm install -g garth-mcp-server
# or check: https://github.com/matin/garth-mcp-server
# Development and testing
pytest>=7.0.0
pytest-asyncio>=0.21.0

13
requirements_file.txt
View File

@@ -1,13 +0,0 @@
# Core dependencies
pydantic-ai>=0.0.1
pyyaml>=6.0
aiohttp>=3.8.0
# LLM dependencies
pydantic>=2.0.0
openai>=1.0.0
# Optional MCP dependencies
# Install garth-mcp-server separately:
# npm install -g garth-mcp-server
# or check: https://github.com/matin/garth-mcp-server

285
setup_garth.py Normal file
View File

@@ -0,0 +1,285 @@
#!/usr/bin/env python3
"""
Garth Setup Script
Helper script to set up Garth authentication and test connection
"""
import os
import json
import sys
from pathlib import Path
try:
import garth
print("✅ Garth module available")
except ImportError:
print("❌ Garth module not found")
print("Install with: pip install garth")
sys.exit(1)
def setup_garth_auth():
"""Setup Garth authentication"""
print("🔐 Setting up Garth authentication...")
print("This will guide you through authenticating with Garmin Connect")
print()
# Check if session already exists
session_path = Path.home() / ".garth"
if session_path.exists():
print("📁 Existing Garth session found")
choice = input("Do you want to use existing session? (y/n): ").strip().lower()
if choice == 'y':
try:
garth.resume(str(session_path))
# Test session
test_result = test_garth_connection()
if test_result:
print("✅ Existing session is valid")
return True
else:
print("❌ Existing session is invalid, creating new one...")
except Exception as e:
print(f"⚠️ Could not resume session: {e}")
print("Creating new session...")
# Create new session
print("\n🆕 Creating new Garth session...")
print("You'll need your Garmin Connect credentials")
print()
try:
# Configure Garth
garth.configure()
# Get credentials
username = input("Garmin Connect username/email: ").strip()
password = input("Garmin Connect password: ").strip()
if not username or not password:
print("❌ Username and password are required")
return False
print("\n🔄 Logging into Garmin Connect...")
# Login
garth.login(username, password)
# Save session
garth.save(str(session_path))
print(f"💾 Session saved to {session_path}")
# Test connection
if test_garth_connection():
print("✅ Authentication successful!")
return True
else:
print("❌ Authentication failed")
return False
except Exception as e:
print(f"❌ Authentication error: {e}")
return False
def test_garth_connection():
"""Test Garth connection"""
print("🧪 Testing Garth connection...")
# Use the endpoint we know works
try:
print(" Trying social profile endpoint...")
user_info = garth.connectapi("/userprofile-service/socialProfile")
if user_info and isinstance(user_info, dict):
display_name = user_info.get('displayName', 'Unknown')
full_name = user_info.get('fullName', 'Unknown')
print(f"✅ Connected as: {display_name} ({full_name})")
# Test activities too
try:
activities = garth.connectapi("/activitylist-service/activities/search/activities", params={"limit": 1})
if activities and len(activities) > 0:
print(f"✅ Activities accessible: Found {len(activities)} recent activity")
else:
print(f"⚠️ Activities accessible but none found")
except Exception as e:
print(f"⚠️ Activities test failed: {str(e)[:50]}")
return True
else:
print("❌ Invalid response from social profile endpoint")
return False
except Exception as e:
print(f"❌ Connection test failed: {e}")
return False
def extract_token_info():
"""Extract token information for config"""
print("\n🎫 Extracting token information...")
session_path = Path.home() / ".garth"
if not session_path.exists():
print("❌ No Garth session found. Run authentication first.")
return None
try:
# Load session data
with open(session_path, 'r') as f:
session_data = json.load(f)
# Extract token (this might need adjustment based on Garth's session format)
token = session_data.get('oauth_token') or session_data.get('token')
if token:
print(f"✅ Token extracted: {token[:20]}...")
print("\n📝 Add this to your config.yaml:")
print(f"garth_token: \"{token}\"")
return token
else:
print("❌ Could not extract token from session")
print("Available session keys:", list(session_data.keys()))
return None
except Exception as e:
print(f"❌ Error extracting token: {e}")
return None
def show_user_info():
"""Show current user information"""
print("\n👤 User Information:")
# Try multiple endpoints to get user info
endpoints_to_try = [
("/userprofile-service/socialProfile", "Social Profile"),
("/user-service/users/settings", "User Settings"),
("/modern/currentuser-service/user/profile", "User Profile"),
("/userprofile-service/userprofile", "User Profile Alt"),
]
user_info = None
working_endpoint = None
for endpoint, name in endpoints_to_try:
try:
print(f" Trying {name}...")
data = garth.connectapi(endpoint)
if data and isinstance(data, dict):
user_info = data
working_endpoint = endpoint
print(f" ✅ {name} successful")
break
except Exception as e:
print(f" ❌ {name} failed: {str(e)[:100]}")
continue
if user_info:
print(f"\n Working Endpoint: {working_endpoint}")
print(f" Data keys available: {list(user_info.keys())}")
# Extract common fields
display_name = (
user_info.get('displayName') or
user_info.get('fullName') or
user_info.get('userName') or
user_info.get('profileDisplayName') or
'Unknown'
)
user_id = (
user_info.get('id') or
user_info.get('userId') or
user_info.get('profileId') or
'Unknown'
)
print(f" Display Name: {display_name}")
print(f" User ID: {user_id}")
# Show first few fields for debugging
print(f" Sample data:")
for key, value in list(user_info.items())[:5]:
print(f" {key}: {str(value)[:50]}")
# Try to get activities
try:
print(f"\n Testing activities access...")
activities = garth.connectapi("/activitylist-service/activities/search/activities",
params={"limit": 1})
if activities and len(activities) > 0:
print(f" Recent Activities: ✅ Available ({len(activities)} found)")
activity = activities[0]
activity_type = activity.get('activityType', {})
if isinstance(activity_type, dict):
type_name = activity_type.get('typeKey', 'Unknown')
else:
type_name = str(activity_type)
print(f" Last Activity: {type_name}")
else:
print(f" Recent Activities: ⚠️ None found")
except Exception as e:
print(f" Recent Activities: ❌ Error ({str(e)[:50]})")
else:
print(" ❌ Could not retrieve any user information")
# Show what endpoints are available
print("\n Available endpoints check:")
test_endpoints = [
"/userstats-service/statistics",
"/wellness-service/wellness",
"/device-service/deviceRegistration",
]
for endpoint in test_endpoints:
try:
garth.connectapi(endpoint)
print(f" ✅ {endpoint} - Available")
except Exception as e:
if "404" in str(e):
print(f" ❌ {endpoint} - Not Found")
elif "403" in str(e):
print(f" ⚠️ {endpoint} - Forbidden (auth issue)")
else:
print(f" ❌ {endpoint} - Error: {str(e)[:30]}")
def main():
"""Main setup function"""
print("Garth Setup and Authentication")
print("=" * 40)
while True:
print("\nOptions:")
print("1. Setup/Login to Garmin Connect")
print("2. Test existing connection")
print("3. Show user information")
print("4. Extract token for config")
print("5. Exit")
choice = input("\nEnter choice (1-5): ").strip()
if choice == "1":
success = setup_garth_auth()
if success:
show_user_info()
elif choice == "2":
test_garth_connection()
elif choice == "3":
show_user_info()
elif choice == "4":
token = extract_token_info()
if token:
# Also show how to set environment variable
print(f"\nOr set as environment variable:")
print(f"export GARTH_TOKEN=\"{token}\"")
elif choice == "5":
print("👋 Goodbye!")
break
else:
print("Invalid choice. Please try again.")
if __name__ == "__main__":
main()

63
simple_garth_test.py Normal file
View File

@@ -0,0 +1,63 @@
#!/usr/bin/env python3
"""
Simple Garth Test - Minimal test to verify Garth is working
"""
import garth
from pathlib import Path
def simple_test():
print("🔧 Simple Garth Connection Test")
print("=" * 40)
# Check if session exists
session_path = Path.home() / ".garth"
if session_path.exists():
print("✅ Session file exists")
try:
garth.resume(str(session_path))
print("✅ Session loaded")
except Exception as e:
print(f"❌ Session load failed: {e}")
return False
else:
print("❌ No session file found")
return False
# Try the simplest possible API call
simple_endpoints = [
"/",
"/ping",
"/health",
"/status"
]
print("\n🧪 Testing simple endpoints...")
for endpoint in simple_endpoints:
try:
result = garth.connectapi(endpoint)
print(f"✅ {endpoint} worked: {type(result)}")
return True
except Exception as e:
print(f"❌ {endpoint} failed: {str(e)[:50]}")
# Try to access the raw client
print("\n🔍 Checking Garth client details...")
try:
client = garth.client
print(f"Client type: {type(client)}")
print(f"Client attributes: {[attr for attr in dir(client) if not attr.startswith('_')][:5]}")
# Try to see if we can get base URL
if hasattr(client, 'domain'):
print(f"Domain: {client.domain}")
if hasattr(client, 'base_url'):
print(f"Base URL: {client.base_url}")
except Exception as e:
print(f"❌ Client inspection failed: {e}")
return False
if __name__ == "__main__":
simple_test()

445
test_custom_mcp.py Normal file
View File

@@ -0,0 +1,445 @@
#!/usr/bin/env python3
"""
Test Custom MCP Implementation
Test our custom Garth MCP wrapper
"""
import asyncio
import json
import logging
import sys
from pathlib import Path
# Import our modules
from config import Config, load_config, create_sample_config
from mcp_client import MCPClient # Updated MCP client
from cache_manager import CacheManager
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class CustomMCPTestApp:
"""Test application for custom MCP implementation"""
def __init__(self, config: Config):
self.config = config
self.mcp_client = MCPClient(config) # Will use custom implementation
async def initialize(self):
"""Initialize MCP client"""
logger.info("Initializing Custom MCP test app...")
await self.mcp_client.initialize()
async def cleanup(self):
"""Cleanup resources"""
await self.mcp_client.cleanup()
async def test_connection_and_tools(self):
"""Test basic connection and list tools"""
print("\n" + "="*60)
print("CUSTOM MCP CONNECTION TEST")
print("="*60)
if not self.mcp_client.is_available:
print("❌ Custom MCP not available")
return False
print("✅ Custom MCP connected")
print(f"📋 Implementation: {self.mcp_client.implementation_type}")
# List available tools
tools = await self.mcp_client.list_tools()
print(f"📋 Found {len(tools)} tools")
if tools:
print("\nAvailable tools:")
for i, tool in enumerate(tools[:10], 1): # Show first 10
print(f" {i:2d}. {tool.name}")
if hasattr(tool, 'description'):
print(f" {tool.description}")
else:
print(" No tools available")
return len(tools) > 0
async def test_user_profile(self):
"""Test user profile retrieval"""
print("\n" + "="*60)
print("USER PROFILE TEST")
print("="*60)
# Check if tool exists
if not await self.mcp_client.has_tool("user_profile"):
print("❌ user_profile tool not available")
return None
print("✅ user_profile tool found")
try:
# Call user_profile tool
print("📞 Calling user_profile tool...")
profile_data = await self.mcp_client.call_tool("user_profile", {})
print("✅ User profile retrieved")
# Show profile summary
if isinstance(profile_data, dict):
display_name = profile_data.get('displayName', 'Unknown')
full_name = profile_data.get('fullName', 'Unknown')
user_name = profile_data.get('userName', 'Unknown')
print(f" Display Name: {display_name}")
print(f" Full Name: {full_name}")
print(f" Username: {user_name}")
print(f" Profile contains {len(profile_data)} fields")
return profile_data
except Exception as e:
print(f"❌ Error getting user profile: {e}")
logger.error(f"User profile error: {e}", exc_info=True)
return None
async def test_activities(self):
"""Test activities retrieval"""
print("\n" + "="*60)
print("ACTIVITIES TEST")
print("="*60)
if not await self.mcp_client.has_tool("get_activities"):
print("❌ get_activities tool not available")
return None
print("✅ get_activities tool found")
try:
print("📞 Calling get_activities (limit=5)...")
activities = await self.mcp_client.call_tool("get_activities", {"limit": 5})
if activities and len(activities) > 0:
print(f"✅ Retrieved {len(activities)} activities")
# Show activity summaries
print("\nRecent Activities:")
for i, activity in enumerate(activities[:3], 1):
activity_id = activity.get('activityId', 'Unknown')
activity_type = activity.get('activityType', {})
type_key = activity_type.get('typeKey', 'Unknown') if isinstance(activity_type, dict) else str(activity_type)
start_time = activity.get('startTimeLocal', 'Unknown time')
print(f" {i}. {type_key} (ID: {activity_id})")
print(f" Start: {start_time}")
# Try to get duration
duration = activity.get('duration')
if duration:
minutes = duration // 60
seconds = duration % 60
print(f" Duration: {minutes}m {seconds}s")
return activities
else:
print("📋 No activities found")
return []
except Exception as e:
print(f"❌ Error getting activities: {e}")
logger.error(f"Activities error: {e}", exc_info=True)
return None
async def test_activity_details(self, activities):
"""Test activity details retrieval"""
print("\n" + "="*60)
print("ACTIVITY DETAILS TEST")
print("="*60)
if not activities or len(activities) == 0:
print("❌ No activities available for details test")
return None
if not await self.mcp_client.has_tool("get_activity_details"):
print("❌ get_activity_details tool not available")
return None
# Get details for first activity
first_activity = activities[0]
activity_id = str(first_activity.get('activityId'))
print(f"📞 Getting details for activity {activity_id}...")
try:
details = await self.mcp_client.call_tool("get_activity_details", {
"activity_id": activity_id
})
print("✅ Activity details retrieved")
if isinstance(details, dict):
print(f" Details contain {len(details)} fields")
# Show some key details
activity_name = details.get('activityName', 'Unnamed')
sport = details.get('sportTypeId', 'Unknown sport')
distance = details.get('distance')
elapsed_duration = details.get('elapsedDuration')
print(f" Activity: {activity_name}")
print(f" Sport: {sport}")
if distance:
print(f" Distance: {distance/1000:.2f} km")
if elapsed_duration:
minutes = elapsed_duration // 60
seconds = elapsed_duration % 60
print(f" Duration: {minutes}m {seconds}s")
return details
except Exception as e:
print(f"❌ Error getting activity details: {e}")
logger.error(f"Activity details error: {e}", exc_info=True)
return None
async def test_daily_metrics(self):
"""Test daily metrics retrieval"""
print("\n" + "="*60)
print("DAILY METRICS TEST")
print("="*60)
metrics_to_test = ["daily_steps", "daily_sleep", "daily_stress"]
results = {}
for metric in metrics_to_test:
if await self.mcp_client.has_tool(metric):
try:
print(f"📞 Testing {metric}...")
data = await self.mcp_client.call_tool(metric, {"days": 1})
if data:
print(f"✅ {metric} data retrieved")
if isinstance(data, list) and len(data) > 0:
print(f" Contains {len(data)} day(s) of data")
results[metric] = data
else:
print(f"⚠️ {metric} returned no data")
except Exception as e:
print(f"❌ Error with {metric}: {e}")
logger.debug(f"{metric} error: {e}")
else:
print(f"❌ {metric} tool not available")
return results
async def test_cache_functionality(self):
"""Test cache functionality"""
print("\n" + "="*60)
print("CACHE FUNCTIONALITY TEST")
print("="*60)
if self.mcp_client.implementation_type != "custom_garth":
print("❌ Cache testing only available with custom implementation")
return False
try:
# Call user_profile twice to test caching
print("📞 First user_profile call (should hit API)...")
profile1 = await self.mcp_client.call_tool("user_profile", {})
print("📞 Second user_profile call (should hit cache)...")
profile2 = await self.mcp_client.call_tool("user_profile", {})
# Verify same data
if profile1 == profile2:
print("✅ Cache working correctly")
else:
print("⚠️ Cache data differs from API data")
# Show cache stats
cache_stats = self.mcp_client.get_cache_stats()
print(f"📊 Cache stats: {cache_stats}")
return True
except Exception as e:
print(f"❌ Cache test failed: {e}")
logger.error(f"Cache test error: {e}", exc_info=True)
return False
async def test_comprehensive_snapshot(self):
"""Test comprehensive data snapshot"""
print("\n" + "="*60)
print("COMPREHENSIVE SNAPSHOT TEST")
print("="*60)
if not await self.mcp_client.has_tool("snapshot"):
print("❌ snapshot tool not available")
return None
try:
print("📞 Getting comprehensive data snapshot...")
snapshot = await self.mcp_client.call_tool("snapshot", {
"start_date": "2024-09-20",
"end_date": "2024-09-24"
})
if isinstance(snapshot, dict):
print("✅ Snapshot retrieved successfully")
print(f" Snapshot contains {len(snapshot)} data categories:")
for category, data in snapshot.items():
if isinstance(data, list):
print(f" - {category}: {len(data)} items")
elif isinstance(data, dict):
print(f" - {category}: {len(data)} fields")
else:
print(f" - {category}: {type(data).__name__}")
return snapshot
except Exception as e:
print(f"❌ Snapshot test failed: {e}")
logger.error(f"Snapshot error: {e}", exc_info=True)
return None
async def run_all_tests(self):
"""Run comprehensive test suite"""
print("🚀 Starting Custom MCP Test Suite")
results = {}
# Test 1: Connection and tools
results['connection'] = await self.test_connection_and_tools()
if not results['connection']:
print("\n❌ Connection failed - skipping remaining tests")
return results
# Test 2: User profile
profile = await self.test_user_profile()
results['user_profile'] = profile is not None
# Test 3: Activities
activities = await self.test_activities()
results['activities'] = activities is not None
# Test 4: Activity details (if we have activities)
if activities and len(activities) > 0:
details = await self.test_activity_details(activities)
results['activity_details'] = details is not None
# Test 5: Daily metrics
metrics = await self.test_daily_metrics()
results['daily_metrics'] = len(metrics) > 0
# Test 6: Cache functionality
results['cache'] = await self.test_cache_functionality()
# Test 7: Comprehensive snapshot
snapshot = await self.test_comprehensive_snapshot()
results['snapshot'] = snapshot is not None
# Summary
self._print_test_summary(results)
return results
def _print_test_summary(self, results):
"""Print test summary"""
print("\n" + "="*60)
print("TEST SUMMARY")
print("="*60)
passed = 0
total = 0
for test_name, result in results.items():
total += 1
if result:
passed += 1
status = "✅ PASS"
else:
status = "❌ FAIL"
print(f"{test_name.replace('_', ' ').title():.<40} {status}")
print("-" * 60)
print(f"Total: {passed}/{total} tests passed")
if self.mcp_client.implementation_type == "custom_garth":
try:
cache_stats = self.mcp_client.get_cache_stats()
print(f"\nCache Status: {cache_stats.get('total_entries', 0)} entries")
for key in cache_stats.get('keys', []):
print(f" - {key}")
except Exception as e:
logger.debug(f"Could not get cache stats: {e}")
def validate_config(config: Config) -> bool:
"""Validate configuration"""
issues = []
if not config.garth_token:
issues.append("GARTH_TOKEN not set")
if issues:
print("❌ Configuration issues:")
for issue in issues:
print(f" - {issue}")
print("\nTo fix:")
print("1. Run 'pip install garth' to install Garth module")
print("2. Run authentication to get token")
print("3. Update config.yaml with your token")
return False
return True
async def main():
"""Main entry point"""
print("Custom MCP Test App - Test Custom Garth Implementation")
print("=" * 60)
try:
# Setup config
create_sample_config()
config = load_config()
if not validate_config(config):
return
# Create and run test app
app = CustomMCPTestApp(config)
try:
await app.initialize()
results = await app.run_all_tests()
# Exit with appropriate code
passed_tests = sum(1 for result in results.values() if result)
total_tests = len(results)
if passed_tests == total_tests:
print(f"\n🎉 All {total_tests} tests passed!")
sys.exit(0)
elif passed_tests > 0:
print(f"\n⚠️ {passed_tests}/{total_tests} tests passed")
sys.exit(1)
else:
print(f"\n❌ All {total_tests} tests failed")
sys.exit(1)
finally:
await app.cleanup()
except KeyboardInterrupt:
print("\n👋 Test interrupted")
except Exception as e:
print(f"\n💥 Test error: {e}")
logger.error(f"Main error: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())