Terra Webhooks

Handle real-time health data delivery from Terra API.

Quick Start

from flask import Flask, request
import hmac
import hashlib

app = Flask(__name__)

TERRA_SIGNING_SECRET = "your_signing_secret_from_dashboard"

@app.route("/webhooks/terra", methods=["POST"])
def handle_terra_webhook():
    # 1. Verify signature
    signature = request.headers.get("terra-signature")
    if not verify_signature(signature, request.get_data()):
        return "Invalid signature", 401

    # 2. Parse payload
    payload = request.get_json()
    event_type = payload.get("type")

    # 3. Handle event
    if event_type == "activity":
        handle_activity(payload)
    elif event_type == "sleep":
        handle_sleep(payload)
    elif event_type == "auth":
        handle_user_connected(payload)

    # 4. Respond immediately
    return "OK", 200

def verify_signature(header: str, body: bytes) -> bool:
    """Verify Terra webhook signature."""
    parts = dict(p.split("=") for p in header.split(","))
    timestamp = parts["t"]
    signature = parts["v1"]

    message = f"{timestamp}.{body.decode()}"
    expected = hmac.new(
        TERRA_SIGNING_SECRET.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, signature)

Webhook Event Types

Authentication Events

Data Events

Processing Events

Event Payloads

auth - User Connected

{
  "type": "auth",
  "user": {
    "user_id": "terra_abc123",
    "provider": "FITBIT",
    "reference_id": "user_12345",
    "scopes": ["activity", "sleep", "body"]
  },
  "status": "authenticated"
}

activity - Workout Data

{
  "type": "activity",
  "user": {
    "user_id": "terra_abc123",
    "provider": "GARMIN",
    "reference_id": "user_12345"
  },
  "data": [{
    "metadata": {
      "start_time": "2025-12-05T07:00:00Z",
      "end_time": "2025-12-05T08:00:00Z",
      "type": "running"
    },
    "calories_data": {
      "total_burned_calories": 450
    },
    "heart_rate_data": {
      "summary": { "avg_hr_bpm": 145, "max_hr_bpm": 175 }
    },
    "distance_data": { "distance_meters": 8500 }
  }]
}

sleep - Sleep Data

{
  "type": "sleep",
  "user": {
    "user_id": "terra_abc123",
    "provider": "OURA"
  },
  "data": [{
    "metadata": {
      "start_time": "2025-12-04T22:30:00Z",
      "end_time": "2025-12-05T06:30:00Z"
    },
    "sleep_durations_data": {
      "sleep_efficiency": 0.92
    },
    "asleep": {
      "duration_deep_sleep_state_seconds": 5400,
      "duration_REM_sleep_state_seconds": 6600
    }
  }]
}

daily - Daily Summary

{
  "type": "daily",
  "user": {
    "user_id": "terra_abc123",
    "provider": "FITBIT"
  },
  "data": [{
    "metadata": {
      "start_time": "2025-12-05T00:00:00Z",
      "end_time": "2025-12-05T23:59:59Z"
    },
    "movement_data": { "steps_count": 10500 },
    "calories_data": { "total_burned_calories": 2400 }
  }]
}

deauth - User Disconnected

{
  "type": "deauth",
  "user": {
    "user_id": "terra_abc123",
    "provider": "FITBIT"
  },
  "status": "deauthenticated"
}

Operations

setup-webhook-endpoint

Create a production-ready webhook handler.

from flask import Flask, request
from celery import Celery
import hmac
import hashlib
import logging

app = Flask(__name__)
celery = Celery()
logger = logging.getLogger(__name__)

TERRA_SIGNING_SECRET = "your_signing_secret"

# Terra webhook source IPs (for additional security)
TERRA_IPS = [
    "18.133.218.210", "18.169.82.189", "18.132.162.19",
    "18.130.218.186", "13.43.183.154", "3.11.208.36",
    "35.214.201.105", "35.214.230.71", "35.214.252.53", "35.214.229.114"
]

@app.route("/webhooks/terra", methods=["POST"])
def terra_webhook():
    # Optional: IP whitelist check
    client_ip = request.remote_addr
    if client_ip not in TERRA_IPS:
        logger.warning(f"Webhook from unknown IP: {client_ip}")
        # Consider: return "Forbidden", 403

    # Verify signature
    signature = request.headers.get("terra-signature")
    raw_body = request.get_data()

    if not signature or not verify_signature(signature, raw_bod

...