1006 lines
30 KiB
Markdown
1006 lines
30 KiB
Markdown
# MCP Garmin Server — Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Build a single-file FastMCP stdio server exposing 15 Garmin Connect tools to Claude Desktop, with interactive MFA handling from within chat.
|
|
|
|
**Architecture:** `server.py` holds all tools and a module-level `Garmin` client shared across calls. Auth state is a global string flag. MFA is handled by running login in a background thread that blocks on a `queue.Queue`; `complete_mfa()` unblocks it. Startup attempts a silent login using cached `~/.garth` tokens, falling back gracefully if none exist.
|
|
|
|
**Tech Stack:** Python 3.14, `mcp[cli]` (FastMCP), `garminconnect`, `pytest`, `unittest.mock`
|
|
|
|
---
|
|
|
|
## File Map
|
|
|
|
| File | Purpose |
|
|
|------|---------|
|
|
| `pyproject.toml` | Project metadata and pip dependencies |
|
|
| `.env.example` | Template for required env vars (committed) |
|
|
| `.gitignore` | Excludes `.env`, `.garth/`, `.venv/` |
|
|
| `server.py` | All 15 MCP tools + auth state + startup logic |
|
|
| `tests/__init__.py` | Empty — makes tests/ a package |
|
|
| `tests/test_server.py` | Unit tests with mocked Garmin client |
|
|
|
|
---
|
|
|
|
### Task 1: Project Scaffolding
|
|
|
|
**Files:**
|
|
- Create: `pyproject.toml`
|
|
- Create: `.env.example`
|
|
- Create: `.gitignore`
|
|
|
|
- [ ] **Step 1: Create `pyproject.toml`**
|
|
|
|
```toml
|
|
[project]
|
|
name = "mcp-garmin"
|
|
version = "0.1.0"
|
|
description = "MCP server for Garmin Connect"
|
|
requires-python = ">=3.11"
|
|
dependencies = [
|
|
"mcp[cli]",
|
|
"garminconnect",
|
|
]
|
|
|
|
[build-system]
|
|
requires = ["hatchling"]
|
|
build-backend = "hatchling.build"
|
|
```
|
|
|
|
- [ ] **Step 2: Create `.env.example`**
|
|
|
|
```
|
|
GARMIN_EMAIL=your@email.com
|
|
GARMIN_PASSWORD=yourpassword
|
|
```
|
|
|
|
- [ ] **Step 3: Create `.gitignore`**
|
|
|
|
```
|
|
.env
|
|
.garth/
|
|
.venv/
|
|
__pycache__/
|
|
*.pyc
|
|
.idea/
|
|
dist/
|
|
```
|
|
|
|
- [ ] **Step 4: Install dependencies**
|
|
|
|
```bash
|
|
.venv/bin/pip install "mcp[cli]" garminconnect pytest
|
|
```
|
|
|
|
Expected: All packages install without errors. Verify with:
|
|
```bash
|
|
.venv/bin/python -c "from garminconnect import Garmin; from mcp.server.fastmcp import FastMCP; print('OK')"
|
|
```
|
|
Expected output: `OK`
|
|
|
|
- [ ] **Step 5: Initialize git and commit**
|
|
|
|
```bash
|
|
git init
|
|
git add pyproject.toml .env.example .gitignore docs/
|
|
git commit -m "chore: project scaffolding and design docs"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: Server Skeleton + Auth Tools
|
|
|
|
**Files:**
|
|
- Create: `server.py`
|
|
- Create: `tests/__init__.py`
|
|
- Create: `tests/test_server.py`
|
|
|
|
- [ ] **Step 1: Write failing tests for auth tools**
|
|
|
|
Create `tests/__init__.py` as an empty file, then create `tests/test_server.py`:
|
|
|
|
```python
|
|
import json
|
|
import queue
|
|
import threading
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import server
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_state():
|
|
original_state = server._auth_state
|
|
original_client = server._client
|
|
for q in (server._mfa_input_queue, server._login_result_queue):
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
yield
|
|
server._auth_state = original_state
|
|
server._client = original_client
|
|
|
|
|
|
def test_check_auth_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
assert server._check_auth() == "Not authenticated. Ask Claude to call authenticate() first."
|
|
|
|
|
|
def test_check_auth_authenticated():
|
|
server._auth_state = "authenticated"
|
|
assert server._check_auth() is None
|
|
|
|
|
|
def test_authenticate_success():
|
|
with patch("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
# mock login() returns immediately, thread puts success in queue
|
|
result = server.authenticate()
|
|
assert result == "Authenticated successfully."
|
|
assert server._auth_state == "authenticated"
|
|
|
|
|
|
def test_authenticate_mfa_required():
|
|
with patch("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
with patch.object(server._login_result_queue, "get", side_effect=queue.Empty):
|
|
result = server.authenticate()
|
|
assert "MFA required" in result
|
|
assert server._auth_state == "mfa_pending"
|
|
|
|
|
|
def test_complete_mfa_success():
|
|
server._auth_state = "mfa_pending"
|
|
server._login_result_queue.put(("success", None))
|
|
result = server.complete_mfa("123456")
|
|
assert result == "MFA accepted. Authenticated successfully."
|
|
assert server._auth_state == "authenticated"
|
|
assert server._mfa_input_queue.get_nowait() == "123456"
|
|
|
|
|
|
def test_complete_mfa_not_in_progress():
|
|
server._auth_state = "unauthenticated"
|
|
result = server.complete_mfa("123456")
|
|
assert result == "No MFA in progress. Call authenticate() first."
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v 2>&1 | head -20
|
|
```
|
|
|
|
Expected: `ModuleNotFoundError: No module named 'server'` — server.py doesn't exist yet.
|
|
|
|
- [ ] **Step 3: Create `server.py`**
|
|
|
|
```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
|
|
|
|
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(os.environ["GARMIN_EMAIL"], os.environ["GARMIN_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:
|
|
return "Timed out waiting for authentication to complete."
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_startup_login()
|
|
mcp.run()
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests to verify they pass**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 6 tests pass. Example output:
|
|
```
|
|
test_server.py::test_check_auth_unauthenticated PASSED
|
|
test_server.py::test_check_auth_authenticated PASSED
|
|
test_server.py::test_authenticate_success PASSED
|
|
test_server.py::test_authenticate_mfa_required PASSED
|
|
test_server.py::test_complete_mfa_success PASSED
|
|
test_server.py::test_complete_mfa_not_in_progress PASSED
|
|
```
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add server.py tests/__init__.py tests/test_server.py
|
|
git commit -m "feat: server skeleton with authenticate and complete_mfa tools"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: Activity Tools
|
|
|
|
**Files:**
|
|
- Modify: `server.py` (add `get_activities`, `get_activity_details`)
|
|
- Modify: `tests/test_server.py` (add activity tests)
|
|
|
|
- [ ] **Step 1: Write failing tests for activity tools**
|
|
|
|
Add the following imports to the top import block of `tests/test_server.py` (after `import pytest`):
|
|
|
|
```python
|
|
from datetime import date
|
|
from garminconnect import GarminConnectAuthenticationError
|
|
```
|
|
|
|
Then append to the bottom of `tests/test_server.py`:
|
|
|
|
```python
|
|
def test_get_activities_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
assert "Not authenticated" in server.get_activities()
|
|
|
|
|
|
def test_get_activities_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = [
|
|
{"activityId": "111", "activityType": {"typeKey": "running"}, "distance": 5000.0},
|
|
{"activityId": "222", "activityType": {"typeKey": "cycling"}, "distance": 20000.0},
|
|
]
|
|
result = server.get_activities("2026-05-01", "2026-05-17", 20)
|
|
data = json.loads(result)
|
|
assert len(data) == 2
|
|
assert data[0]["activityId"] == "111"
|
|
server._client.get_activities_by_date.assert_called_once_with("2026-05-01", "2026-05-17")
|
|
|
|
|
|
def test_get_activities_limit():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = [{"activityId": str(i)} for i in range(10)]
|
|
data = json.loads(server.get_activities("2026-05-01", "2026-05-17", 3))
|
|
assert len(data) == 3
|
|
|
|
|
|
def test_get_activities_defaults_to_today():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = []
|
|
server.get_activities()
|
|
today = date.today().isoformat()
|
|
server._client.get_activities_by_date.assert_called_once_with(today, today)
|
|
|
|
|
|
def test_get_activities_auth_error():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.side_effect = GarminConnectAuthenticationError("auth")
|
|
assert "Authentication error" in server.get_activities()
|
|
|
|
|
|
def test_get_activity_details_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activity_details.return_value = {"activityId": "111", "laps": []}
|
|
data = json.loads(server.get_activity_details("111"))
|
|
assert data["activityId"] == "111"
|
|
server._client.get_activity_details.assert_called_once_with("111")
|
|
|
|
|
|
def test_get_activity_details_error():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activity_details.side_effect = Exception("not found")
|
|
assert "Error fetching activity details" in server.get_activity_details("999")
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify new ones fail**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py::test_get_activities_success -v
|
|
```
|
|
|
|
Expected: `AttributeError: module 'server' has no attribute 'get_activities'`
|
|
|
|
- [ ] **Step 3: Add activity tools to `server.py`**
|
|
|
|
Add the following before the `if __name__ == "__main__":` block:
|
|
|
|
```python
|
|
@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}"
|
|
```
|
|
|
|
- [ ] **Step 4: Run all tests to verify they pass**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 13 tests pass.
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add server.py tests/test_server.py
|
|
git commit -m "feat: add get_activities and get_activity_details tools"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 4: Health Stats Tools
|
|
|
|
**Files:**
|
|
- Modify: `server.py` (add 6 health tools)
|
|
- Modify: `tests/test_server.py` (add health tests)
|
|
|
|
- [ ] **Step 1: Write failing tests for health tools**
|
|
|
|
Append to the bottom of `tests/test_server.py`:
|
|
|
|
```python
|
|
def test_get_sleep_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_sleep_data.return_value = {
|
|
"dailySleepDTO": {"sleepScores": {"overall": {"value": 85}}}
|
|
}
|
|
data = json.loads(server.get_sleep("2026-05-17"))
|
|
assert data["dailySleepDTO"]["sleepScores"]["overall"]["value"] == 85
|
|
server._client.get_sleep_data.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_sleep_no_data():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_sleep_data.return_value = None
|
|
assert server.get_sleep("2026-05-17") == "No sleep data found for 2026-05-17"
|
|
|
|
|
|
def test_get_heart_rate_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_heart_rates.return_value = {"restingHeartRate": 52}
|
|
data = json.loads(server.get_heart_rate("2026-05-17"))
|
|
assert data["restingHeartRate"] == 52
|
|
server._client.get_heart_rates.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_stress_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_stress_data.return_value = {"stressValuesArray": [[1000, 35]]}
|
|
data = json.loads(server.get_stress("2026-05-17"))
|
|
assert "stressValuesArray" in data
|
|
|
|
|
|
def test_get_body_battery_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_body_battery.return_value = [{"bodyBatteryStatList": []}]
|
|
data = json.loads(server.get_body_battery("2026-05-17"))
|
|
assert isinstance(data, list)
|
|
server._client.get_body_battery.assert_called_once_with("2026-05-17", "2026-05-17")
|
|
|
|
|
|
def test_get_hrv_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_hrv_data.return_value = {"hrvSummary": {"weeklyAvg": 45}}
|
|
data = json.loads(server.get_hrv("2026-05-17"))
|
|
assert data["hrvSummary"]["weeklyAvg"] == 45
|
|
|
|
|
|
def test_get_spo2_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_spo2_data.return_value = {"spO2HourlyAverages": []}
|
|
data = json.loads(server.get_spo2("2026-05-17"))
|
|
assert "spO2HourlyAverages" in data
|
|
|
|
|
|
def test_health_tools_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
for fn in (server.get_sleep, server.get_heart_rate, server.get_stress,
|
|
server.get_body_battery, server.get_hrv, server.get_spo2):
|
|
assert "Not authenticated" in fn("2026-05-17")
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify new ones fail**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py::test_get_sleep_success -v
|
|
```
|
|
|
|
Expected: `AttributeError: module 'server' has no attribute 'get_sleep'`
|
|
|
|
- [ ] **Step 3: Add health tools to `server.py`**
|
|
|
|
Add the following before the `if __name__ == "__main__":` block:
|
|
|
|
```python
|
|
@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}"
|
|
```
|
|
|
|
- [ ] **Step 4: Run all tests to verify they pass**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 21 tests pass.
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add server.py tests/test_server.py
|
|
git commit -m "feat: add health stats tools (sleep, HR, stress, body battery, HRV, SpO2)"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 5: Steps & Daily Metrics Tools
|
|
|
|
**Files:**
|
|
- Modify: `server.py` (add `get_steps`, `get_daily_stats`)
|
|
- Modify: `tests/test_server.py` (add steps/metrics tests)
|
|
|
|
- [ ] **Step 1: Write failing tests**
|
|
|
|
Append to the bottom of `tests/test_server.py`:
|
|
|
|
```python
|
|
def test_get_steps_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_steps_data.return_value = [{"steps": 8342, "primaryActivityLevel": "active"}]
|
|
data = json.loads(server.get_steps("2026-05-17"))
|
|
assert data[0]["steps"] == 8342
|
|
server._client.get_steps_data.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_steps_no_data():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_steps_data.return_value = []
|
|
assert server.get_steps("2026-05-17") == "No steps data found for 2026-05-17"
|
|
|
|
|
|
def test_get_daily_stats_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_stats.return_value = {
|
|
"totalSteps": 9231,
|
|
"totalKilocalories": 2100.0,
|
|
"floorsAscended": 5,
|
|
}
|
|
data = json.loads(server.get_daily_stats("2026-05-17"))
|
|
assert data["totalSteps"] == 9231
|
|
server._client.get_stats.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_steps_and_stats_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
for fn in (server.get_steps, server.get_daily_stats):
|
|
assert "Not authenticated" in fn("2026-05-17")
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify new ones fail**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py::test_get_steps_success -v
|
|
```
|
|
|
|
Expected: `AttributeError: module 'server' has no attribute 'get_steps'`
|
|
|
|
- [ ] **Step 3: Add steps and daily stats tools to `server.py`**
|
|
|
|
Add the following before the `if __name__ == "__main__":` block:
|
|
|
|
```python
|
|
@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}"
|
|
```
|
|
|
|
- [ ] **Step 4: Run all tests to verify they pass**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 25 tests pass.
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add server.py tests/test_server.py
|
|
git commit -m "feat: add get_steps and get_daily_stats tools"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 6: Gear, Devices & Profile Tools
|
|
|
|
**Files:**
|
|
- Modify: `server.py` (add `get_devices`, `get_gear`, `get_user_profile`)
|
|
- Modify: `tests/test_server.py` (add gear/device/profile tests)
|
|
|
|
- [ ] **Step 1: Write failing tests**
|
|
|
|
Append to the bottom of `tests/test_server.py`:
|
|
|
|
```python
|
|
def test_get_devices_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_devices.return_value = [
|
|
{"deviceId": "abc123", "productDisplayName": "Forerunner 965", "softwareVersion": "21.20"}
|
|
]
|
|
data = json.loads(server.get_devices())
|
|
assert data[0]["productDisplayName"] == "Forerunner 965"
|
|
server._client.get_devices.assert_called_once()
|
|
|
|
|
|
def test_get_devices_empty():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_devices.return_value = []
|
|
assert server.get_devices() == "No devices found."
|
|
|
|
|
|
def test_get_gear_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.display_name = "john.doe"
|
|
server._client.get_gear.return_value = [
|
|
{"gearPk": "shoe1", "customMakeModel": "Nike Pegasus", "totalDistance": 320000.0}
|
|
]
|
|
data = json.loads(server.get_gear())
|
|
assert data[0]["customMakeModel"] == "Nike Pegasus"
|
|
server._client.get_gear.assert_called_once_with("john.doe")
|
|
|
|
|
|
def test_get_user_profile_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_user_profile.return_value = {
|
|
"displayName": "john.doe",
|
|
"age": 35,
|
|
"weight": 75000.0,
|
|
}
|
|
data = json.loads(server.get_user_profile())
|
|
assert data["displayName"] == "john.doe"
|
|
server._client.get_user_profile.assert_called_once()
|
|
|
|
|
|
def test_gear_devices_profile_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
for fn in (server.get_devices, server.get_gear, server.get_user_profile):
|
|
assert "Not authenticated" in fn()
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify new ones fail**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py::test_get_devices_success -v
|
|
```
|
|
|
|
Expected: `AttributeError: module 'server' has no attribute 'get_devices'`
|
|
|
|
- [ ] **Step 3: Add gear, devices, and profile tools to `server.py`**
|
|
|
|
Add the following before the `if __name__ == "__main__":` block:
|
|
|
|
```python
|
|
@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:
|
|
result = _client.get_gear(_client.display_name)
|
|
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}"
|
|
```
|
|
|
|
- [ ] **Step 4: Run all tests to verify they pass**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 30 tests pass.
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add server.py tests/test_server.py
|
|
git commit -m "feat: add get_devices, get_gear, and get_user_profile tools"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 7: README & Claude Desktop Config
|
|
|
|
**Files:**
|
|
- Create: `README.md`
|
|
|
|
- [ ] **Step 1: Create `README.md`**
|
|
|
|
```markdown
|
|
# mcp-garmin
|
|
|
|
MCP server exposing Garmin Connect data to Claude Desktop.
|
|
|
|
## Setup
|
|
|
|
### 1. Install dependencies
|
|
|
|
```bash
|
|
python -m venv .venv
|
|
.venv/bin/pip install "mcp[cli]" garminconnect
|
|
```
|
|
|
|
### 2. Configure Claude Desktop
|
|
|
|
Edit `~/Library/Application Support/Claude/claude_desktop_config.json`:
|
|
|
|
```json
|
|
{
|
|
"mcpServers": {
|
|
"garmin": {
|
|
"command": "/absolute/path/to/mcp-garmin/.venv/bin/python",
|
|
"args": ["/absolute/path/to/mcp-garmin/server.py"],
|
|
"env": {
|
|
"GARMIN_EMAIL": "your@email.com",
|
|
"GARMIN_PASSWORD": "yourpassword"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
Replace `/absolute/path/to/mcp-garmin` with the actual path to this directory.
|
|
|
|
### 3. Restart Claude Desktop
|
|
|
|
Quit and relaunch Claude Desktop to pick up the new server config.
|
|
|
|
### 4. First-time authentication
|
|
|
|
On first use (no cached session in `~/.garth`), ask Claude:
|
|
|
|
> "Call authenticate()"
|
|
|
|
If Garmin requires MFA, a prompt will appear in the chat. Check your email or phone for the code, then:
|
|
|
|
> "Call complete_mfa with code 123456"
|
|
|
|
Once authenticated, the session is cached in `~/.garth`. Future server restarts will log in silently without MFA.
|
|
|
|
## Available Tools
|
|
|
|
| Tool | Parameters | Description |
|
|
|------|-----------|-------------|
|
|
| `authenticate()` | — | Initiate Garmin login |
|
|
| `complete_mfa(code)` | `code: str` | Provide MFA code to finish login |
|
|
| `get_activities(start_date, end_date, limit)` | all optional | List recent activities |
|
|
| `get_activity_details(activity_id)` | `activity_id: str` | Full detail for one activity |
|
|
| `get_sleep(date)` | optional | Sleep score and stages |
|
|
| `get_heart_rate(date)` | optional | Resting HR and daily timeline |
|
|
| `get_stress(date)` | optional | Stress level timeline |
|
|
| `get_body_battery(date)` | optional | Body battery charge/drain |
|
|
| `get_hrv(date)` | optional | HRV status and nightly average |
|
|
| `get_spo2(date)` | optional | Blood oxygen readings |
|
|
| `get_steps(date)` | optional | Step count and daily goal |
|
|
| `get_daily_stats(date)` | optional | Calories, floors, intensity minutes |
|
|
| `get_devices()` | — | Paired Garmin devices |
|
|
| `get_gear()` | — | Gear items with mileage stats |
|
|
| `get_user_profile()` | — | Profile, age, weight, HR zones |
|
|
|
|
All `date` parameters accept `YYYY-MM-DD` format and default to today when omitted.
|
|
|
|
## Running Tests
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/ -v
|
|
```
|
|
```
|
|
|
|
- [ ] **Step 2: Run the full test suite one final time**
|
|
|
|
```bash
|
|
.venv/bin/pytest tests/test_server.py -v
|
|
```
|
|
|
|
Expected: All 30 tests pass. No warnings or errors.
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add README.md
|
|
git commit -m "docs: add README with setup and Claude Desktop config instructions"
|
|
```
|