Sync CRM data on a schedule

Uses: Python · TypeScript · CLI

The most common Engini job isn’t an agent at all - it’s a scheduled sync. One connector reads, another writes, and the whole thing is two tool calls plus error handling.

Discover both sides

Slugs differ per connector, so resolve them once and reuse. (Discovery is shown in the CLI - it’s a one-time setup step; the same lookups are client.tools.get(...) / client.tools.get_one(...) in the SDKs.)

engini tools list --application salesforce | jq -r '.[] | "\(.tool_slug)\t\(.description)"'
engini tools list --application hubspot    | jq -r '.[] | "\(.tool_slug)\t\(.description)"'

Read the write-side contract carefully before you build the payload - it tells you which fields are required and whether the tool supports filters or paging:

engini tools get <write-tool-slug> | jq '{input_schema, supports_filters, supports_top_offset}'

The sync

from engini import Engini
from engini.errors import EnginiRateLimitError, EnginiToolExecutionError
import time

client = Engini()   # reads ENGINI_API_KEY

READ  = "<read-tool-slug>"
WRITE = "<write-tool-slug>"

def with_retry(fn, attempts=5):
    """Engini's SDKs do not retry automatically - own it explicitly."""
    for i in range(attempts):
        try:
            return fn()
        except EnginiRateLimitError:
            time.sleep(2 ** i)          # honour Retry-After when you have it
    raise RuntimeError("still rate-limited after retries")

since = "2026-08-01"
source = with_retry(lambda: client.tools.execute(READ, {"modifiedSince": since}))
records = source.output or []
print(f"{len(records)} records to sync")

failed = []
for r in records:
    try:
        with_retry(lambda: client.tools.execute(WRITE, {
            "email": r.get("Email"),
            "firstName": r.get("FirstName"),
            "lastName": r.get("LastName"),
        }))
    except EnginiToolExecutionError as e:
        failed.append((r.get("Id"), e.error_message, e.history_id))

if failed:
    for rid, msg, hid in failed:
        print(f"FAILED {rid}: {msg} (history {hid})")
    raise SystemExit(1)

import { Engini, EnginiRateLimitError, EnginiToolExecutionError } from "@engini/sdk";

const client = new Engini();   // reads ENGINI_API_KEY
const READ = "<read-tool-slug>";
const WRITE = "<write-tool-slug>";

// the SDKs do not retry automatically - own it explicitly
async function withRetry<T>(fn: () => Promise<T>, attempts = 5): Promise<T> {
  for (let i = 0; i < attempts; i++) {
    try {
      return await fn();
    } catch (e) {
      if (!(e instanceof EnginiRateLimitError)) throw e;
      await new Promise((r) => setTimeout(r, 2 ** i * 1000));
    }
  }
  throw new Error("still rate-limited after retries");
}

const source = await withRetry(() => client.tools.execute(READ, { modifiedSince: "2026-08-01" }));
const records = (source.output as any[]) ?? [];
console.log(`${records.length} records to sync`);

const failed: Array<[unknown, string | null, number | null]> = [];
for (const r of records) {
  try {
    await withRetry(() =>
      client.tools.execute(WRITE, {
        email: r.Email, firstName: r.FirstName, lastName: r.LastName,
      }),
    );
  } catch (e) {
    if (e instanceof EnginiToolExecutionError) failed.push([r.Id, e.errorMessage, e.historyId]);
    else throw e;
  }
}

if (failed.length) {
  for (const [id, msg, hid] of failed) console.error(`FAILED ${id}: ${msg} (history ${hid})`);
  process.exit(1);
}

Three things that matter more than the sync logic:

Partial failure is the normal case. One bad record shouldn’t abort the run - collect failures, finish the batch, then exit non-zero so your scheduler notices.
Keep history_id for every failure. It’s how you or support reconstruct exactly what was sent.
Retries are yours. Neither SDK retries automatically, by design - see errors & pagination.

Large source sets

If the read tool reports supports_top_offset, page it rather than pulling everything into memory. (The flag means top and offset together - a tool can support SelectTop alone, capping results without paging; check the input_schema for which of SelectTop / SelectOffset it takes. Salesforce, for instance, takes top but not offset.)

offset, page = 0, 500
while True:
    batch = client.tools.execute(READ, {"modifiedSince": since}, top=page, offset=offset)
    rows = batch.output or []
    if not rows:
        break
    handle(rows)
    offset += len(rows)

let offset = 0;
const page = 500;
for (;;) {
  const batch = await client.tools.execute(READ, { modifiedSince: since }, { top: page, offset });
  const rows = (batch.output as any[]) ?? [];
  if (!rows.length) break;
  handle(rows);
  offset += rows.length;
}

Same contract as the catalog paging recipe, applied to tool output.

Scheduling it

GitHub Actions

on:
  schedule:
    - cron: "0 2 * * *"
jobs:
  sync:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install engini
      - run: python sync.py
        env:
          ENGINI_API_KEY: ${{ secrets.ENGINI_API_KEY }}

Add a pre-flight check so a dead credential fails loudly instead of silently syncing nothing:

engini whoami --quiet || { echo "Engini auth failed"; exit 1; }
engini connections list | jq -e '.[] | select(.is_alive == false)' && echo "::warning::unhealthy connection"

From the CLI alone

For a simple one-way push, you may not need Python at all:

engini tools call <read-tool-slug> --args '{"modifiedSince":"2026-08-01"}' --json > src.json
jq -c '.output[]' src.json | while read -r row; do
  engini tools call <write-tool-slug> --args "$row" --quiet || echo "failed: $row" >> failures.log
done

Debug a failed tool call - reading executionInfo when the CRM rejects a record
Run Engini in CI/CD - exit codes, secrets, and pipeline patterns
Scripted automation - the minimal version of this pattern

Start here

Connect apps

Run tools

Build agents

Automate

Discover both sides

The sync

Large source sets

Scheduling it

From the CLI alone

​Discover both sides

​The sync

​Large source sets

​Scheduling it

​From the CLI alone

​Related

Discover both sides

The sync

Large source sets

Scheduling it

From the CLI alone

Related