TypeScript Client

The @pipe0/client package is a fully typed TypeScript SDK for the pipe0 API. It turns pipe and search IDs into autocomplete-friendly, compile-checked entrypoints, so you get real feedback from your editor the moment a pipe expects different config than you've written.

Installation

npm install @pipe0/client

Creating a client

import { Pipe0 } from "@pipe0/client";

export const pipe0 = new Pipe0({
  apiKey: process.env.PIPE0_API_KEY,
});

All options:

Option	Default	Notes
`apiKey`	`process.env.PIPE0_API_KEY`	Sent as `Authorization: Bearer …`.
`baseUrl`	`https://api.pipe0.com`	Override for self-hosted / staging.
`credentials`	`include`	Fetch credentials mode; `same-origin` when you set a `baseUrl`.
`pollingTimeoutMs`	`900000` (15 minutes)	How long `pipe()` / `search()` poll before throwing `Pipe0TimeoutError`.
`minPollingIntervalMs`	`1000`	First polling interval. Backs off exponentially from here.
`maxPollingIntervalMs`	`3 * minPollingIntervalMs`	Cap for the backoff.
`defaultBatchSize`	`100`	Chunk size for `pipeInBatches` (matches the API's 100-record cap).
`maxConcurrentBatches`	`5`	Parallel requests in `pipeInBatches` and `searchAll`.

Data enrichment (pipes)

Use pipes.pipe() to enrich up to 100 input objects at a time. Use pipes.pipeInBatches() to enrich any number of input objects.

Split a name

const result = await pipe0.pipes.pipe({
  pipes: [{ pipe_id: "person:name:split@1" }],
  input: [{ id: "1", name: "John Doe" }],
});

const record = result.records["1"]; // "1" is the id property
record.fields.first_name.value; // "John"
record.fields.last_name.value; // "Doe"

Chain pipes

const result = await pipe0.pipes.pipe({
  pipes: [
    { pipe_id: "person:profile:waterfall@1" },
    {
      pipe_id: "prompt:run@1",
      config: {
        model: "google-low",
        prompt: {
          template: `
            Profile: {{ profile }}
            {% output icp_fit, type: "boolean", description: "Does this match our ICP?" %}
            {% output reasoning, type: "string", description: "Why or why not?" %}
          `,
        },
      },
    },
    {
      pipe_id: "message:send:slack@1",
      run_if: {
        action: "run",
        when: {
          logic: "and",
          conditions: [
            { field_name: "icp_fit", property: "value", operator: "eq", value: true },
          ],
        },
      },
      connector: {
        strategy: "first",
        connections: [{ type: "vault", connection: "slack_abcd123" }],
      },
      config: { channel_id: "C0123456789", message: "New ICP match: {{ reasoning }}" },
    },
  ],
  input: [{ id: "1", profile_url: "https://www.linkedin.com/in/jane-doe" }],
});

The run_if means the Slack pipe only runs when the previous step marked icp_fit true. Conditional execution is part of the typed payload. message:send:slack@1 sends through your own Slack workspace, so it needs a vault connection ID like slack_abcd123 (see Connections).

Large inputs: `pipeInBatches`

Splits input into chunks and runs them with bounded concurrency.

import { Pipe0BatchError } from "@pipe0/client";

try {
  const batches = await pipe0.pipes.pipeInBatches(
    {
      pipes: [{ pipe_id: "person:workemail:waterfall@1" }],
      input: tenThousandRows,
    },
    {
      stopOnError: false,
      onBatchComplete: (i, res) => console.log(`batch ${i}: ${res.status}`),
    },
  );
} catch (err) {
  if (err instanceof Pipe0BatchError) {
    console.error(`${err.errors.length} batches failed`);
    console.log(`${err.successfulBatches.length} succeeded`);
  }
}

Advanced: Manual polling

If you want control over the polling process, use the manual handlers.

// Sends a valid API request
const runId = await pipe0.pipes.create({
  pipes: [{ pipe_id: "person:workemail:waterfall@1" }],
  input: largeInput,
});

// Check the status of the task
const status = await pipe0.pipes.check(runId);

// Or manually wait until complete (same as pipes.pipe())
const done = await pipe0.pipes.waitUntilComplete(runId, {
  onPoll: (response) => console.log(response.status),
});

Running searches

Searches discover new records. Sources are prospecting datasets (Crustdata, Amplemarket, Parallel) and systems you already own (HubSpot, Salesforce, Attio, pipe0 sheets and buckets, Postgres, Databricks).

Find people by title and location

const result = await pipe0.searches.search({
  search: {
    search_id: "people:profiles:crustdata@2",
    config: {
      limit: 25,
      filters: {
        current_job_titles: { include: ["Head of RevOps"] },
        locations: { include: ["San Francisco", "New York"] },
      },
    },
  },
});

for (const row of result.results) {
  row.name.value; // string
  row.job_title.value; // string
  row.company_domain.value; // typed per output_fields
}

Each request accepts a single search. See the search catalog for every search_id and its available filters.

The manual polling trio exists for searches too: searches.create(), searches.check(runId), and searches.waitUntilComplete(runId) mirror their pipes counterparts.

Many pages, many searches: `searchAll`

searches.searchAll() runs several searches concurrently, follows each response's next_page automatically, and merges the rows into one list. Use it when one filter set isn't enough, or when you want more than one page without writing the pagination loop.

const { results, errors } = await pipe0.searches.searchAll({
  searches: [
    { search: { search_id: "people:profiles:crustdata@2", config: { /* … */ } } },
    { search: { search_id: "people:profiles:amplemarket@2", config: { /* … */ } } },
  ],
  maxPages: 5, // per search; Infinity fetches until next_page is null
  dedupeBy: ["profile_url"], // composite keys supported, first match wins
  stopOnError: false, // collect per-search failures in `errors`
});

Rows are normalized by default: every row gets the union of all field names across every search, with absent fields set to null.

Combining searches with pipes

The output of a search is a set of records. Feed its results into a pipe to enrich further:

const search = await pipe0.searches.search({ /* … */ });

const enriched = await pipe0.pipes.pipe({
  pipes: [{ pipe_id: "person:workemail:waterfall@1" }],
  input: search.results,
});

Error handling

The client throws typed errors you can discriminate on:

import {
  Pipe0TimeoutError,
  Pipe0AbortError,
  Pipe0TaskError,
  Pipe0ServerError,
  Pipe0BatchError,
} from "@pipe0/client";

try {
  await pipe0.pipes.pipe({ /* … */ });
} catch (err) {
  if (err instanceof Pipe0TimeoutError) {
    // Polling exceeded pollingTimeoutMs. err.runId is still valid,
    // call pipes.check(err.runId) later.
  } else if (err instanceof Pipe0TaskError) {
    // The API returned a task-level error.
    console.error(err.responseBody);
  } else if (err instanceof Pipe0ServerError) {
    // Non-2xx response.
  }
}

Cancellation uses standard AbortController:

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

await pipe0.pipes.pipe(payload, { signal: controller.signal });

Next steps

Pipe catalog

Every pipe_id and its config shape.

Search catalog

Every search_id and its filters.

TypeScript Client

On this page