feat: server skeleton with authenticate and complete_mfa tools
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
100
server.py
Normal file
100
server.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user