From 170d7ce37d5483895f42914b87e598ced3c1a884 Mon Sep 17 00:00:00 2001 From: Christophe Vila Date: Sun, 17 May 2026 19:40:59 +0200 Subject: [PATCH] feat: server skeleton with authenticate and complete_mfa tools Co-Authored-By: Claude Sonnet 4.6 --- server.py | 100 +++++++++++++++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_server.py | 66 ++++++++++++++++++++++++++++ 3 files changed, 166 insertions(+) create mode 100644 server.py create mode 100644 tests/__init__.py create mode 100644 tests/test_server.py diff --git a/server.py b/server.py new file mode 100644 index 0000000..4614c54 --- /dev/null +++ b/server.py @@ -0,0 +1,100 @@ +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.get("GARMIN_EMAIL", ""), os.environ.get("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() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..3d06de0 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,66 @@ +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."