Export every lead or reply to CSV, JSON Lines or your own format

A function that streams every lead to a CSV file, replies to JSON Lines, and any listing to a writer of your own, plus a walk you can resume when a long export dies part-way.

Someone asks for “all the leads in a spreadsheet”. The exporter plugin adds outbound.export(listing, { to }): it walks a listing page by page under the rate budget and writes each record as it arrives, so a large export never sits in memory. It writes .csv and .jsonl itself. For any other format you pass a writer, and the package adds no dependency for it.

The exporter opens its file fresh on every call and takes no cursor. It cannot continue a file it started earlier. This guide shows the plain export first, then a walk of your own that can resume.

Before you start

• Node 22 or later.
• The code below runs on the in-memory adapter, so it needs no vendor key. Each sample on this page is a file in the repository that the test suite runs; the output shown is what it printed. The samples write under the system’s temporary folder and print the file contents.
• For real data: an Instantly, Smartlead or HeyReach API key. A read-only key is enough. No method of the SDK writes to a vendor.

What you get

leads.csv: 5 rows
ref,campaignRef,email,firstName,lastName,company,status
memory:acme:lead_1,memory:acme:cmp_founders,ana@example.com,Ana,Ruiz,"Northwind, Inc.",active
memory:acme:lead_2,memory:acme:cmp_founders,ben@example.com,Ben,Okafor,"The ""Good"" Company",replied
memory:acme:lead_3,memory:acme:cmp_founders,caro@example.com,Caro,,,active
memory:acme:lead_4,memory:acme:cmp_agencies,dev@example.com,Dev,Shah,Lumen Studio,active
memory:acme:lead_5,memory:acme:cmp_agencies,eli@example.com,Eli,Brandt,Lumen Studio,bounced

replies.jsonl: 2 rows
columns: provider, ref, schemaVersion, sourceId, workspaceId, bodyHtml, bodyText, campaignRef, channel, leadRef, senderRef, sentAt, stepIndex, subject, threadId, variantKey, category, categorySource, direction, isAuto
{"provider":"memory","ref":"memory:acme:msg_1","schemaVersion":1,"sourceId":"msg_1","workspaceId":"acme","bodyHtml":null,"bodyText":null,"campaignRef":"memory:acme:cmp_founders","channel":"email","leadRef":"memory:acme:lead_2","senderRef":null,"sentAt":"2026-11-12T10:30:00Z","stepIndex":null,"subject":"Re: Quick question","threadId":null,"variantKey":null,"category":"positive","categorySource":"vendor","direction":"in","isAuto":null}
{"provider":"memory","ref":"memory:acme:msg_2","schemaVersion":1,"sourceId":"msg_2","workspaceId":"acme","bodyHtml":null,"bodyText":null,"campaignRef":"memory:acme:cmp_agencies","channel":"email","leadRef":"memory:acme:lead_4","senderRef":null,"sentAt":"2026-11-13T07:15:00Z","stepIndex":null,"subject":"Out of office","threadId":null,"variantKey":null,"category":"ooo","categorySource":"vendor","direction":"in","isAuto":null}

Write the export

outbound.export takes anything you can iterate with for await, which is what every listAll() returns.

import type { Outbound } from "outbound-sdk";
import type { ExporterExtension, ExportResult } from "outbound-sdk/exporter";

export const LEAD_COLUMNS = [
  "ref",
  "campaignRef",
  "email",
  "firstName",
  "lastName",
  "company",
  "status",
];

/** Every lead on the instance, one row each, to a `.csv` or `.jsonl` file. */
export const exportLeads = (
  outbound: Outbound & ExporterExtension,
  file: string
): Promise<ExportResult> =>
  outbound.export(outbound.leads.listAll(), {
    columns: LEAD_COLUMNS,
    to: file,
  });

/** Every reply, with every field the record has except `raw`. */
export const exportReplies = (
  outbound: Outbound & ExporterExtension,
  file: string
): Promise<ExportResult> =>
  outbound.export(outbound.replies.listAll(), { to: file });

• The extension picks the format. .csv gets a header row and the usual CSV quoting: a value with a comma, a quote or a line break is quoted, and quotes are doubled. .jsonl is one JSON object per line. Anything else is refused.
• Name the columns. Without columns, they are the fields of the first record, minus raw. That is every field the record has, as the reply export above shows. A missing value is an empty cell in CSV and null in JSON Lines.
• raw is left out unless you pass includeRaw: true. It is the vendor’s whole payload and holds everything the normalized fields were taken from.
• An object or array in a CSV cell (a lead’s custom fields, its tags) is written as JSON text.

The result is rows and columns. It says nothing about whether the listing was complete: see the cap.

Try it without a key

// One client with two campaigns, five leads and two replies. One company
// name holds a comma and one a quote, to show what the CSV writer does.
import type { MemoryData } from "outbound-sdk/memory";

export const seed: MemoryData = {
  campaigns: [
    { id: "cmp_founders", name: "SaaS founders" },
    { id: "cmp_agencies", name: "Agency owners" },
  ],
  leads: [
    {
      campaignId: "cmp_founders",
      company: "Northwind, Inc.",
      email: "ana@example.com",
      firstName: "Ana",
      id: "lead_1",
      lastName: "Ruiz",
    },
    {
      campaignId: "cmp_founders",
      company: 'The "Good" Company',
      email: "ben@example.com",
      firstName: "Ben",
      id: "lead_2",
      lastName: "Okafor",
      status: "replied",
    },
    {
      campaignId: "cmp_founders",
      email: "caro@example.com",
      firstName: "Caro",
      id: "lead_3",
    },
    {
      campaignId: "cmp_agencies",
      company: "Lumen Studio",
      email: "dev@example.com",
      firstName: "Dev",
      id: "lead_4",
      lastName: "Shah",
    },
    {
      campaignId: "cmp_agencies",
      company: "Lumen Studio",
      email: "eli@example.com",
      firstName: "Eli",
      id: "lead_5",
      lastName: "Brandt",
      status: "bounced",
    },
  ],
  messages: [
    {
      campaignId: "cmp_founders",
      category: "positive",
      direction: "in",
      id: "msg_1",
      leadId: "lead_2",
      sentAt: "2026-11-12T10:30:00Z",
      subject: "Re: Quick question",
    },
    {
      campaignId: "cmp_agencies",
      category: "ooo",
      direction: "in",
      id: "msg_2",
      leadId: "lead_4",
      sentAt: "2026-11-13T07:15:00Z",
      subject: "Out of office",
    },
  ],
};

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";

import { exportLeads, exportReplies } from "./export.ts";
import { seed } from "./seed.ts";

const outbound = createOutbound({
  adapter: memory({ data: seed, workspaceId: "acme" }),
  plugins: [exporter()],
});

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));
const csv = path.join(dir, "leads.csv");
const jsonl = path.join(dir, "replies.jsonl");

const leads = await exportLeads(outbound, csv);

console.log(`leads.csv: ${leads.rows} rows`);

console.log(await readFile(csv, "utf-8"));

const replies = await exportReplies(outbound, jsonl);

console.log(`replies.jsonl: ${replies.rows} rows`);

console.log(`columns: ${replies.columns.join(", ")}`);

console.log(await readFile(jsonl, "utf-8"));

await rm(dir, { recursive: true });

That is the script that printed the two files above.

Point it at your vendor

Swap the adapter. Nothing else changes.

import { createOutbound } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { instantly } from "outbound-sdk/instantly";
import { redact } from "outbound-sdk/redact";

// Reads INSTANTLY_API_KEY. Swap the import for smartlead() or heyreach().
export const outbound = createOutbound({
  adapter: instantly({ workspaceId: "acme" }),
  // Drop redact() only when the file is allowed to hold personal data.
  plugins: [redact({ strategy: "hash" }), exporter()],
});

Mask personal data

List redact before exporter. It is then the outermost plugin, and every record is masked before the exporter sees it. With strategy: "hash" the same value always becomes the same token, so rows can still be counted and joined without the file holding an address.

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";
import { redact } from "outbound-sdk/redact";

import { exportLeads } from "./export.ts";
import { seed } from "./seed.ts";

// `redact` is listed first, so it is outermost and nothing reaches the file
// unmasked. `hash` keeps a stable token per value: rows can still be joined.
const outbound = createOutbound({
  adapter: memory({ data: seed, workspaceId: "acme" }),
  plugins: [redact({ strategy: "hash" }), exporter()],
});

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));
const file = path.join(dir, "leads.csv");

await exportLeads(outbound, file);

console.log(await readFile(file, "utf-8"));

await rm(dir, { recursive: true });

ref,campaignRef,email,firstName,lastName,company,status
memory:acme:lead_1,memory:acme:cmp_founders,redacted:8e43ca37701228e7,redacted:dea210f058b407db,redacted:7c8c32faff7cede0,redacted:79e871f6196f16fe,active
memory:acme:lead_2,memory:acme:cmp_founders,redacted:f871a76fb7b15231,redacted:030d756286e59f22,redacted:95d5798c5815192d,redacted:a041f470be21d010,replied
memory:acme:lead_3,memory:acme:cmp_founders,redacted:1e91676e0a6d94ac,redacted:ccb05dc4bbe84de7,,,active
memory:acme:lead_4,memory:acme:cmp_agencies,redacted:eb2b6c0d061bbd5c,redacted:9c24f45a7ea9e466,redacted:4b48c39df4ae16d3,redacted:95cb02fc236fa8e9,active
memory:acme:lead_5,memory:acme:cmp_agencies,redacted:e245111f6151dbb3,redacted:8d6d4f52ee2db641,redacted:f4cdc96ba6078192,redacted:95cb02fc236fa8e9,bounced

Refs, campaign and status are untouched. The last two rows share a company and so share a token. redact always drops raw, whatever includeRaw says.

Another format through a writer

A RowWriter is two functions: write(row) and close(). This one writes gzipped JSON Lines with Node’s own zlib. A Parquet or DuckDB writer has the same shape, with that library’s calls inside.

import { once } from "node:events";
import { createWriteStream } from "node:fs";
import { finished } from "node:stream/promises";
import { createGzip } from "node:zlib";

import type { RowWriter } from "outbound-sdk/exporter";

/**
 * Gzipped JSON Lines. Each row holds the export's columns and nothing else,
 * so the writer writes the row as it is.
 */
export const gzipJsonl = (file: string): RowWriter => {
  const gzip = createGzip();
  const done = finished(gzip.pipe(createWriteStream(file)));

  return {
    close: async () => {
      gzip.end();
      await done;
    },
    write: async (row) => {
      const line = JSON.stringify(row);

      // Wait when the stream is full, so a large export never sits in memory.
      if (!gzip.write(`${line}\n`)) {
        await once(gzip, "drain");
      }
    },
  };
};

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { gunzipSync } from "node:zlib";

import { createOutbound } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";

import { gzipJsonl } from "./gzip-writer.ts";
import { seed } from "./seed.ts";

const outbound = createOutbound({
  adapter: memory({ data: seed, workspaceId: "acme" }),
  plugins: [exporter()],
});

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));
const file = path.join(dir, "leads.jsonl.gz");

const result = await outbound.export(outbound.leads.listAll(), {
  columns: ["ref", "email", "status"],
  to: gzipJsonl(file),
});

console.log(`${result.rows} rows handed to the writer`);

console.log(gunzipSync(await readFile(file)).toString("utf-8"));

await rm(dir, { recursive: true });

5 rows handed to the writer
{"ref":"memory:acme:lead_1","email":"ana@example.com","status":"active"}
{"ref":"memory:acme:lead_2","email":"ben@example.com","status":"replied"}
{"ref":"memory:acme:lead_3","email":"caro@example.com","status":"active"}
{"ref":"memory:acme:lead_4","email":"dev@example.com","status":"active"}
{"ref":"memory:acme:lead_5","email":"eli@example.com","status":"bounced"}

columns and includeRaw apply to a writer as they do to a file: each row it is handed holds exactly the export’s columns, and raw only when you asked for it. close() is called once, also when the listing fails or is empty, so a writer can release its file.

Resume a long export yourself

The exporter cannot resume. It opens the file for writing, which empties it, and it has no cursor option of its own. For an export long enough to be interrupted, walk the listing yourself: listAll() exposes a cursor that moves each time a whole page has been handed out, and accepts one to start from.

import { appendFile, readFile } from "node:fs/promises";

import type { Lead, Outbound } from "outbound-sdk";

/** Where an unfinished export stopped. Keep it somewhere that survives a crash. */
export interface ExportState {
  cursor?: string | undefined;
  done: boolean;
}

const refsIn = async (file: string): Promise<Set<string>> => {
  const lines = await readFile(file, "utf-8").catch(() => "");

  return new Set(
    lines
      .split("\n")
      .filter((line) => line !== "")
      .map((line) => String(JSON.parse(line).ref))
  );
};

/**
 * Appends leads to a `.jsonl` file, starting from `state.cursor`. The cursor
 * is saved each time a page has been written, so a run that dies restarts on
 * the page it was writing. That page is read again: delivery is at least
 * once, and the refs already in the file are skipped.
 */
export const exportLeadsFrom = async (
  outbound: Outbound,
  file: string,
  state: ExportState,
  onRow?: (lead: Lead) => void
): Promise<{ skipped: number; written: number }> => {
  const seen = await refsIn(file);
  const walk = outbound.leads.listAll({ cursor: state.cursor });
  let skipped = 0;
  let written = 0;

  for await (const lead of walk) {
    // The walker moves its cursor once a whole page has been handed out, and
    // every row of that page has been appended by now.
    state.cursor = walk.cursor;

    if (seen.has(lead.ref)) {
      skipped += 1;
    } else {
      const row = { email: lead.email, ref: lead.ref, status: lead.status };

      await appendFile(file, `${JSON.stringify(row)}\n`);
      seen.add(lead.ref);
      written += 1;
      onRow?.(lead);
    }
  }

  state.cursor = undefined;
  state.done = true;

  return { skipped, written };
};

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound } from "outbound-sdk";
import { memory } from "outbound-sdk/memory";

import { exportLeadsFrom } from "./resumable.ts";
import type { ExportState } from "./resumable.ts";
import { seed } from "./seed.ts";

// Two leads a page, so five leads are three pages.
const outbound = createOutbound({
  adapter: memory({ data: seed, pageSize: 2, workspaceId: "acme" }),
});

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));
const file = path.join(dir, "leads.jsonl");
const state: ExportState = { done: false };
let rows = 0;

// The first attempt dies after its third row, in the middle of page two.
try {
  await exportLeadsFrom(outbound, file, state, () => {
    rows += 1;

    if (rows === 3) {
      throw new Error("process killed");
    }
  });
} catch (error) {
  console.log(`first attempt: ${String(error)}, ${rows} rows written`);
}

console.log(`a cursor was saved: ${state.cursor !== undefined}`);

// The second attempt starts from the saved cursor. Page two is read again.
const resumed = await exportLeadsFrom(outbound, file, state);

console.log(
  `second attempt: ${resumed.written} written, ${resumed.skipped} already in the file, done ${state.done}`
);

console.log(await readFile(file, "utf-8"));

await rm(dir, { recursive: true });

first attempt: Error: process killed, 3 rows written
a cursor was saved: true
second attempt: 2 written, 1 already in the file, done true
{"email":"ana@example.com","ref":"memory:acme:lead_1","status":"active"}
{"email":"ben@example.com","ref":"memory:acme:lead_2","status":"replied"}
{"email":"caro@example.com","ref":"memory:acme:lead_3","status":"active"}
{"email":"dev@example.com","ref":"memory:acme:lead_4","status":"active"}
{"email":"eli@example.com","ref":"memory:acme:lead_5","status":"bounced"}

The first attempt died in the middle of page two. The saved cursor points at the start of page two, so the second attempt reads that page again and the one row already in the file is skipped. This is at-least-once delivery: the page being written when the run died is always replayed, so the writer must drop what it already has. Here that is a set of the refs in the file. For a file too large to read back, keep the refs of the last page beside the cursor.

Keep the cursor somewhere that survives the crash (a row in a table, a file next to the export). A cursor belongs to the listing that issued it: the same collection, the same adapter and workspace.

What each vendor can serve

The same export, run on the in-memory adapter limited to what each vendor’s adapter declares. This shows which calls a vendor’s adapter would refuse. It does not show what that vendor’s API returns.

import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound, OutboundError } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";

import { asVendor, VENDORS } from "../_shared/as-vendor.ts";
import type { Vendor } from "../_shared/as-vendor.ts";
import { exportLeads, LEAD_COLUMNS } from "./export.ts";
import { seed } from "./seed.ts";

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));

const exportAs = async (vendor: Vendor): Promise<string> => {
  const outbound = createOutbound({
    adapter: asVendor(vendor, seed, "acme"),
    plugins: [exporter()],
  });

  try {
    const result = await exportLeads(outbound, path.join(dir, `${vendor}.csv`));

    return `${vendor}: ok, ${result.rows} rows`;
  } catch (error) {
    if (!OutboundError.is(error)) {
      throw error;
    }

    // No list of every lead: export one campaign's leads at a time.
    const campaign = outbound.campaign("cmp_founders");

    const result = await outbound.export(campaign.leads.listAll(), {
      columns: LEAD_COLUMNS,
      to: path.join(dir, `${vendor}.cmp_founders.csv`),
    });

    return `${vendor}: ${error.code}: ${error.message}\n${vendor}: one campaign, ${result.rows} rows`;
  }
};

const lines = await Promise.all(VENDORS.map(exportAs));

console.log(lines.join("\n"));

await rm(dir, { recursive: true });

instantly: ok, 5 rows
smartlead: ok, 5 rows
heyreach: Unsupported: Listing leads is not available: the heyreach adapter does not support it (leads.list)
heyreach: one campaign, 3 rows

HeyReach is refused because it has no list of every lead: leads exist only inside a campaign or a saved list. Export outbound.campaign(id).leads.listAll() once per campaign, as the fallback above does for one. A person in two campaigns is then two rows.

What it costs in vendor requests

An export costs exactly what the listing costs: one walk of every page. outbound.estimate({ action: "leads.list", records }) gives the expected request count where the adapter declares a cost. To spend less, pass filters to the listing (campaignId, since) and max.

When it cannot answer

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound, OutboundError } from "outbound-sdk";
import type { OutboundPlugin } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";

import { exportLeads } from "./export.ts";
import { seed } from "./seed.ts";

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));

const outbound = createOutbound({
  adapter: memory({ data: seed, pageSize: 2, workspaceId: "acme" }),
  plugins: [exporter()],
});

// 1. A file that is neither .csv nor .jsonl is refused.
try {
  await exportLeads(outbound, path.join(dir, "leads.parquet"));
} catch (error) {
  if (OutboundError.is(error)) {
    console.log("wrong extension:", error.code, "-", error.message);
  }
}

// 2. The vendor fails on the second page. The error reaches the caller, and
//    the file is left holding the rows written before it.
let pages = 0;

const failsOnPageTwo: OutboundPlugin = {
  name: "fails-on-page-two",
  wrap: (operation, next, info) => {
    pages += 1;

    if (pages === 2) {
      throw new OutboundError("Provider", "The vendor responded 503", {
        provider: info.provider,
        status: 503,
      });
    }

    return next(operation);
  },
};

const flaky = createOutbound({
  adapter: memory({ data: seed, pageSize: 2, workspaceId: "acme" }),
  plugins: [exporter(), failsOnPageTwo],
});

const partial = path.join(dir, "partial.csv");

try {
  await exportLeads(flaky, partial);
} catch (error) {
  if (OutboundError.is(error)) {
    const written = await readFile(partial, "utf-8");
    const lines = written.trim().split("\n");

    console.log(
      `vendor failed: ${error.code}, file left with ${lines.length - 1} of 5 rows`
    );
  }
}

// 3. `max` stops the walk early. The exporter does not know: ask the listing.
const capped = outbound.leads.listAll({ max: 3 });

const result = await outbound.export(capped, {
  columns: ["ref"],
  to: path.join(dir, "capped.csv"),
});

console.log(`capped: ${result.rows} rows, truncated ${capped.truncated}`);

// 4. A listing with no records still writes the file. Name the columns, or
//    the header is empty: they are otherwise taken from the first record.
const empty = path.join(dir, "empty.csv");

const none = await outbound.export(
  outbound.replies.listAll({ category: "unsubscribe" }),
  { columns: ["ref", "category"], to: empty }
);

console.log(
  `empty: ${none.rows} rows, file is ${JSON.stringify(await readFile(empty, "utf-8"))}`
);

await rm(dir, { recursive: true });

wrong extension: Invalid - export() writes .csv or .jsonl files; pass a writer for other formats
vendor failed: Provider, file left with 2 of 5 rows
capped: 3 rows, truncated true
empty: 0 rows, file is "ref,category\n"

Limits

• No resume, no append: every export() call starts the file again.
• Columns are fixed by the first record or by columns. A field that only later records have is not added.
• A reply or message has no body unless the listing asked for it with include: ["body"], which can cost extra requests per record on some vendors.
• The exporter writes what the listing yields. On an unverified adapter, compare a few rows with the vendor’s own app before sending the file on.