Reverse ETL from a Warehouse to Clay with TypeScript

Pull a segment (simulating a warehouse query result), enrich it via Clay's real batch endpoints, and format the result for loading back into a warehouse table — the JSONL batch upload is a genuine fit for this pattern, not a separate warehouse-specific Clay feature.

What you will build

An Express service exposing POST /sync-warehouse-segment, which takes a segment of accounts (as a warehouse query result would return), batch-enriches them via Clay, and returns activation-ready records.

POST /sync-warehouse-segment (warehouseRows: [{account_id, domain}])
    ↓
POST /routines/{id}/run-batch/upload-url   → presigned S3 PUT URL + file_id
    ↓
PUT <upload_url>                            (JSONL: one Enrich Company input per domain)
    ↓
POST /routines/{id}/run-batch/start         (file_id) → routine_run_id
    ↓
GET  /routines/run-batch/{run_id}/results   (poll until complete → result_url)
    ↓
GET  <result_url>                           (separate JSONL fetch — the real output)
    ↓
activationRows[] (ready to load back into a warehouse table)

POST /sync-warehouse-segment (warehouseRows: [{account_id, domain}])
    ↓
POST /routines/{id}/run-batch/upload-url   → presigned S3 PUT URL + file_id
    ↓
PUT <upload_url>                            (JSONL: one Enrich Company input per domain)
    ↓
POST /routines/{id}/run-batch/start         (file_id) → routine_run_id
    ↓
GET  /routines/run-batch/{run_id}/results   (poll until complete → result_url)
    ↓
GET  <result_url>                           (separate JSONL fetch — the real output)
    ↓
activationRows[] (ready to load back into a warehouse table)

AI Prompt

Implement a TypeScript/Express service that performs a reverse-ETL
style sync: take a segment of accounts (as if pulled from a data
warehouse), enrich them via Clay's real batch routine endpoints, and
return records shaped for loading back into a warehouse table.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- There is no warehouse-specific Clay connector or reverse-ETL
  endpoint. Extracting from and loading back into the warehouse are
  both the caller's own responsibility (via their warehouse's own
  client library), entirely outside Clay -- Clay's only role is
  enriching the segment via its ordinary batch endpoints.
- The batch flow is the same 4 real steps used for bulk CSV
  enrichment: (1) POST /routines/{id}/run-batch/upload-url returns
  {"upload_url","file_id"}; (2) PUT JSONL (one
  {"id":"<domain>","inputs":{"Company Identifier":"<domain>"}} per
  line) to upload_url; (3) POST /routines/{id}/run-batch/start with
  {"file_id"} returns {"routine_run_id","status":"in_progress"}; (4)
  poll GET /routines/run-batch/{routine_run_id}/results. The
  completed response does NOT inline results, only a separate
  "result_url" that must be fetched for the real per-row JSONL.
- Each row's real data is under result["Enrich Company"] (industry,
  employee_count, annual_revenue).
- Cap the segment at <=5 rows per request -- each costs a real Clay
  credit (1 credit/row for Enrich Company).
- Run the verification step below before finishing.

Implement a TypeScript/Express service that performs a reverse-ETL
style sync: take a segment of accounts (as if pulled from a data
warehouse), enrich them via Clay's real batch routine endpoints, and
return records shaped for loading back into a warehouse table.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- There is no warehouse-specific Clay connector or reverse-ETL
  endpoint. Extracting from and loading back into the warehouse are
  both the caller's own responsibility (via their warehouse's own
  client library), entirely outside Clay -- Clay's only role is
  enriching the segment via its ordinary batch endpoints.
- The batch flow is the same 4 real steps used for bulk CSV
  enrichment: (1) POST /routines/{id}/run-batch/upload-url returns
  {"upload_url","file_id"}; (2) PUT JSONL (one
  {"id":"<domain>","inputs":{"Company Identifier":"<domain>"}} per
  line) to upload_url; (3) POST /routines/{id}/run-batch/start with
  {"file_id"} returns {"routine_run_id","status":"in_progress"}; (4)
  poll GET /routines/run-batch/{routine_run_id}/results. The
  completed response does NOT inline results, only a separate
  "result_url" that must be fetched for the real per-row JSONL.
- Each row's real data is under result["Enrich Company"] (industry,
  employee_count, annual_revenue).
- Cap the segment at <=5 rows per request -- each costs a real Clay
  credit (1 credit/row for Enrich Company).
- Run the verification step below before finishing.

Prerequisites

Node.js 18+ and TypeScript
A Clay Public API key (clay_scoped_...)
The routine ID for Enrich Company, discovered once via clay routines list

1. Create the project

mkdir warehouse-reverse-etl && cd warehouse-reverse-etl
npm init -y
npm install express
npm install

mkdir warehouse-reverse-etl && cd warehouse-reverse-etl
npm init -y
npm install express
npm install -D typescript @types/express @types/node
npx tsc --init

2. Discover the real schema

Reuses the batch mechanics and Enrich Company result shape from the waterfall-enrichment and bulk-CSV-enrichment examples.

3. Configure credentials

CLAY_API_KEY=
CLAY_ROUTINE_ID_ENRICH_COMPANY=

In a CI/sandbox test environment, these are typically provided as pre-configured secrets; in your own deployment, set them as real environment variables or via your platform's secrets manager.

4. Implement the service

import express, { Request, Response as ExpressResponse } from "express";

const CLAY_API_BASE = "https://api.clay.com/public/v0";
const CLAY_API_KEY = process.env.CLAY_API_KEY;
const ENRICH_COMPANY_ROUTINE_ID = process.env.CLAY_ROUTINE_ID_ENRICH_COMPANY;

if (!CLAY_API_KEY) throw new Error("CLAY_API_KEY env var is required");
if (!ENRICH_COMPANY_ROUTINE_ID) throw new Error("CLAY_ROUTINE_ID_ENRICH_COMPANY env var is required");

async function clayFetch(path: string, init: RequestInit = {}): Promise<globalThis
  return fetch(`${CLAY_API_BASE}${path}`, {
    ...init,
    headers: {
      "clay-api-key": CLAY_API_KEY as string,
      "Content-Type": "application/json",
      ...(init.headers || {}),
    },
  });
}

interface EnrichedCompany {
  industry?: string;
  employee_count?: number;
  annual_revenue?: string;
}

interface BatchRow {
  id: string;
  status: string;
  result?: { "Enrich Company"?: EnrichedCompany };
}

// Same batch mechanics used in bulk-csv-enrichment: upload-url, PUT
// JSONL, start, poll, then a SEPARATE fetch to result_url (the completed
// run-batch/results response never inlines the data). Clay has no
// warehouse-specific connector -- the "reverse ETL" framing is entirely
// about what the caller does before and after this same generic batch call.
async function batchEnrichDomains(domains: string[]): Promise<BatchRow[]> {
  const uploadUrlRes = await clayFetch(`/routines/${ENRICH_COMPANY_ROUTINE_ID}/run-batch/upload-url`, {
    method: "POST",
    body: JSON.stringify({}),
  });
  if (!uploadUrlRes.ok) {
    throw new Error(`upload-url failed: ${uploadUrlRes.status} ${await uploadUrlRes.text()}`);
  }
  const { upload_url, file_id } = (await uploadUrlRes.json()) as { upload_url: string; file_id:

  const jsonl = domains
    .map((domain) => JSON.stringify({ id: domain, inputs: { "Company Identifier": domain } }))
    .join("\n");
  const putRes = await fetch(upload_url, {
    method: "PUT",
    headers: { "Content-Type": "application/octet-stream" },
    body: jsonl,
  });
  if (!putRes.ok) throw new Error(`file upload failed: ${putRes.status}`);

  const startRes = await clayFetch(`/routines/${ENRICH_COMPANY_ROUTINE_ID}/run-batch/start`, {
    method: "POST",
    body: JSON.stringify({ file_id }),
  });
  if (!startRes.ok) {
    throw new Error(`run-batch start failed: ${startRes.status} ${await startRes.text()}`);
  }
  const { routine_run_id } = (await startRes.json()) as { routine_run_id: string };

  const resultsPath = `/routines/run-batch/${routine_run_id}/results`;
  let resultUrl: string | null = null;
  for (let attempt = 0; attempt < 15; attempt++) {
    const res = await clayFetch(resultsPath);
    if (!res.ok) throw new Error(`batch results failed: ${res.status} ${await res.text()
    const body = (await res.json()) as { status: string; result_url?: string };
    if (body.status === "complete") {
      resultUrl = body.result_url ?? null;
      break;
    }
    await new Promise((r) => setTimeout(r, 5000));
  }
  if (!resultUrl) throw new Error(`batch run ${routine_run_id} did not complete in time`);

  const outputRes = await fetch(resultUrl);
  if (!outputRes.ok) throw new Error(`fetching result_url failed: ${outputRes.status}`);
  const outputText = await outputRes.text();
  return outputText
    .split("\n")
    .filter((line) => line.trim().length > 0)
    .map((line) => JSON.parse(line) as BatchRow);
}

interface WarehouseRow {
  account_id: string;
  domain: string;
}

const app = express();
app.use(express.json());

// "Reverse ETL": a segment extracted from a warehouse query (simulated here
// as the request body -- a real caller would pull this via their warehouse's
// own client/API, outside Clay) is enriched via Clay, then reshaped into a
// flat record set ready to load back ("activate") into a warehouse table.
app.post("/sync-warehouse-segment", async (req: Request, res: ExpressResponse) => {
  const { warehouseRows } = req.body ?? {};
  if (!Array.isArray(warehouseRows) || warehouseRows.length === 0) {
    return res.status(400).json({ error: "warehouseRows must be a non-empty array of {account_id, domain}" });
  }
  const capped: WarehouseRow[] = warehouseRows.slice(0, 5);

  try {
    const domains = capped.map((r) => r.domain);
    const enrichedRows = await batchEnrichDomains(domains);
    const byDomain = new Map(enrichedRows.map((r) => [r.id, r.result?.["Enrich Company"]]));

    const activationRows = capped.map((row) => {
      const enriched = byDomain.get(row.domain);
      return {
        account_id: row.account_id,
        domain: row.domain,
        industry: enriched?.industry ?? null,
        employee_count: enriched?.employee_count ?? null,
        annual_revenue: enriched?.annual_revenue ?? null,
        enriched_at: new Date().toISOString(),
      };
    });

    res.json({ requested: capped.length, activationRows });
  } catch (err) {
    res.status(502).json({ error: (err as Error).message });
  }
});

const port = Number(process.env.PORT) || 3000;
app.listen(port, () => console.log(`listening on ${port}`));

import express, { Request, Response as ExpressResponse } from "express";

const CLAY_API_BASE = "https://api.clay.com/public/v0";
const CLAY_API_KEY = process.env.CLAY_API_KEY;
const ENRICH_COMPANY_ROUTINE_ID = process.env.CLAY_ROUTINE_ID_ENRICH_COMPANY;

if (!CLAY_API_KEY) throw new Error("CLAY_API_KEY env var is required");
if (!ENRICH_COMPANY_ROUTINE_ID) throw new Error("CLAY_ROUTINE_ID_ENRICH_COMPANY env var is required");

async function clayFetch(path: string, init: RequestInit = {}): Promise<globalThis.Response> {
  return fetch(`${CLAY_API_BASE}${path}`, {
    ...init,
    headers: {
      "clay-api-key": CLAY_API_KEY as string,
      "Content-Type": "application/json",
      ...(init.headers || {}),
    },
  });
}

interface EnrichedCompany {
  industry?: string;
  employee_count?: number;
  annual_revenue?: string;
}

interface BatchRow {
  id: string;
  status: string;
  result?: { "Enrich Company"?: EnrichedCompany };
}

// Same batch mechanics used in bulk-csv-enrichment: upload-url, PUT
// JSONL, start, poll, then a SEPARATE fetch to result_url (the completed
// run-batch/results response never inlines the data). Clay has no
// warehouse-specific connector -- the "reverse ETL" framing is entirely
// about what the caller does before and after this same generic batch call.
async function batchEnrichDomains(domains: string[]): Promise<BatchRow[]> {
  const uploadUrlRes = await clayFetch(`/routines/${ENRICH_COMPANY_ROUTINE_ID}/run-batch/upload-url`, {
    method: "POST",
    body: JSON.stringify({}),
  });
  if (!uploadUrlRes.ok) {
    throw new Error(`upload-url failed: ${uploadUrlRes.status} ${await uploadUrlRes.text()}`);
  }
  const { upload_url, file_id } = (await uploadUrlRes.json()) as { upload_url: string; file_id: string };

  const jsonl = domains
    .map((domain) => JSON.stringify({ id: domain, inputs: { "Company Identifier": domain } }))
    .join("\n");
  const putRes = await fetch(upload_url, {
    method: "PUT",
    headers: { "Content-Type": "application/octet-stream" },
    body: jsonl,
  });
  if (!putRes.ok) throw new Error(`file upload failed: ${putRes.status}`);

  const startRes = await clayFetch(`/routines/${ENRICH_COMPANY_ROUTINE_ID}/run-batch/start`, {
    method: "POST",
    body: JSON.stringify({ file_id }),
  });
  if (!startRes.ok) {
    throw new Error(`run-batch start failed: ${startRes.status} ${await startRes.text()}`);
  }
  const { routine_run_id } = (await startRes.json()) as { routine_run_id: string };

  const resultsPath = `/routines/run-batch/${routine_run_id}/results`;
  let resultUrl: string | null = null;
  for (let attempt = 0; attempt < 15; attempt++) {
    const res = await clayFetch(resultsPath);
    if (!res.ok) throw new Error(`batch results failed: ${res.status} ${await res.text()}`);
    const body = (await res.json()) as { status: string; result_url?: string };
    if (body.status === "complete") {
      resultUrl = body.result_url ?? null;
      break;
    }
    await new Promise((r) => setTimeout(r, 5000));
  }
  if (!resultUrl) throw new Error(`batch run ${routine_run_id} did not complete in time`);

  const outputRes = await fetch(resultUrl);
  if (!outputRes.ok) throw new Error(`fetching result_url failed: ${outputRes.status}`);
  const outputText = await outputRes.text();
  return outputText
    .split("\n")
    .filter((line) => line.trim().length > 0)
    .map((line) => JSON.parse(line) as BatchRow);
}

interface WarehouseRow {
  account_id: string;
  domain: string;
}

const app = express();
app.use(express.json());

// "Reverse ETL": a segment extracted from a warehouse query (simulated here
// as the request body -- a real caller would pull this via their warehouse's
// own client/API, outside Clay) is enriched via Clay, then reshaped into a
// flat record set ready to load back ("activate") into a warehouse table.
app.post("/sync-warehouse-segment", async (req: Request, res: ExpressResponse) => {
  const { warehouseRows } = req.body ?? {};
  if (!Array.isArray(warehouseRows) || warehouseRows.length === 0) {
    return res.status(400).json({ error: "warehouseRows must be a non-empty array of {account_id, domain}" });
  }
  const capped: WarehouseRow[] = warehouseRows.slice(0, 5);

  try {
    const domains = capped.map((r) => r.domain);
    const enrichedRows = await batchEnrichDomains(domains);
    const byDomain = new Map(enrichedRows.map((r) => [r.id, r.result?.["Enrich Company"]]));

    const activationRows = capped.map((row) => {
      const enriched = byDomain.get(row.domain);
      return {
        account_id: row.account_id,
        domain: row.domain,
        industry: enriched?.industry ?? null,
        employee_count: enriched?.employee_count ?? null,
        annual_revenue: enriched?.annual_revenue ?? null,
        enriched_at: new Date().toISOString(),
      };
    });

    res.json({ requested: capped.length, activationRows });
  } catch (err) {
    res.status(502).json({ error: (err as Error).message });
  }
});

const port = Number(process.env.PORT) || 3000;
app.listen(port, () => console.log(`listening on ${port}`));

5. Run

npx tsc
node --env-file=.env dist/index.js

npx tsc
node --env-file=.env dist/index.js

curl -X POST http://localhost:3000/sync-warehouse-segment \
  -H "Content-Type: application/json" \
  -d '{"warehouseRows":[{"account_id":"acct_001","domain":"stripe.com"},{"account_id":"acct_002","domain":"notion.so"}]}'

curl -X POST http://localhost:3000/sync-warehouse-segment \
  -H "Content-Type: application/json" \
  -d '{"warehouseRows":[{"account_id":"acct_001","domain":"stripe.com"},{"account_id":"acct_002","domain":"notion.so"}]}'

6. Verify the result

{
  "requested": 2,
  "activationRows": [
    {"account_id": "acct_001", "domain": "stripe.com", "industry"

{
  "requested": 2,
  "activationRows": [
    {"account_id": "acct_001", "domain": "stripe.com", "industry": "Technology, Information and Internet", "employee_count": 17184, "annual_revenue": "1B-10B", "enriched_at": "2026-09-02T21:23:41.597Z"},
    {"account_id": "acct_002", "domain": "notion.so", "industry": "Software Development", "employee_count": 7768, "annual_revenue": "1B-10B", "enriched_at": "2026-09-02T21:23:41.603Z"}
  ]
}

Both simulated warehouse rows were enriched with real Clay data and reshaped into activation-ready records, preserving the original account_id for the load-back join key.

How it works

"Reverse ETL" names the direction of data flow (warehouse → third-party tool, rather than the usual tool → warehouse), not a distinct API. Clay's part is exactly the same batch-enrichment mechanism used elsewhere in this content set; what makes it "reverse ETL" is entirely the surrounding pipeline — pulling the segment from a real warehouse client beforehand, and writing the enriched rows back via that same client afterward.

Common issues

Looking for a Clay warehouse connector (Snowflake, BigQuery, etc.)

Cause: "reverse ETL" tools often ship native warehouse connectors, so it's natural to expect one from Clay too.

Fix: no such connector exists in the Public API — extract and load are both your own warehouse client's job; Clay only sees the segment you hand it and the enriched result you take back.

Losing the join key (`account_id`) during enrichment

Cause: keying the enrichment batch by the warehouse's internal ID instead of the domain the routine actually needs.

Fix: use the domain as the batch row id (what Clay's routine needs), then re-join to the original account_id client-side by domain, as this example does.

Next steps

Replace the simulated warehouseRows input with a real query against your warehouse (e.g. via its official SDK).
Write activationRows back with that same SDK's load/upsert call to complete the round trip.

verification:
  status: verified
  tested_at: "2026-09-02"
  product_version: "public/v0"
  command: "npx tsc && node --env-file=.env dist/index.js && curl -X POST http://localhost:3000/sync-warehouse-segment -d '{

verification:
  status: verified
  tested_at: "2026-09-02"
  product_version: "public/v0"
  command: "npx tsc && node --env-file=.env dist/index.js && curl -X POST http://localhost:3000/sync-warehouse-segment -d '{\"warehouseRows\":[{\"account_id\":\"acct_001\",\"domain\":\"stripe.com\"},{\"account_id\":\"acct_002\",\"domain\":\"notion.so\"}]}'"
  expected_result: "Both simulated warehouse rows are enriched with real Clay firmographic data and returned with their original account_id preserved for the load-back join."