How to Build a Health Tracking AI Agent with Wearable Data Integration
Purpose
Health tracking is a mess. My Garmin watch tracks workouts. My phone counts steps. I manually log water intake in one app, meals in another, and notes about how I felt during training in yet another. There’s no unified view, no AI insights, no way to correlate sleep quality with running performance.
I needed a system that could pull data automatically from my wearable, let me log manual entries quickly, and use AI to generate actionable insights. This post shows how to build exactly that using OpenClaw, Garmin Connect, Telegram, and LangChain.
The Problem
Traditional health apps are siloed. They don’t talk to each other. My Garmin data stays in Garmin Connect. My water tracking stays in a hydration app. My training notes are scattered across notes apps.
For athletes training for events like marathons, this fragmentation is a real problem. At 55, training for my third marathon, I need all the help I can get to optimize recovery, track hydration, and spot overtraining patterns before they become injuries.
What I needed was:
- Automated data pulls from Garmin Connect (workouts, heart rate, sleep, steps)
- Quick manual logging for things wearables can’t track (water, meals, how I feel)
- An AI layer that aggregates everything and gives me insights
Environment
I built this system using:
- Python 3.11 - Core language
- OpenClaw - Data aggregation and orchestration hub
- garminconnect library - Python client for Garmin Connect API
- python-telegram-bot - For the manual input interface
- LangChain - AI analysis layer
- PostgreSQL - Time-series health data storage
Solution
Step 1: Set Up OpenClaw as Your Data Hub
OpenClaw serves as the central orchestration platform. Install it first:
pip install openclaw
# Or clone from source for more controlgit clone https://github.com/openclaw/openclaw.gitcd openclawpip install -e .Configure OpenClaw to act as your health data aggregator:
from claw import Agent, Tool, Scheduleimport os
# Create health tracking agenthealth_agent = Agent( name="health_tracker", model="gpt-4", tools=[], # We'll add tools below schedule=Schedule.cron("0 * * * *") # Run hourly)
# Configure data sourceshealth_agent.add_data_source( name="garmin", type="wearable", credentials={ "email": os.environ.get("GARMIN_EMAIL"), "password": os.environ.get("GARMIN_PASSWORD") })
health_agent.add_data_source( name="manual_logs", type="telegram", credentials={ "bot_token": os.environ.get("TELEGRAM_BOT_TOKEN") })Step 2: Configure Garmin Connect Integration
Set up authentication with Garmin Connect to pull workout data, heart rate, sleep metrics, and step counts:
from garminconnect import Garminfrom datetime import datetime, dateimport osimport json
class GarminHealthClient: def __init__(self): self.email = os.environ.get("GARMIN_EMAIL") self.password = os.environ.get("GARMIN_PASSWORD") self.client = None
def connect(self): """Authenticate with Garmin Connect.""" self.client = Garmin(self.email, self.password) self.client.login() return self.client
def fetch_daily_metrics(self, target_date: date = None): """Fetch all daily health metrics.""" if not self.client: self.connect()
if target_date is None: target_date = date.today()
date_str = target_date.strftime("%Y-%m-%d")
try: metrics = { "date": date_str, "fetched_at": datetime.now().isoformat(), "activities": self.client.get_activities(date_str), "steps": self.client.get_steps_data(date_str), "sleep": self.client.get_sleep_data(date_str), "heart_rate": self.client.get_heart_rates(date_str), "stress": self.client.get_stress_data(date_str), "body_battery": self.client.get_body_battery(date_str) } return metrics except Exception as e: print(f"Error fetching Garmin data: {e}") return None
def fetch_activities(self, limit: int = 10): """Fetch recent activities.""" if not self.client: self.connect()
activities = self.client.get_activities(0, limit) return activities
# Usageif __name__ == "__main__": garmin = GarminHealthClient() garmin.connect()
today_metrics = garmin.fetch_daily_metrics() print(json.dumps(today_metrics, indent=2))Handle API rate limits with proper caching:
import jsonfrom datetime import datetime, timedeltafrom pathlib import Pathfrom garminconnect import Garminimport os
class CachedGarminClient: def __init__(self, cache_dir: str = "./health_cache"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) self.client = None
def connect(self): self.client = Garmin( os.environ["GARMIN_EMAIL"], os.environ["GARMIN_PASSWORD"] ) self.client.login()
def get_cached_data(self, key: str, max_age_minutes: int = 60): """Get data from cache if still valid.""" cache_file = self.cache_dir / f"{key}.json"
if cache_file.exists(): with open(cache_file) as f: cached = json.load(f)
cached_time = datetime.fromisoformat(cached["fetched_at"]) if datetime.now() - cached_time < timedelta(minutes=max_age_minutes): return cached["data"]
return None
def save_to_cache(self, key: str, data): """Save data to cache.""" cache_file = self.cache_dir / f"{key}.json" with open(cache_file, "w") as f: json.dump({ "fetched_at": datetime.now().isoformat(), "data": data }, f)
def fetch_with_cache(self, fetch_func, key: str, max_age_minutes: int = 60): """Fetch data with caching.""" cached = self.get_cached_data(key, max_age_minutes) if cached: return cached
if not self.client: self.connect()
data = fetch_func() self.save_to_cache(key, data) return dataStep 3: Create a Telegram Bot Interface
Build a Telegram bot for manual data entry. This is crucial for tracking things wearables can’t capture:
import telebotfrom datetime import datetimeimport jsonfrom pathlib import Path
bot = telebot.TeleBot(os.environ.get("TELEGRAM_BOT_TOKEN"))DATA_DIR = Path("./health_data")DATA_DIR.mkdir(exist_ok=True)
def save_record(user_id: int, record: dict): """Save health record to user's data file.""" user_file = DATA_DIR / f"user_{user_id}.jsonl"
with open(user_file, "a") as f: f.write(json.dumps(record) + "\n")
def parse_water_amount(text: str) -> int: """Parse water amount from various formats.""" text = text.lower().strip()
# Handle "500ml" or "500 ml" or just "500" if "ml" in text: return int(text.replace("ml", "").strip()) elif "l" in text: # Convert liters to ml return int(float(text.replace("l", "").strip()) * 1000) else: return int(text)
@bot.message_handler(commands=["start"])def handle_start(message): """Send welcome message with available commands.""" welcome = """Health Tracker Bot
Commands:/water <amount> - Log water (e.g., /water 500 or /water 1l)/event <note> - Log daily event/feeling/meal <description> - Log meal/sleep <hours> <quality> - Log sleep (quality: 1-10)/workout <type> <duration> - Log manual workout/stats - View today's summary/help - Show this message""" bot.reply_to(message, welcome)
@bot.message_handler(func=lambda m: m.text.startswith("/water"))def log_water(message): """Log water intake: /water 500 or /water 1l""" try: parts = message.text.split(maxsplit=1) if len(parts) < 2: bot.reply_to(message, "Usage: /water <amount>\nExample: /water 500 or /water 1l") return
amount_ml = parse_water_amount(parts[1])
record = { "user_id": message.from_user.id, "type": "water_intake", "amount_ml": amount_ml, "timestamp": datetime.now().isoformat(), "source": "telegram" }
save_record(message.from_user.id, record) bot.reply_to(message, f"Logged {amount_ml}ml of water")
except ValueError: bot.reply_to(message, "Invalid amount. Use: /water 500 or /water 1l")
@bot.message_handler(func=lambda m: m.text.startswith("/event"))def log_event(message): """Log daily event/feeling: /event felt tired during run""" event_text = message.text[7:].strip()
if not event_text: bot.reply_to(message, "Usage: /event <description>\nExample: /event felt tired during run") return
record = { "user_id": message.from_user.id, "type": "daily_event", "description": event_text, "timestamp": datetime.now().isoformat(), "source": "telegram" }
save_record(message.from_user.id, record) bot.reply_to(message, f"Event logged: {event_text}")
@bot.message_handler(func=lambda m: m.text.startswith("/meal"))def log_meal(message): """Log meal: /meal chicken salad with rice""" meal_text = message.text[6:].strip()
if not meal_text: bot.reply_to(message, "Usage: /meal <description>") return
record = { "user_id": message.from_user.id, "type": "meal", "description": meal_text, "timestamp": datetime.now().isoformat(), "source": "telegram" }
save_record(message.from_user.id, record) bot.reply_to(message, f"Meal logged: {meal_text}")
@bot.message_handler(func=lambda m: m.text.startswith("/sleep"))def log_sleep(message): """Log sleep: /sleep 7.5 8""" try: parts = message.text.split() if len(parts) < 3: bot.reply_to(message, "Usage: /sleep <hours> <quality 1-10>") return
hours = float(parts[1]) quality = int(parts[2])
if quality < 1 or quality > 10: bot.reply_to(message, "Quality must be between 1 and 10") return
record = { "user_id": message.from_user.id, "type": "sleep_manual", "hours": hours, "quality": quality, "timestamp": datetime.now().isoformat(), "source": "telegram" }
save_record(message.from_user.id, record) bot.reply_to(message, f"Sleep logged: {hours}h, quality: {quality}/10")
except (ValueError, IndexError): bot.reply_to(message, "Usage: /sleep <hours> <quality 1-10>")
@bot.message_handler(commands=["stats"])def show_stats(message): """Show today's health summary.""" user_id = message.from_user.id user_file = DATA_DIR / f"user_{user_id}.jsonl"
if not user_file.exists(): bot.reply_to(message, "No data logged yet. Start with /water, /event, /meal, or /sleep") return
today = datetime.now().strftime("%Y-%m-%d") total_water = 0 events = [] meals = []
with open(user_file) as f: for line in f: record = json.loads(line) if record["timestamp"].startswith(today): if record["type"] == "water_intake": total_water += record["amount_ml"] elif record["type"] == "daily_event": events.append(record["description"]) elif record["type"] == "meal": meals.append(record["description"])
summary = f"Today's Summary ({today}):\n\n" summary += f"Water: {total_water}ml\n"
if events: summary += f"\nEvents:\n- " + "\n- ".join(events)
if meals: summary += f"\n\nMeals:\n- " + "\n- ".join(meals)
bot.reply_to(message, summary)
if __name__ == "__main__": print("Health Tracker Bot starting...") bot.infinity_polling()Step 4: Implement the AI Analysis Layer
Build an AI agent using LangChain that aggregates all data streams and generates insights:
from langchain_openai import ChatOpenAIfrom langchain.prompts import PromptTemplatefrom langchain.tools import Toolfrom langchain.agents import initialize_agent, AgentTypefrom typing import Dict, Any, Listimport jsonfrom datetime import datetime, date
class HealthInsightsAgent: def __init__(self, openai_api_key: str): self.llm = ChatOpenAI( model="gpt-4", temperature=0.3, openai_api_key=openai_api_key ) self.tools = self._create_tools() self.agent = initialize_agent( tools=self.tools, llm=self.llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True )
def _create_tools(self) -> List[Tool]: """Create tools for the health agent."""
def analyze_sleep_quality(sleep_data: str) -> str: """Analyze sleep patterns and quality.""" prompt = PromptTemplate( template=""" Analyze this sleep data and provide insights: {sleep_data}
Provide: 1. Sleep quality score (1-10) 2. Key observations 3. Recommendations """, input_variables=["sleep_data"] ) return prompt.format(sleep_data=sleep_data)
def check_hydration(water_data: str) -> str: """Check hydration status.""" prompt = PromptTemplate( template=""" Analyze water intake data: {water_data}
Daily recommendation: 2000-3000ml Provide hydration status and recommendations. """, input_variables=["water_data"] ) return prompt.format(water_data=water_data)
def assess_training_load(activity_data: str) -> str: """Assess training load and recovery.""" prompt = PromptTemplate( template=""" Analyze training data: {activity_data}
Consider: - Training volume and intensity - Recovery indicators - Signs of overtraining - Recommendations for rest or intensity adjustment """, input_variables=["activity_data"] ) return prompt.format(activity_data=activity_data)
return [ Tool( name="analyze_sleep", func=analyze_sleep_quality, description="Analyze sleep quality and patterns" ), Tool( name="check_hydration", func=check_hydration, description="Check hydration status based on water intake" ), Tool( name="assess_training", func=assess_training_load, description="Assess training load and recovery needs" ) ]
def generate_daily_report( self, garmin_data: Dict[str, Any], manual_logs: List[Dict[str, Any]] ) -> str: """Generate comprehensive daily health report."""
# Aggregate data steps = garmin_data.get("steps", [{}])[0].get("totalSteps", 0) if garmin_data.get("steps") else 0 sleep_hours = self._extract_sleep_hours(garmin_data.get("sleep", {}))
water_ml = sum( log.get("amount_ml", 0) for log in manual_logs if log.get("type") == "water_intake" )
events = [ log.get("description", "") for log in manual_logs if log.get("type") == "daily_event" ]
activities = garmin_data.get("activities", [])[:3] # Last 3 activities
prompt = PromptTemplate( template=""" Generate a daily health report with actionable insights.
DATA: - Steps today: {steps} - Sleep hours: {sleep_hours} - Water intake: {water_ml}ml - Recent workouts: {activities} - User notes: {events}
Provide: 1. Overall health status (1-10) 2. Key observations (2-3 bullet points) 3. Recommendations for today 4. Alerts (if any metrics are concerning) """, input_variables=["steps", "sleep_hours", "water_ml", "activities", "events"] )
report = self.llm.invoke(prompt.format( steps=steps, sleep_hours=sleep_hours, water_ml=water_ml, activities=json.dumps(activities), events="\n".join(events) if events else "None" ))
return report.content
def _extract_sleep_hours(self, sleep_data: Dict) -> float: """Extract sleep hours from Garmin sleep data.""" if not sleep_data: return 0.0
daily_sleep = sleep_data.get("dailySleepDTO", {}) sleep_time = daily_sleep.get("sleepTimeSeconds", 0) return round(sleep_time / 3600, 1) if sleep_time else 0.0
def check_overtraining_risk( self, garmin_data: Dict[str, Any], manual_logs: List[Dict[str, Any]] ) -> Dict[str, Any]: """Check for overtraining risk indicators."""
activities = garmin_data.get("activities", []) stress_data = garmin_data.get("stress", []) body_battery = garmin_data.get("body_battery", [])
# Recent activity count recent_activities = len([a for a in activities if a])
# Average stress avg_stress = 0 if stress_data and len(stress_data) > 0: stress_values = [ s.get("stressLevel", 0) for s in stress_data if s.get("stressLevel") is not None ] avg_stress = sum(stress_values) / len(stress_values) if stress_values else 0
# Body battery charged body_battery_charged = 0 if body_battery and len(body_battery) > 0: body_battery_charged = body_battery[0].get("charged", 0)
# User-reported fatigue from events fatigue_mentions = [ log.get("description", "") for log in manual_logs if log.get("type") == "daily_event" and any(word in log.get("description", "").lower() for word in ["tired", "exhausted", "fatigue", "sore"]) ]
return { "recent_activity_count": recent_activities, "average_stress": round(avg_stress, 1), "body_battery_charged": body_battery_charged, "fatigue_mentions": len(fatigue_mentions), "overtraining_risk": self._calculate_risk( recent_activities, avg_stress, body_battery_charged, len(fatigue_mentions) ) }
def _calculate_risk( self, activity_count: int, avg_stress: float, body_battery: int, fatigue_count: int ) -> str: """Calculate overtraining risk level.""" risk_score = 0
if activity_count > 5: # More than 5 activities in recent period risk_score += 2 if avg_stress > 50: risk_score += 2 if body_battery < 50: risk_score += 2 if fatigue_count > 0: risk_score += fatigue_count
if risk_score >= 6: return "HIGH" elif risk_score >= 3: return "MODERATE" else: return "LOW"
# Usage exampleif __name__ == "__main__": import os
agent = HealthInsightsAgent(openai_api_key=os.environ["OPENAI_API_KEY"])
# Sample data (would come from Garmin and manual logs) sample_garmin = { "steps": [{"totalSteps": 8500}], "sleep": {"dailySleepDTO": {"sleepTimeSeconds": 27000}}, # 7.5 hours "activities": [{"type": "running", "distance": 10000}], "stress": [{"stressLevel": 45}], "body_battery": [{"charged": 75}] }
sample_manual = [ {"type": "water_intake", "amount_ml": 500, "timestamp": "..."}, {"type": "water_intake", "amount_ml": 750, "timestamp": "..."}, {"type": "daily_event", "description": "felt good during morning run"} ]
report = agent.generate_daily_report(sample_garmin, sample_manual) print(report)
risk = agent.check_overtraining_risk(sample_garmin, sample_manual) print(f"\nOvertraining Risk: {risk}")Step 5: Create Visualization and Alerts
Build a simple dashboard and set up proactive notifications:
from dataclasses import dataclassfrom typing import Optionalimport json
@dataclassclass HealthDashboard: """Simple health dashboard for daily tracking."""
def format_daily_summary(self, data: dict) -> str: """Format daily health summary for display.""" summary = """HEALTH DASHBOARD - {date}================================
ACTIVITY--------Steps: {steps:,}Active Minutes: {active_min}Calories: {calories}
SLEEP-----Duration: {sleep_hours}hQuality Score: {sleep_quality}/100
HYDRATION---------Water: {water_ml}ml / 2500ml goalProgress: {water_pct}%
RECOVERY--------Body Battery: {body_battery}%Stress Level: {stress_level}
TRAINING--------Recent Workouts: {workout_count}Overtraining Risk: {risk_level}
NOTES-----{notes}
RECOMMENDATIONS--------------{recommendations}""" return summary.format(**data)
def format_alert(self, alert_type: str, data: dict) -> str: """Format alert for Telegram notification.""" alerts = { "low_hydration": ( "WARNING: Low Hydration\n\n" f"Current: {data.get('water_ml', 0)}ml\n" f"Target: 2500ml\n" f"Deficit: {2500 - data.get('water_ml', 0)}ml\n\n" "Drink at least 500ml more water today." ), "overtraining": ( "ALERT: Overtraining Risk\n\n" f"Risk Level: {data.get('risk_level', 'UNKNOWN')}\n" f"Indicators:\n" f"- Activities this week: {data.get('activity_count', 0)}\n" f"- Average stress: {data.get('avg_stress', 0)}\n" f"- Body battery: {data.get('body_battery', 0)}%\n\n" "Consider a rest day or light activity." ), "poor_sleep": ( "NOTICE: Poor Sleep Quality\n\n" f"Sleep duration: {data.get('sleep_hours', 0)}h\n" f"Target: 7-8 hours\n\n" "Consider adjusting your evening routine." ) } return alerts.get(alert_type, "Unknown alert type")
class AlertManager: """Manage health alerts and notifications."""
def __init__(self, telegram_bot, user_id: int): self.bot = telegram_bot self.user_id = user_id self.alert_thresholds = { "water_min": 1500, # Alert if below this by 6pm "sleep_min_hours": 6, "body_battery_min": 30, "stress_max": 70 }
def check_and_alert(self, health_data: dict) -> list: """Check health data and send alerts if needed.""" alerts_sent = []
# Check hydration water = health_data.get("water_ml", 0) current_hour = datetime.now().hour
if current_hour >= 18 and water < self.alert_thresholds["water_min"]: alert = self._send_alert("low_hydration", {"water_ml": water}) alerts_sent.append(alert)
# Check body battery body_battery = health_data.get("body_battery", 100) if body_battery < self.alert_thresholds["body_battery_min"]: alert = self._send_alert("overtraining", { "risk_level": "MODERATE", "body_battery": body_battery, "activity_count": health_data.get("activity_count", 0), "avg_stress": health_data.get("avg_stress", 0) }) alerts_sent.append(alert)
# Check sleep sleep_hours = health_data.get("sleep_hours", 8) if sleep_hours < self.alert_thresholds["sleep_min_hours"]: alert = self._send_alert("poor_sleep", {"sleep_hours": sleep_hours}) alerts_sent.append(alert)
return alerts_sent
def _send_alert(self, alert_type: str, data: dict) -> str: """Send alert via Telegram.""" dashboard = HealthDashboard() message = dashboard.format_alert(alert_type, data)
# In production, send via Telegram bot # self.bot.send_message(self.user_id, message) print(f"ALERT: {message}") return alert_typeWhy This Works
This architecture solves the fragmentation problem by:
-
Centralized aggregation: OpenClaw pulls from Garmin Connect automatically, eliminating manual data export/import.
-
Frictionless manual input: Telegram bot makes logging water, meals, and feelings as easy as sending a message. No app switching required.
-
Intelligent insights: The AI layer doesn’t just store data - it correlates sleep with performance, spots overtraining patterns, and provides actionable recommendations.
-
Proactive alerts: Instead of passively displaying data, the system alerts you when hydration is low or overtraining risk is high.
For a 55-year-old training for a marathon, these insights can be the difference between successful training and injury.
Common Mistakes
I made these mistakes when building this system:
Over-engineering too early: I initially tried to integrate Garmin, Fitbit, Apple Watch, and Oura Ring all at once. Don’t. Start with one wearable, get it working, then expand.
Ignoring data validation: Bad data leads to bad AI insights. A Garmin sync error once recorded 50,000 steps in an hour. Always validate incoming data for reasonable ranges.
Not handling API rate limits: Garmin has strict rate limits. I got temporarily blocked during testing. Implement caching (shown in the code) and respect rate limits.
Making manual entry cumbersome: My first Telegram bot had too many required fields. People won’t log if it takes more than 5 seconds. Keep it to /water 500 not /water amount:500 unit:ml time:14:30.
Skipping the AI analysis: Just aggregating data isn’t enough. The value comes from AI-generated insights that help you make decisions about training, recovery, and lifestyle.
Summary
This post showed how to build a health tracking AI agent that integrates wearable data with manual input. The key components are:
- OpenClaw as the data aggregation hub
- Garmin Connect integration for automated wearable data pulls
- Telegram bot for quick manual logging
- LangChain AI agent for generating insights and alerts
The system works because it removes friction from health tracking. Wearable data flows in automatically. Manual logging takes seconds via Telegram. AI turns raw data into recommendations.
Start simple: one wearable, basic manual logging, simple insights. Then expand as you learn what metrics matter most for your goals.
Final Words + More Resources
My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me
Here are also the most important links from this article along with some further resources that will help you in this scope:
- 👨💻 OpenClaw
- 👨💻 Garmin Connect Python Library
- 👨💻 python-telegram-bot
- 👨💻 LangChain Documentation
- 👨💻 LangGraph Guide
Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!
Comments