Sync GoHighLevel to Google Sheets and a Data Warehouse

async function* getAllContacts(token, locationId) {
  const base = 'https://services.leadconnectorhq.com';
  const headers = {
    Authorization: `Bearer ${token}`,
    Version: '2021-07-28',
  };
  const limit = 100;
  let startAfter = null;

  while (true) {
    const params = new URLSearchParams({ locationId, limit: String(limit) });
    if (startAfter) params.append('startAfter', startAfter);

    const res = await fetch(`${base}/contacts/search?${params}`, { headers });
    if (!res.ok) throw new Error(`GHL error: ${res.status}`);

    const data = await res.json();
    const contacts = data.contacts ?? [];

    for (const contact of contacts) {
      yield contact;
    }

    if (contacts.length < limit) break; // last page
    startAfter = data.meta?.startAfterRecord ?? contacts[contacts.length - 1].id;
  }
}

// Usage
for await (const contact of getAllContacts(process.env.GHL_TOKEN, process.env.GHL_LOCATION_ID)) {
  console.log(contact.id, contact.email);
}

import time
import requests

def get_all_contacts(token, location_id):
    base = "https://services.leadconnectorhq.com"
    headers = {"Authorization": f"Bearer {token}", "Version": "2021-07-28"}
    limit = 100
    start_after = None

    while True:
        params = {"locationId": location_id, "limit": limit}
        if start_after:
            params["startAfter"] = start_after

        res = requests.get(f"{base}/contacts/search", headers=headers, params=params)
        res.raise_for_status()
        data = res.json()
        contacts = data.get("contacts", [])

        yield from contacts

        if len(contacts) < limit:
            break  # last page

        meta = data.get("meta", {})
        start_after = meta.get("startAfterRecord") or contacts[-1]["id"]
        time.sleep(0.2)  # be kind to the rate limit

import { google } from 'googleapis';

async function appendContactsToSheet(contacts, spreadsheetId, sheetName) {
  const auth = new google.auth.GoogleAuth({
    keyFile: process.env.GOOGLE_APPLICATION_CREDENTIALS,
    // or: credentials: JSON.parse(process.env.GOOGLE_SERVICE_ACCOUNT_JSON)
    scopes: ['https://www.googleapis.com/auth/spreadsheets'],
  });
  const sheets = google.sheets({ version: 'v4', auth });

  const rows = contacts.map((c) => [
    c.id,
    c.email ?? '',
    c.firstName ?? '',
    c.lastName ?? '',
    c.phone ?? '',
    (c.tags ?? []).join(', '),
    c.dateAdded ?? '',
    c.dateUpdated ?? '',
  ]);

  await sheets.spreadsheets.values.append({
    spreadsheetId,
    range: `${sheetName}!A1`,
    valueInputOption: 'RAW',
    insertDataOption: 'INSERT_ROWS',
    requestBody: { values: rows },
  });

  console.log(`Appended ${rows.length} contacts`);
}

import json
import os
import gspread
from google.oauth2.service_account import Credentials

SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
creds = Credentials.from_service_account_info(
    json.loads(os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"]),
    scopes=SCOPES,
)
gc = gspread.authorize(creds)

def append_contacts_to_sheet(contacts, spreadsheet_id, sheet_name):
    sh = gc.open_by_key(spreadsheet_id)
    ws = sh.worksheet(sheet_name)

    rows = [
        [
            c.get("id", ""),
            c.get("email", ""),
            c.get("firstName", ""),
            c.get("lastName", ""),
            c.get("phone", ""),
            ", ".join(c.get("tags", [])),
            c.get("dateAdded", ""),
            c.get("dateUpdated", ""),
        ]
        for c in contacts
    ]
    if rows:
        ws.append_rows(rows, value_input_option="RAW")
    print(f"Appended {len(rows)} contacts")

import postgres from 'postgres';

const sql = postgres(process.env.DATABASE_URL);

async function upsertContacts(contacts) {
  const rows = contacts.map((c) => ({
    id: c.id,
    location_id: c.locationId,
    email: c.email ?? null,
    first_name: c.firstName ?? null,
    last_name: c.lastName ?? null,
    phone: c.phone ?? null,
    tags: c.tags ?? [],
    date_added: c.dateAdded ? new Date(c.dateAdded) : null,
    date_updated: c.dateUpdated ? new Date(c.dateUpdated) : null,
    synced_at: new Date(),
  }));

  await sql`
    INSERT INTO ghl_contacts ${sql(rows)}
    ON CONFLICT (id) DO UPDATE SET
      email = EXCLUDED.email,
      first_name = EXCLUDED.first_name,
      last_name = EXCLUDED.last_name,
      phone = EXCLUDED.phone,
      tags = EXCLUDED.tags,
      date_updated = EXCLUDED.date_updated,
      synced_at = EXCLUDED.synced_at
  `;
  console.log(`Upserted ${rows.length} contacts`);
}

CREATE TABLE IF NOT EXISTS ghl_contacts (
  id            TEXT PRIMARY KEY,
  location_id   TEXT NOT NULL,
  email         TEXT,
  first_name    TEXT,
  last_name     TEXT,
  phone         TEXT,
  tags          TEXT[] DEFAULT '{}',
  date_added    TIMESTAMPTZ,
  date_updated  TIMESTAMPTZ,
  synced_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_ghl_contacts_location
  ON ghl_contacts (location_id);

CREATE INDEX IF NOT EXISTS idx_ghl_contacts_email
  ON ghl_contacts (email);

{
  "name": "ghl_export_contacts",
  "description": "Export all GoHighLevel contacts for a location to a JSON array, paginating automatically.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "locationId": { "type": "string" },
      "updatedSince": {
        "type": "string",
        "description": "ISO 8601 datetime. Only return contacts updated after this time."
      },
      "limit": {
        "type": "number",
        "description": "Max contacts to return (default 1000, max 10000)"
      }
    },
    "required": ["locationId"]
  }
}