API-driven e-commerce automation: ActiveCampaign custom events vs Shopify Flow, with a tested Python relay

Most Shopify stores start their email automation with two built-in pieces: Shopify Flow for logic inside the store, and the email platform's native Shopify integration for customers, orders and abandoned carts. That covers the standard flows. It stops covering them the moment marketing asks for something the platforms don't model: "email people who keep looking at trail shoes but never buy", "segment by lifetime value after refunds", "tell us when an order sync silently fails". This guide shows where the built-in tools end, and builds the smallest relay that gets Shopify data into ActiveCampaign's API reliably.

1. The short answer

	Shopify Flow	ActiveCampaign native integration	Custom relay to ActiveCampaign's API
Best at	Actions inside Shopify: tags, metafields, notes, internal alerts	Syncing customers, orders and carts with no code	Signals and values neither of the others model
Sees storefront browsing	Only through the "Customer left online store" trigger, at product level	Site tracking on pages it can tag	Any standard pixel event, e.g. collection_viewed
Calls external APIs	Send HTTP request: Grow, Advanced and Plus plans only	Not applicable	Yes, any endpoint
Retries	Optional, up to 24 hours on 4XX, 5XX and 429	Handled by the vendor, not visible to you	Yours: outbox, backoff, dead letters
Custom computed fields	Liquid in the request body	Fixed set of synced fields	Anything you can compute
Cost	Free app	Included with ActiveCampaign	A small server and your time

2. Where Flow and the native integration stop

Shopify Flow is free on Basic, Grow, Advanced and Plus. Its Send HTTP request action is not available on Basic. It waits up to 30 seconds for a response, treats 2XX and 3XX as success, and for 4XX, 5XX and 429 responses lets you choose retry (for up to 24 hours), fail or ignore. Secrets are stored in Flow settings and redacted from run logs. For many stores, that's enough to push one value per event. A workflow on Order paid could send the customer's Shopify-computed total spent to ActiveCampaign's contact sync endpoint with a body like this:

POST https://ACCOUNT.api-us1.com/api/3/contact/sync
Api-Token: {{secrets.ac_token}}

{"contact": {"email": "{{order.customer.email}}",
  "fieldValues": [{"field": "12", "value": "{{order.customer.amountSpent.amount}}"}]}}

If that covers your needs, stop here: it's less to run than anything below. Flow runs out of road in three places:

• Browsing signals. The Customer left online store without making a purchase trigger covers product browse, cart and checkout abandonment, and its actions are tags, metafields and notes. It doesn't tell you that someone viewed the same collection three times in two days.
• State across events. Each Flow run handles one event. Rules like "only once per category per week" or "not if they bought anything since" need storage Flow doesn't give you.
• Visibility. A failed run shows in Flow's run log, one workflow at a time. There's no queue you can count, alert on or replay in bulk.

ActiveCampaign's native Shopify integration syncs customers and orders and powers its abandoned cart automations. Use it for that: don't rebuild order sync. A relay should add what the integration doesn't send, not duplicate orders it already sends, or revenue reporting will count them twice.

3. The architecture

Two rules shape everything else. First, answer Shopify fast: Shopify allows one second to connect and five seconds for the whole request, then retries 8 times over 4 hours and deletes an Admin API subscription after 8 consecutive failures. Second, respect ActiveCampaign's budget: 5 requests per second per account, shared with every other tool using your API key, with 429 and a Retry-After header when you exceed it. An outbox between the two solves both.

4. Capturing category views with a custom pixel

Shopify's Web Pixels API exposes standard events, including collection_viewed with the collection's ID, title and product variants. A custom pixel runs in a sandbox and can only send data out, so it posts to the relay:

// Shopify admin > Settings > Customer events > Add custom pixel. Set its permission to require marketing consent.
const RELAY = 'https://relay.example.com/pixel';

function send(name, event, extra) {
  const customer = init.data.customer;  // set for logged-in customers only
  fetch(RELAY, {
    method: 'POST',
    keepalive: true,                          // survives the shopper leaving the page
    headers: {'Content-Type': 'text/plain'},  // a "simple" request: no CORS preflight
    body: JSON.stringify(Object.assign(
      {name: name, client_id: event.clientId, email: customer ? customer.email : null}, extra)),
  });
}

analytics.subscribe('collection_viewed', (e) => send('collection_viewed', e, {
  collection: {id: e.data.collection.id, title: e.data.collection.title},
}));
analytics.subscribe('checkout_completed', (e) => send('checkout_completed', e, {
  email: e.data.checkout.email,  // links this browser's client_id to an email for later visits
}));

• Identity is the hard part. The pixel gets a clientId per browser. An email is only available for logged-in customers (init.data.customer) or after a completed checkout, which links that browser to an email for later visits.
• Consent. Set the pixel's customer privacy permission so it only runs with marketing consent, and document the processing in your privacy policy.
• text/plain on purpose. A JSON content type would trigger a CORS preflight from the sandbox; plain text is a simple request and the relay parses it as JSON anyway.

5. The relay

#!/usr/bin/env python3
"""Shopify -> ActiveCampaign relay for what the native integration and Shopify Flow don't do:
lifetime value on the contact, and a "category_abandoned" custom event.
Python 3.9+, standard library only. One process: an HTTP receiver, a SQLite outbox and a sender thread.

  export SHOPIFY_WEBHOOK_SECRET=...            # app client secret, or the key in Settings > Notifications
  export AC_API=https://ACCOUNT.api-us1.com AC_TOKEN=...    # Settings > Developer
  export AC_ACTID=... AC_EVENT_KEY=...         # Website > Site Tracking > Event Tracking
  export AC_FIELD_LTV=12 AC_FIELD_ORDERS=13    # IDs of two custom contact fields you created
  python3 shopify_ac_relay.py                  # listens on :8080, keeps state in relay.db

  POST /webhooks/shopify   orders/paid and orders/updated, HMAC-verified
  POST /pixel              collection_viewed and checkout_completed from a custom pixel (untrusted input)
  GET  /health             outbox counts, for an uptime monitor
"""
import base64
import hashlib
import hmac
import json
import os
import random
import re
import sqlite3
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

E = os.environ
AC_EVENT_URL = E.get('AC_EVENT_URL', 'https://trackcmp.net/event')
ABANDON_AFTER = int(E.get('ABANDON_AFTER', 7200))  # seconds since the last category view
NUDGE_EVERY = int(E.get('NUDGE_EVERY', 7 * 86400))  # at most one nudge per email and category
MAX_ATTEMPTS = 10  # with RETRY_BASE=30: retries for about 3 to 4.5 hours, like Shopify's 4
RETRY_BASE = float(E.get('RETRY_BASE', 30))
COUNTED = ('paid', 'partially_paid', 'partially_refunded', 'refunded')
EMAIL = re.compile(r'^[^@\s]{1,64}@[^@\s]{1,190}\.[A-Za-z]{2,}$')

db = sqlite3.connect(E.get('RELAY_DB', 'relay.db'), check_same_thread=False, isolation_level=None)
lock = threading.Lock()
db.executescript('''
CREATE TABLE IF NOT EXISTS seen (webhook_id TEXT PRIMARY KEY, at REAL);
CREATE TABLE IF NOT EXISTS orders (id TEXT PRIMARY KEY, email TEXT, total REAL, counted INT, updated TEXT,
  at REAL);
CREATE TABLE IF NOT EXISTS clients (client_id TEXT PRIMARY KEY, email TEXT);
CREATE TABLE IF NOT EXISTS views (client_id TEXT, collection TEXT, title TEXT, at REAL);
CREATE TABLE IF NOT EXISTS nudges (email TEXT, collection TEXT, at REAL, PRIMARY KEY (email, collection));
CREATE TABLE IF NOT EXISTS outbox (id INTEGER PRIMARY KEY, kind TEXT, email TEXT, payload TEXT,
  attempts INT DEFAULT 0, next_at REAL DEFAULT 0, status TEXT DEFAULT 'pending', error TEXT);
''')


def q(sql, *args):
    with lock:
        return db.execute(sql, args).fetchall()


def enqueue_contact(email):
    """Send absolute values, not increments: a retried or duplicated job then can't double-count."""
    ltv, n = q('SELECT COALESCE(SUM(total), 0), COUNT(*) FROM orders WHERE email = ? AND counted = 1',
               email)[0]
    payload = json.dumps({'email': email, 'ltv': round(ltv, 2), 'orders': n})
    with lock:  # a newer value replaces a pending one; no point sending a stale total first
        db.execute("DELETE FROM outbox WHERE kind = 'contact' AND email = ? AND status = 'pending'", (email,))
        db.execute("INSERT INTO outbox (kind, email, payload) VALUES ('contact', ?, ?)", (email, payload))


def on_order(o):
    email = (o.get('email') or (o.get('customer') or {}).get('email') or '').lower()
    if not email:
        return
    # current_total_price_set reflects edits and refunds; shop_money keeps one currency
    total = float(o['current_total_price_set']['shop_money']['amount'])
    counted = int(o.get('financial_status') in COUNTED)
    with lock:  # orders/updated can arrive out of order: keep the newest version of each order
        db.execute('''INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET
                      total = excluded.total, counted = excluded.counted, updated = excluded.updated
                      WHERE excluded.updated > orders.updated''',
                   (str(o['id']), email, total, counted, o.get('updated_at', ''), time.time()))
    enqueue_contact(email)


def sweep(now=None):
    """Known customers who viewed a category at least twice in 48 hours, then went quiet without ordering."""
    now = now or time.time()
    rows = q('''SELECT c.email, v.collection, MAX(v.title), MIN(v.at)
                FROM views v JOIN clients c USING (client_id)
                WHERE v.at > ? GROUP BY c.email, v.collection HAVING COUNT(*) >= 2 AND MAX(v.at) < ?''',
             now - 172800, now - ABANDON_AFTER)
    for email, coll, title, first in rows:
        if q('SELECT 1 FROM orders WHERE email = ? AND at > ?', email, first):
            continue  # bought something since
        if not q('SELECT 1 FROM orders WHERE email = ?', email):
            continue  # pixel emails are unverified: only nudge people who already ordered
        if q('SELECT 1 FROM nudges WHERE email = ? AND collection = ? AND at > ?',
             email, coll, now - NUDGE_EVERY):
            continue
        with lock:
            db.execute('INSERT OR REPLACE INTO nudges VALUES (?, ?, ?)', (email, coll, now))
            db.execute("INSERT INTO outbox (kind, email, payload) VALUES ('event', ?, ?)",
                       (email, json.dumps({'email': email, 'event': 'category_abandoned',
                                           'data': title[:255]})))


def call_ac(kind, p):
    if kind == 'contact':
        fields = [{'field': E['AC_FIELD_LTV'], 'value': '%.2f' % p['ltv']},
                  {'field': E['AC_FIELD_ORDERS'], 'value': p['orders']}]
        body = json.dumps({'contact': {'email': p['email'], 'fieldValues': fields}}).encode()
        req = urllib.request.Request(E['AC_API'] + '/api/3/contact/sync', body,
                                     {'Api-Token': E['AC_TOKEN'], 'Content-Type': 'application/json'})
    else:  # Event Tracking API: form-encoded, event name up to 32 characters
        form = {'actid': E['AC_ACTID'], 'key': E['AC_EVENT_KEY'], 'event': p['event'],
                'eventdata': p['data'], 'visit': json.dumps({'email': p['email']})}
        req = urllib.request.Request(AC_EVENT_URL, urllib.parse.urlencode(form).encode())
    with urllib.request.urlopen(req, timeout=20) as r:
        body = json.load(r)
    if kind == 'event' and body.get('success') != 1:
        raise ValueError('event rejected: %s' % body)


def sender():
    while True:
        job = q("SELECT id, kind, payload, attempts FROM outbox WHERE status = 'pending' AND next_at <= ? "
                "ORDER BY id LIMIT 1", time.time())
        if not job:
            time.sleep(1)
            continue
        jid, kind, payload, attempts = job[0]
        delay, error = None, None
        try:
            call_ac(kind, json.loads(payload))
        except urllib.error.HTTPError as e:
            error = 'HTTP %d' % e.code
            if e.code == 429:  # 5 requests/second per account, shared with every other client of the key
                delay = float(e.headers.get('Retry-After') or RETRY_BASE * 2 ** attempts)
            elif e.code >= 500:
                delay = RETRY_BASE * 2 ** attempts
        except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
            error, delay = str(e)[:200], RETRY_BASE * 2 ** attempts
        except ValueError as e:  # bad response or rejected event: retrying won't help
            error = str(e)[:200]
        with lock:
            if error is None:
                db.execute("UPDATE outbox SET status = 'done', error = NULL WHERE id = ?", (jid,))
            elif delay is None or attempts + 1 >= MAX_ATTEMPTS:
                db.execute("UPDATE outbox SET status = 'dead', attempts = ?, error = ? WHERE id = ?",
                           (attempts + 1, error, jid))
            else:  # jitter, so a backlog doesn't retry in lockstep after an outage
                db.execute('UPDATE outbox SET attempts = ?, next_at = ?, error = ? WHERE id = ?',
                           (attempts + 1, time.time() + min(delay, 3600) * random.uniform(1, 1.5),
                            error, jid))
        time.sleep(0.25)  # ponytail: fixed 4 req/s pacing, one sender; a token bucket if you add senders


class Handler(BaseHTTPRequestHandler):
    def reply(self, code, body=b''):
        self.send_response(code)
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path != '/health':
            return self.reply(404)
        counts = dict(q('SELECT status, COUNT(*) FROM outbox GROUP BY status'))
        oldest = q("SELECT MIN(id) FROM outbox WHERE status = 'pending'")[0][0]
        self.reply(200, json.dumps({'outbox': counts, 'oldest_pending_id': oldest}).encode())

    def do_POST(self):
        size = int(self.headers.get('Content-Length') or 0)
        if size > 1_000_000:
            return self.reply(413)
        raw = self.rfile.read(size)
        if self.path == '/webhooks/shopify':
            key = E['SHOPIFY_WEBHOOK_SECRET'].encode()
            digest = base64.b64encode(hmac.new(key, raw, hashlib.sha256).digest())
            if not hmac.compare_digest(digest, (self.headers.get('X-Shopify-Hmac-Sha256') or '').encode()):
                return self.reply(401)
            wid = self.headers.get('X-Shopify-Webhook-Id')
            if q('SELECT 1 FROM seen WHERE webhook_id = ?', wid):  # Shopify retries for 4 hours
                return self.reply(200)
            if self.headers.get('X-Shopify-Topic') in ('orders/paid', 'orders/updated'):
                on_order(json.loads(raw))  # idempotent, so marking it seen afterwards is safe
            q('INSERT OR IGNORE INTO seen VALUES (?, ?)', wid, time.time())
            return self.reply(200)  # answer within Shopify's 5 seconds; the sender does the slow part
        if self.path == '/pixel' and size <= 4096:
            try:
                ev = json.loads(raw)
                cid, name = str(ev['client_id'])[:64], ev['name']
                email = str(ev.get('email') or '').lower()
                if EMAIL.match(email):
                    q('INSERT OR REPLACE INTO clients VALUES (?, ?)', cid, email)
                if name == 'collection_viewed':
                    c = ev['collection']
                    q('INSERT INTO views VALUES (?, ?, ?, ?)', cid, str(c['id'])[:80], str(c['title'])[:255],
                      time.time())
            except (ValueError, KeyError, TypeError):
                return self.reply(400)
            return self.reply(204)
        self.reply(404)

    def log_message(self, *a):
        pass


if __name__ == '__main__':
    threading.Thread(target=sender, daemon=True).start()

    def sweeper():
        while True:
            sweep()
            time.sleep(60)
    threading.Thread(target=sweeper, daemon=True).start()
    ThreadingHTTPServer(('', int(E.get('PORT', 8080))), Handler).serve_forever()

The decisions that matter, and why:

• Absolute values, not increments. The contact job sends the total lifetime value and order count recomputed from the database, never "add $50". A retried, duplicated or reordered job then can't drift the number. Pending jobs for the same email are replaced, so a burst of order updates sends one request.
• Lifetime value after refunds. current_total_price_set.shop_money reflects edits and refunds and stays in the shop's currency in multi-currency stores. Orders count only when their financial status is paid, partially paid, partially refunded or refunded; pending and voided orders don't.
• Out-of-order updates. orders/updated can arrive after a newer update. The upsert keeps a row only if its updated_at is newer than the stored one.
• De-duplication. X-Shopify-Webhook-Id identifies a delivery; a retried delivery is acknowledged without processing. Processing is idempotent anyway, so the delivery is marked as seen only after it succeeds: a crash mid-way means Shopify retries and nothing is lost.
• The pixel is untrusted. Anyone can post to /pixel, so it only records views and email links, with size and format checks. The sweep nudges only emails that already have an order, at most once per category per week, so a forged event can't turn the relay into a spam cannon.
• Retries by error type. 429 waits for Retry-After; 5XX and network errors back off exponentially from 30 seconds, capped at an hour, with jitter; other 4XX errors and rejected events go straight to dead letters, because retrying a bad request doesn't fix it.

6. Test results

The test runs the relay against a mock ActiveCampaign that answers 429 three times, 503 once and rejects one contact with 422, with retries sped up to a one-second base:

7. Who retries for how long

A relay retry window similar to Shopify's own is deliberate: if ActiveCampaign is down for longer than a few hours, you want dead letters and an alert, not a queue that silently grows for a day. Point a keyword monitor at /health and alert when the dead count rises or the oldest pending job stops moving; after an outage, a single query replays them:

sqlite3 relay.db "UPDATE outbox SET status = 'pending', attempts = 0, next_at = 0 WHERE status = 'dead'"

8. The ActiveCampaign side

• Plan. ActiveCampaign's pricing page lists custom event tracking on Plus, Pro and Enterprise, not Starter. API access and webhooks are on every plan. Third-party trackers put the four tiers at about $15, $49, $79 and $145 a month billed yearly for 1,000 contacts; the official page shows prices only after you configure a plan.
• Setup. Create two custom contact fields (lifetime value, order count) and note their IDs. Turn on event tracking under Website, Site Tracking, and copy the account ID and event key. Event names can be up to 32 characters.
• Automation. Start an automation on the category_abandoned event, branch on the event data (the collection title), wait, and send the category email. Segment newsletters with the lifetime value field.
• Two APIs, two styles. The v3 API takes JSON with an Api-Token header; the event endpoint at trackcmp.net/event takes a form-encoded body with the account ID and event key and returns {"success": 1}. The relay checks that flag, because a rejected event isn't an HTTP error.

9. Our opinion: trade-offs

• Start with Flow and the native integration. They're free, hosted and supported. If a workflow on Order paid with one HTTP request does the job, a relay is overhead.
• Build the relay when the data model is yours. Category-level intent, lifetime value net of refunds, or rules that remember past events are where a hundred lines of your own code beat any number of Flow steps.
• Keep the relay boring. One process, one SQLite file, one sender. It handles thousands of orders a day; reach for a message broker only when one sender at 4 requests a second becomes the bottleneck.
• Why ActiveCampaign for this. It accepts custom events and arbitrary contact fields over a plain API, documents its rate limit, and lets automations branch on event data. That's what makes a custom signal usable by the marketing team without another developer request.

Related: an n8n abandoned cart workflow for Shopify, n8n vs Make for e-commerce automation and monitoring with UptimeRobot.

10. Tools

Top picks

Tool	Role	Why it fits
ActiveCampaign	Email and marketing automation with an open API	Custom events on Plus and up, custom contact fields through contact sync, a documented 5 requests per second limit with Retry-After, and a native Shopify integration for orders and carts so your relay only adds what's missing.

Alternatives: free and open-source tools

We don't earn anything from these links.

• Shopify Flow: free on every current plan; the HTTP action needs Grow or higher.
• Shopify Web Pixels API: the standard events and the sandbox your custom pixel runs in.
• ActiveCampaign developer reference: contact sync, event tracking and rate limits.
• SQLite: the relay's whole storage layer, included with Python.

11. Pitfalls

• Calling the API inside the webhook handler. One slow ActiveCampaign response and Shopify's 5-second timeout counts a failure; eight in a row and an Admin API subscription is deleted.
• Verifying HMAC on parsed JSON. The signature covers the raw body. Re-serialized JSON won't match.
• Sending increments. "Add $50 to LTV" plus at-least-once delivery equals wrong totals. Send the recomputed total.
• Sending orders twice. If the native integration syncs orders, don't post them to the e-commerce API from the relay too.
• Trusting pixel data. It's public. Never let it change money fields or create contacts on its own.
• Retrying everything. A 422 won't succeed on attempt ten. Dead-letter it and alert.
• Forgetting the shared rate limit. Other tools on the same ActiveCampaign key draw from the same 5 requests per second; leave headroom.