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`.
`autoRetry429`	`true`	Absorb rate-limit 429s automatically (see below).
`retryAfterCapMs`	`30000`	Upper bound on a single `Retry-After` wait.
`rateLimitMaxWaitMs`	`120000` (2 minutes)	Total wait budget for `org-rate-limited` retries per request.
`queueFullMaxWaitMs`	`900000` (15 minutes)	Total wait budget for `queue-limit-exceeded` retries per create.
`priorityFloor`	`2`	Worst task priority `pipeInBatches` voluntarily submits at.
`pacingProbeMs`	`15000`	Re-probe delay when batch submission is capacity-blocked with nothing in flight.
`cancelOnAbort`	`true`	Aborting also cancels still-pending runs server-side.
`onCapacity`	—	Called with a capacity snapshot whenever fresh `X-P*-Capacity` headers are seen.

Rate limits: handled for you

pipe0's API never rejects correct clients — work is accepted and scheduled by task priority. The SDK implements the whole contract so you usually don't have to think about it:

429s are retried automatically. org-rate-limited waits out the server's Retry-After; queue-limit-exceeded (the queued-record hard ceiling) backoff-retries the create. Only an exhausted wait budget throws Pipe0RateLimitError.
pipeInBatches paces itself. The rolling pool submits the next batch only when the capacity headers say it would still be admitted at priorityFloor or better, so large jobs don't sink themselves into the slow lane.
Aborting cancels server-side. When your AbortSignal fires, still-pending runs are canceled remotely and their records return to your capacity instantly.
Live capacity is exposed via pipe0.capacity and the onCapacity callback.

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),
});

// Cancel a run that is still queued — its records return to your
// organization's capacity immediately. Throws Pipe0ServerError with
// status 409 once the run is already processing or finished.
const canceled = await pipe0.pipes.cancel(runId);

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,
  Pipe0CanceledError,
  Pipe0TaskError,
  Pipe0ServerError,
  Pipe0RateLimitError,
  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 Pipe0RateLimitError) {
    // The automatic 429 handling exhausted its wait budget.
    console.error(err.problemType, err.attempts);
  } else if (err instanceof Pipe0CanceledError) {
    // The run was canceled server-side (subclass of Pipe0AbortError).
  } else if (err instanceof Pipe0ServerError) {
    // Any other non-2xx response — err.status / err.problemType are set.
  }
}

Cancellation uses standard AbortController. By default an abort also cancels still-pending runs server-side so their records return to your capacity immediately (cancelOnAbort: false restores client-only aborts):

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