> ## Documentation Index
> Fetch the complete documentation index at: https://docs.engini.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Sync CRM data on a schedule

> Pull records from one system and push them to another, on a cron, with no LLM in the loop.

<Info>**Uses:** Python · TypeScript · CLI</Info>

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.)

```bash theme={null}
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:

```bash theme={null}
engini tools get <write-tool-slug> | jq '{input_schema, supports_filters, supports_top_offset}'
```

## The sync

<CodeGroup>
  ```python Python theme={null}
  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)
  ```

  ```typescript TypeScript theme={null}
  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);
  }
  ```
</CodeGroup>

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](/sdk/errors-and-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.)

<CodeGroup>
  ```python Python theme={null}
  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)
  ```

  ```typescript TypeScript theme={null}
  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;
  }
  ```
</CodeGroup>

Same contract as the [catalog paging recipe](/examples/pagination), applied to tool output.

## Scheduling it

```yaml GitHub Actions theme={null}
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:

```bash theme={null}
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:

```bash theme={null}
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
```

## Related

* [Debug a failed tool call](/examples/debug-failed-execution) - reading `executionInfo` when the CRM rejects a record
* [Run Engini in CI/CD](/examples/ci-automation) - exit codes, secrets, and pipeline patterns
* [Scripted automation](/examples/scripted-automation) - the minimal version of this pattern