mcp-garmin/server.py

import json
import logging
import os
import queue
import threading
from datetime import date

from garminconnect import Garmin, GarminConnectAuthenticationError
from mcp.server.fastmcp import FastMCP
from rich._log_render import LogRender
from rich.console import Console
from rich.containers import Renderables
from rich.logging import RichHandler
from rich.table import Table
from rich.text import Text

mcp = FastMCP("garmin")


class _PathFirstLogRender(LogRender):
    """Rich's default log layout is time/level/message/path, with path
    stealing fixed width from the message column that's cramped further
    still by Console falling back to an 80-column default (stderr here is
    a pipe to the parent process, not a real terminal, so Rich can't
    detect an actual width). Reordering to time/level/path/message and
    widening the console both help: message gets the whole remaining line
    instead of wrapping every ~30 characters."""

    def __call__(self, console, renderables, log_time=None, time_format=None,
                 level="", path=None, line_no=None, link_path=None):
        output = Table.grid(padding=(0, 1))
        output.expand = True
        if self.show_time:
            output.add_column(style="log.time")
        if self.show_level:
            output.add_column(style="log.level", width=self.level_width)
        if self.show_path and path:
            output.add_column(style="log.path")
        output.add_column(ratio=1, style="log.message", overflow="fold")

        row = []
        if self.show_time:
            log_time = log_time or console.get_datetime()
            time_format = time_format or self.time_format
            log_time_display = (
                time_format(log_time) if callable(time_format) else Text(log_time.strftime(time_format))
            )
            if log_time_display == self._last_time and self.omit_repeated_times:
                row.append(Text(" " * len(log_time_display)))
            else:
                row.append(log_time_display)
                self._last_time = log_time_display
        if self.show_level:
            row.append(level)
        if self.show_path and path:
            path_text = Text()
            path_text.append(path, style=f"link file://{link_path}" if link_path else "")
            if line_no:
                path_text.append(":")
                path_text.append(str(line_no), style=f"link file://{link_path}#{line_no}" if link_path else "")
            row.append(path_text)
        row.append(Renderables(renderables))

        output.add_row(*row)
        return output


def _configure_wide_logging() -> None:
    """Replace the RichHandler FastMCP installed (via configure_logging()
    in its __init__, already run by the time FastMCP("garmin") above
    returns) with one that isn't squeezed into 80 columns."""
    handler = RichHandler(console=Console(stderr=True, width=200), rich_tracebacks=True)
    handler._log_render = _PathFirstLogRender(show_time=True, show_level=True, show_path=True)
    logging.root.handlers = [handler]


_configure_wide_logging()

# Where garth caches the Garmin session (SSO/OAuth tokens), so a restart
# resumes an existing session instead of doing a full login (and MFA
# challenge) every time. GARMIN_TOKENSTORE lets the caller (smartrund)
# override this; garminconnect's own GARMINTOKENS env fallback is not used
# since nothing here sets it.
_TOKENSTORE_PATH = os.environ.get("GARMIN_TOKENSTORE") or os.path.expanduser("~/.garth")

_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(tokenstore=_TOKENSTORE_PATH)
        _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(tokenstore=_TOKENSTORE_PATH)
            _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_activity_splits(activity_id: str) -> str:
    """Get lap/split summaries for a single activity by its ID."""
    if err := _check_auth():
        return err
    try:
        result = _client.get_activity_splits(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 splits: {exc}"


@mcp.tool()
def get_workout_by_id(workout_id: str) -> str:
    """Get a structured workout's step-by-step plan by its ID."""
    if err := _check_auth():
        return err
    try:
        result = _client.get_workout_by_id(workout_id)
        return json.dumps(result, indent=2)
    except GarminConnectAuthenticationError:
        return "Authentication error. Call authenticate() again."
    except Exception as exc:
        return f"Error fetching workout: {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()