Batch CSV Enrichment

Job: enrich a list without losing track of which row worked, failed, or needs review.

Recommended CSV shape

Use a stable row key.

lead_id,first_name,last_name,linkedin_url,company_name,company_domain
lead-001,Jane,Smith,https://www.linkedin.com/in/example-person/,ExampleCo,example.com

Here is the starter shape for this workflow. If you are reading this outside the repo, create the same starter file:

cat > leads.csv <<'CSV'
lead_id,first_name,last_name,linkedin_url,company_name,company_domain
lead-001,Jane,Smith,https://www.linkedin.com/in/example-person/,ExampleCo,example.com
lead-002,Sam,Lee,https://www.linkedin.com/in/example-two/,Acme,acme.com
CSV

Column	Required	Why
`lead_id`	Yes	Stable row key for retries and output review
`linkedin_url`	Yes for this example	Best starting point for the prebuilt email play
`first_name`, `last_name`	Helpful	Improves matching
`company_name`, `company_domain`	Helpful	Improves matching and validation

If your file has different column names, ask Claude Code to map your columns to this shape before running.

The play pattern

The workflow maps each row to an email step.

import { definePlay } from 'deepline';

type Lead = {
  lead_id: string;
  first_name: string;
  last_name: string;
  linkedin_url: string;
  company_name: string;
  company_domain: string;
};

export default definePlay(
  'lead-email-waterfall',
  async (ctx, input: { csv: string }) => {
    const leads = await ctx.csv<Lead>(input.csv);

    const withEmail = await ctx
      .dataset('lead_email', leads)
      .withColumn('email', async (lead, rowCtx) => {
        const result = await rowCtx.tools.execute({
          id: 'person_email',
          tool: 'test_rate_limit',
          input: {
            key: lead.linkedin_url,
          },
          description: 'Find a verified work email for this lead.',
        });
        return result.status === 'SUCCESS'
          ? `${lead.first_name}.${lead.last_name}@${lead.company_domain}`.toLowerCase()
          : null;
      })
      .run({
        key: 'lead_id',
        description: 'Find one verified work email per lead.',
      });

    const rows = await ctx
      .dataset('email_validation', withEmail)
      .withColumn('email_status', async (lead, rowCtx) => {
        if (!lead.email) return 'missing';
        const result = await rowCtx.tools.execute({
          id: 'verify_email',
          tool: 'test_rate_limit',
          input: { key: String(lead.email) },
          description: 'Validate the candidate email.',
        });
        return result.status;
      })
      .run({
        key: 'lead_id',
        description: 'Validate the emails that were found.',
      });

    return { rows };
  },
);

Pilot it first

The play owns its CSV contract. A play that declares input.csv can be run with --csv. Reserved run flags such as --file keep their command meaning, so a play that declares input.file should be run with --input '{"file":"leads.csv"}'. This example reads rows with ctx.csv(input.csv).

deepline plays check docs-examples/sdk-v2/lead-email-waterfall.play.ts

head -n 4 leads.csv > /tmp/leads-pilot.csv

deepline plays run docs-examples/sdk-v2/lead-email-waterfall.play.ts \
  --csv /tmp/leads-pilot.csv \
  --watch

deepline runs export <run-id> --out /tmp/leads-pilot-enriched.csv

Inspect /tmp/leads-pilot-enriched.csv. You want the original columns plus the email and validation columns added by the play.

Verify a tool contract before adding it

A tool name does not define its payload or response structure. Before you add a provider step, examine these items:

The live schema
The price
The connection state
The result getters
The limits

deepline tools describe <tool-id> --json

Run the first pilot without extract_js if the saved output structure is unknown. Examine the exported pilot. Then, add the smallest applicable extractor. For a deepline enrich provider step, the extractor receives a wrapper. The raw tool output is under output_data.result. Do not assume that provider fields are at the wrapper root. A saved CSV cell can also include matched_result. Deepline adds this field after extraction. It is not an input to extract_js. Use the getters from tools describe. Do not write a provider path if an applicable getter is available. If the getter does not match the pilot, keep the raw result. Report the contract difference. Do not scale the run until you know the output structure.

Choose pilot rows deliberately

Do not estimate a full list from the first account or from easy rows. Build a small representative pilot. Include input conditions that can change coverage or cost. Examples include geography, company size, available identifiers, and role seniority. Use deepline billing balance --json before and after the pilot. Calculate the balance difference. A composed play can put provider charges in child runs. Thus, do not use only the parent run to measure cost.

Keep bulk data in files

Play input is for control parameters and file references. Do not pass a full spreadsheet as inline JSON through --input. Use ctx.csv(input.csv) with a staged file path instead:

deepline plays run docs-examples/sdk-v2/lead-email-waterfall.play.ts \
  --csv leads.csv \
  --watch

Inline submitted JSON has a hard 1 MiB ceiling. Payloads above that are rejected with guidance to use staged files or ctx.csv inputs. Deepline automatically externalizes moderately large retry payloads between 100 KB and 1 MiB so runs can recover from platform retries, but row data should still live in CSV files.

Run the full file

deepline plays run docs-examples/sdk-v2/lead-email-waterfall.play.ts \
  --csv leads.csv \
  --watch

deepline runs export <run-id> --out leads-enriched.csv

Why this matters

ctx.dataset makes every row resumable. If row 83 fails, you do not need to rerun the whole list blindly. If a row is missing required input, keep it in the output and mark it for review instead of deleting it silently. For scheduled refreshes, add a stale window:

import { definePlay } from 'deepline';

type Account = {
  domain: string;
};

export default definePlay(
  'account-refresh',
  async (ctx, input: { accounts: Account[] }) => {
    await ctx
      .dataset('account_refresh', input.accounts)
      .withColumn('company', (account, rowCtx) =>
        rowCtx.tools.execute({
          id: 'company',
          tool: 'test_rate_limit',
          input: { key: account.domain },
          description: 'Refresh the company profile.',
          staleAfterSeconds: 86_400,
        }),
      )
      .run({ key: 'domain' });

    return { ok: true };
  },
);

That says: reuse today’s completed rows, refresh tomorrow.

Getting started

Tutorials

How-to guides

Explanations

Real-world examples

Reference

Batch CSV Enrichment

Recommended CSV shape

The play pattern

Pilot it first

Verify a tool contract before adding it

Choose pilot rows deliberately

Keep bulk data in files

Run the full file

Why this matters

​Recommended CSV shape

​The play pattern

​Pilot it first

​Verify a tool contract before adding it

​Choose pilot rows deliberately

​Keep bulk data in files

​Run the full file

​Why this matters

Recommended CSV shape

The play pattern

Pilot it first

Verify a tool contract before adding it

Choose pilot rows deliberately

Keep bulk data in files

Run the full file

Why this matters