325 lines
10 KiB
Python
325 lines
10 KiB
Python
import json
|
|
import os
|
|
import queue
|
|
import threading
|
|
from datetime import date
|
|
|
|
from garminconnect import Garmin, GarminConnectAuthenticationError
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("garmin")
|
|
|
|
_client: Garmin | None = None
|
|
_auth_state: str = "unauthenticated"
|
|
_mfa_input_queue: queue.Queue = queue.Queue()
|
|
_login_result_queue: queue.Queue = queue.Queue()
|
|
|
|
|
|
def _today() -> str:
|
|
return date.today().isoformat()
|
|
|
|
|
|
def _check_auth() -> str | None:
|
|
if _auth_state != "authenticated":
|
|
return "Not authenticated. Ask Claude to call authenticate() first."
|
|
return None
|
|
|
|
|
|
def _prompt_mfa() -> str:
|
|
return _mfa_input_queue.get(timeout=300)
|
|
|
|
|
|
def _startup_login() -> None:
|
|
global _client, _auth_state
|
|
email = os.environ.get("GARMIN_EMAIL")
|
|
password = os.environ.get("GARMIN_PASSWORD")
|
|
if not email or not password:
|
|
raise SystemExit(
|
|
"GARMIN_EMAIL and GARMIN_PASSWORD environment variables are required."
|
|
)
|
|
try:
|
|
_client = Garmin(email, password)
|
|
_client.login()
|
|
_auth_state = "authenticated"
|
|
except Exception:
|
|
_auth_state = "unauthenticated"
|
|
|
|
|
|
@mcp.tool()
|
|
def authenticate() -> str:
|
|
"""Initiate Garmin authentication using credentials from environment variables."""
|
|
global _client, _auth_state
|
|
|
|
email = os.environ.get("GARMIN_EMAIL", "")
|
|
password = os.environ.get("GARMIN_PASSWORD", "")
|
|
if not email or not password:
|
|
return "GARMIN_EMAIL and GARMIN_PASSWORD environment variables are required."
|
|
|
|
def _do_login() -> None:
|
|
try:
|
|
_client.login()
|
|
_login_result_queue.put(("success", None))
|
|
except Exception as exc:
|
|
_login_result_queue.put(("error", str(exc)))
|
|
|
|
_client = Garmin(email, password)
|
|
_client.prompt_mfa = _prompt_mfa
|
|
threading.Thread(target=_do_login, daemon=True).start()
|
|
|
|
try:
|
|
status, err = _login_result_queue.get(timeout=10)
|
|
if status == "success":
|
|
_auth_state = "authenticated"
|
|
return "Authenticated successfully."
|
|
return f"Authentication failed: {err}"
|
|
except queue.Empty:
|
|
_auth_state = "mfa_pending"
|
|
return (
|
|
"MFA required. Garmin has sent a verification code to your registered "
|
|
"email or phone. Call complete_mfa(code) with the code you received."
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def complete_mfa(code: str) -> str:
|
|
"""Provide the MFA verification code to complete Garmin authentication."""
|
|
global _auth_state
|
|
|
|
if _auth_state != "mfa_pending":
|
|
return "No MFA in progress. Call authenticate() first."
|
|
|
|
_mfa_input_queue.put(code)
|
|
|
|
try:
|
|
status, err = _login_result_queue.get(timeout=30)
|
|
if status == "success":
|
|
_auth_state = "authenticated"
|
|
return "MFA accepted. Authenticated successfully."
|
|
return f"Authentication failed after MFA: {err}"
|
|
except queue.Empty:
|
|
_auth_state = "unauthenticated"
|
|
return "Timed out waiting for authentication to complete. Call authenticate() again."
|
|
|
|
|
|
@mcp.tool()
|
|
def get_activities(start_date: str = "", end_date: str = "", limit: int = 20) -> str:
|
|
"""List Garmin activities between two dates (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
start = start_date or _today()
|
|
end = end_date or _today()
|
|
try:
|
|
result = _client.get_activities_by_date(start, end)
|
|
return json.dumps(result[:limit], indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching activities: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_activity_details(activity_id: str) -> str:
|
|
"""Get full details for a single activity by its ID (splits, laps, metrics)."""
|
|
if err := _check_auth():
|
|
return err
|
|
try:
|
|
result = _client.get_activity_details(activity_id)
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching activity details: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_sleep(date: str = "") -> str:
|
|
"""Get sleep data for a date (YYYY-MM-DD): score, stages, duration. Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_sleep_data(cdate)
|
|
if not result:
|
|
return f"No sleep data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching sleep data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_heart_rate(date: str = "") -> str:
|
|
"""Get heart rate data for a date (YYYY-MM-DD): resting HR and daily timeline. Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_heart_rates(cdate)
|
|
if not result:
|
|
return f"No heart rate data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching heart rate data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_stress(date: str = "") -> str:
|
|
"""Get stress level timeline for a date (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_stress_data(cdate)
|
|
if not result:
|
|
return f"No stress data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching stress data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_body_battery(date: str = "") -> str:
|
|
"""Get body battery charge/drain values for a date (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_body_battery(cdate, cdate)
|
|
if not result:
|
|
return f"No body battery data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching body battery data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_hrv(date: str = "") -> str:
|
|
"""Get HRV status and nightly average for a date (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_hrv_data(cdate)
|
|
if not result:
|
|
return f"No HRV data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching HRV data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_spo2(date: str = "") -> str:
|
|
"""Get SpO2 (blood oxygen) readings for a date (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_spo2_data(cdate)
|
|
if not result:
|
|
return f"No SpO2 data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching SpO2 data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_steps(date: str = "") -> str:
|
|
"""Get step count and daily goal for a date (YYYY-MM-DD). Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_steps_data(cdate)
|
|
if not result:
|
|
return f"No steps data found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching steps data: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_daily_stats(date: str = "") -> str:
|
|
"""Get daily summary for a date (YYYY-MM-DD): calories, floors, intensity minutes. Defaults to today."""
|
|
if err := _check_auth():
|
|
return err
|
|
cdate = date or _today()
|
|
try:
|
|
result = _client.get_stats(cdate)
|
|
if not result:
|
|
return f"No daily stats found for {cdate}"
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching daily stats: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_devices() -> str:
|
|
"""List paired Garmin devices with model name and firmware version."""
|
|
if err := _check_auth():
|
|
return err
|
|
try:
|
|
result = _client.get_devices()
|
|
if not result:
|
|
return "No devices found."
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching devices: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_gear() -> str:
|
|
"""List gear items (shoes, bikes) with mileage and usage stats."""
|
|
if err := _check_auth():
|
|
return err
|
|
try:
|
|
profile = _client.get_user_profile()
|
|
user_id = profile.get("userId") or profile.get("id")
|
|
if not user_id:
|
|
return "Could not determine user profile ID required for gear lookup."
|
|
result = _client.get_gear(str(user_id))
|
|
if not result:
|
|
return "No gear found."
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching gear: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_user_profile() -> str:
|
|
"""Get Garmin user profile: display name, age, weight, and HR zones."""
|
|
if err := _check_auth():
|
|
return err
|
|
try:
|
|
result = _client.get_user_profile()
|
|
if not result:
|
|
return "No user profile found."
|
|
return json.dumps(result, indent=2)
|
|
except GarminConnectAuthenticationError:
|
|
return "Authentication error. Call authenticate() again."
|
|
except Exception as exc:
|
|
return f"Error fetching user profile: {exc}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_startup_login()
|
|
mcp.run()
|