fitbit_get_hrv

Instructions

Implementation Reference

• src/fitbit_mcp/tools/hrv_tools.py:34-77 (handler)

  Main tool handler for fitbit_get_hrv. Decorated with @mcp.tool() and @require_auth. Accepts optional start_date, end_date, and live flag. Parses dates, then either fetches live from the Fitbit API via _fetch_live() or queries the local SQLite cache via db.query_hrv(). Returns formatted JSON response with HRV entries.

  @mcp.tool()
  @require_auth
  async def fitbit_get_hrv(
      start_date: str | None = None,
      end_date: str | None = None,
      live: bool = False,
  ) -> str:
      """Get nightly HRV (heart rate variability) data.
  
      Returns data from the local cache by default. Use live=True to fetch
      from Fitbit API. Run fitbit_sync first to populate the cache.
  
      HRV data is sparse: only nights with on-wrist sleep tracking produce readings.
      Requires Fitbit Premium for access to this endpoint.
  
      Args:
          start_date: Start date as "YYYY-MM-DD", "YYYY-MM", or "30d". Default: last 30 days.
          end_date: End date as "YYYY-MM-DD". Default: today.
          live: If true, fetch directly from Fitbit API instead of cache.
  
      Returns one entry per night with daily_rmssd and deep_rmssd (ms).
      RMSSD = root mean square of successive RR interval differences.
      Higher values generally indicate better recovery and parasympathetic activity.
      """
      start, end = parse_date(start_date, end_date, default_days=30)
  
      if live:
          entries = await anyio.to_thread.run_sync(lambda: _fetch_live(start, end))
      else:
          await anyio.to_thread.run_sync(lambda: auto_sync_if_stale("hrv"))
          def _query():
              conn = db.get_db()
              rows = db.query_hrv(conn, start.isoformat(), end.isoformat())
              conn.close()
              return rows
          entries = await anyio.to_thread.run_sync(_query)
  
      if not entries:
          return format_response({
              "message": "No HRV data found for this period.",
              "hint": "Try live=True to fetch directly from the API.",
          })
  
      return format_response({"hrv": entries, "count": len(entries)})

• src/fitbit_mcp/tools/hrv_tools.py:36-41 (schema)

  Input schema/type hints for the tool: start_date (optional str), end_date (optional str), live (optional bool, default False). Return type is str (JSON serialized).

  async def fitbit_get_hrv(
      start_date: str | None = None,
      end_date: str | None = None,
      live: bool = False,
  ) -> str:
      """Get nightly HRV (heart rate variability) data.

• src/fitbit_mcp/tools/hrv_tools.py:34-34 (registration)

  Tool registration via @mcp.tool() decorator on the mcp instance from src/fitbit_mcp/mcp_instance.py

  @mcp.tool()

• src/fitbit_mcp/tools/hrv_tools.py:13-31 (helper)

  Private helper function that fetches HRV data live from the Fitbit API using chunked date ranges (up to HRV_MAX_RANGE_DAYS=30 per request). Parses the response into a sorted list of dicts with date, daily_rmssd, and deep_rmssd.

  def _fetch_live(start_date, end_date) -> list[dict]:
      """Fetch HRV data directly from the API."""
      from ..config import HRV_MAX_RANGE_DAYS
      results = {}
      d = start_date
      while d <= end_date:
          chunk_end = min(d + timedelta(days=HRV_MAX_RANGE_DAYS - 1), end_date)
          path = f"/1/user/-/hrv/date/{d}/{chunk_end}.json"
          data = api.get(path)
          for entry in data.get("hrv", []):
              ds = entry.get("dateTime")
              if ds and "value" in entry:
                  results[ds] = {
                      "date": ds,
                      "daily_rmssd": entry["value"].get("dailyRmssd"),
                      "deep_rmssd": entry["value"].get("deepRmssd"),
                  }
          d = chunk_end + timedelta(days=1)
      return sorted(results.values(), key=lambda x: x["date"])

• src/fitbit_mcp/db.py:280-285 (helper)

  Database helper to query cached HRV data from the SQLite hrv table within a date range.

  def query_hrv(conn: sqlite3.Connection, start_date: str, end_date: str) -> list[dict]:
      rows = conn.execute(
          "SELECT * FROM hrv WHERE date >= ? AND date <= ? ORDER BY date",
          (start_date, end_date),
      ).fetchall()
      return _rows_to_dicts(rows)