Pipelines API

Compose and run DAGs of scraping, enrichment, and transformation steps under /api/pipelines.

Pipelines API

A pipeline is a directed acyclic graph (DAG) of nodes that chains scraping, enrichment, and transformation steps. Submitting a pipeline starts the root jobs synchronously; downstream jobs are spawned as each predecessor reaches done.

All endpoints live under /api/pipelines and require an authenticated session. Mutating routes additionally require an active (non-suspended) account. Generic errors: 401 no session, 403 not owner / account suspended, 404 pipeline or node id unknown; endpoint-specific causes are listed inline.

See also: Pipeline orchestration and Filter module.

Graph shape

The definition document describes the DAG. Edges are explicit and reference node ids; they are not inferred from any per-node inputs field.

{
  "nodes": [
    {"id": "n1", "type": "scrap",  "config": {"queries": ["dentist"], "zones": ["Paris"]}, "x": 100, "y": 100},
    {"id": "n2", "type": "emails", "config": {"mode": "normal"}, "x": 320, "y": 100},
    {"id": "n3", "type": "verify", "config": {}, "x": 540, "y": 100}
  ],
  "edges": [
    {"id": "e1", "from": "n1", "to": "n2"},
    {"id": "e2", "from": "n2", "to": "n3"}
  ]
}

Node types

filter and sort preserve the upstream type; type compatibility is resolved by walking back to the nearest non-passthrough ancestor.

verify reads the email column, so it also connects to any pois output as soon as that column exists upstream: a block earlier in the chain produces it, or the chain starts from an import. The enrichments between emails and verify can come in any order, for example scrap → emails → legal_data → verify.

Column guarantee

No module ever drops a column. Every enrichment and check node outputs all the columns it received (in their original order) plus its own columns appended at the end. This holds across an entire chain: a scrap → legal_ids → legal_data → emails pipeline ends with the Google Maps columns (lien_google_maps, note, nb_avis, lat, lon, …), the identifiers, the legal profile, and the emails side by side. Custom columns from an import are passed through untouched as well. The only way a column disappears is an explicit transformation the pipeline asks for (a filter rule, a sort top_n cut applies to rows, never columns).

The full machine-readable contract for every node (category, input/output, the needs/produces columns, and each block's config_schema) is served live at GET /api/pipelines/schema. That endpoint is the single source of truth; this table is a summary.

Validation rules

The server rejects a definition with HTTP 400 if any of the following hold:

Roots must be one of scrap, sirene, or import. Fan-out (one node feeding several successors) is allowed; fan-in is not.

Portable envelope

Pipelines export and import as a single self-describing JSON envelope. The same shape is produced by the editor's Export button and accepted by Import and AI generation:

{
  "schema_version": 1,
  "name": "Dentists Paris",
  "definition": { "nodes": [...], "edges": [...] },
  "meta": { "exported_from": "outsend.xyz", "kind": "pipeline" }
}

POST /api/pipelines and POST /api/pipelines/validate accept either the full envelope or a bare definition. Before validation the server normalizes the definition: it generates missing edge ids, auto-lays out nodes that have no x/y, applies config_schema defaults, coerces newline-separated strings into string[] fields, and strips config fields not in the schema. This means a minimal hand-written or AI-generated {nodes, edges} (no coordinates, no edge ids) is accepted as-is.

GET /api/pipelines/schema

Return the canonical, machine-readable pipeline schema: the single source of truth shared by the editor, import, AI generation, and the planned MCP create_pipeline tool. Public (no auth): it is format documentation, identical for every caller.

Response 200

{
  "schema_version": 1,
  "compat": { "pois_any": ["pois", "pois_email"], "pois_email": ["pois_email"], "any_pois": ["pois", "pois_email", "verified"] },
  "root_types": ["import", "scrap", "sirene"],
  "nodes": {
    "scrap": {
      "category": "source", "is_root": true,
      "input": null, "output": "pois",
      "needs": [], "produces": ["nom", "site_web", "telephone", "..."],
      "config_schema": {
        "queries": {"type": "string[]", "required": true, "label": "..."},
        "zones":   {"type": "string[]", "required": true, "label": "..."}
      }
    },
    "emails": { "category": "enrich", "input": "pois_any", "output": "pois_email",
                "needs": ["site_web"], "produces": ["email", "email_personal"],
                "config_schema": {"mode": {"type": "enum", "enum": ["normal", "deep"], "default": "normal"}} }
  }
}

config_schema field types: string, string[], int, float, bool, enum (with enum list), object. required and default are optional per field.

POST /api/pipelines/validate

Tells whether a definition would be accepted by POST /api/pipelines, without creating or running anything. Works with a session or an API key.

It runs the same checks as the creation, in this order:

1. Structure (reason: "definition"): nodes, edges and config, see Validation rules.
2. Zones (reason: "zones"): the country and every zone of each scrap node are resolved exactly as at launch. The message names the refused zone, the country, and a suggestion when one exists.
3. Cost (reason: "cost"): the worst-case EF estimate against the account threshold.
4. Room to start (reason: "busy"): the pipeline's root jobs must fit in the 5 concurrent jobs of the account. This one is transient: wait for a job to finish, or cancel one.

Check with validate before creating. A definition that passes is accepted by POST /api/pipelines as long as the account still has room for its root jobs. Limited to 120 calls per 10 minutes per account (see Limits).

Request body

{ "definition": { "nodes": [...], "edges": [...] }, "schema_version": 1 }

Optional structure_only: true stops after step 1. The editor uses it to load an imported definition that is then fixed on screen.

Response 200: valid

{
  "ok": true,
  "definition": { "nodes": [...with ids, x/y, defaults...], "edges": [...] },
  "summary": { "n_nodes": 3, "n_edges": 2, "types": ["scrap", "emails", "verify"] }
}

Response 200: invalid (note: still HTTP 200, with ok: false)

{ "ok": false, "reason": "zones", "error": "Unrecognised zone for country FRA: « Pariss » (did you mean « Paris »). Fix the zone, then try again." }

error is in English for API-key callers and in French for the web app.

The returned definition is the normalized form, ready to load into the editor or submit verbatim to POST /api/pipelines.

Generate a pipeline with any AI

You don't need to write the JSON by hand. Two ways:

1. Built-in (inside outsend). The editor's 🤖 Build with AI button sends the schema above plus your plain-language description to Claude using your own key (BYOK), then validates and lays out the result.

2. Bring-your-own assistant (copy/paste anywhere). Open any AI assistant (claude.ai, Claude Desktop, Cursor, ChatGPT) and:

1. Paste the contract: either this page, or the whole docs bundle at /docs/llms-full.txt, or just the JSON of GET /api/pipelines/schema.
2. Add your request, e.g. "Compose an outsend pipeline that finds dentists in Berlin, gets their emails, verifies deliverability, and keeps the top-rated. Return only the JSON envelope."
3. The assistant returns a {schema_version, name, definition} envelope. Paste it into the editor's Import dialog (it is validated before anything runs), or POST it to /api/pipelines.

Because the editor, import, and this API all accept the same envelope and the server normalizes it (missing edge ids, coordinates, and config defaults are filled in), a hand-assembled {nodes, edges} works without any layout fields.

POST /api/pipelines

Create a pipeline and launch its root jobs.

Request body

{
  "name": "Dentists Paris",
  "definition": { "nodes": [...], "edges": [...] },
  "low_priority": false
}

name is optional (≤ 120 chars, defaults to "Pipeline"). The definition is normalized (see Portable envelope) before validation, so a minimal {nodes, edges} works.

low_priority is optional (defaults to false). Set it to true to make the pipeline low priority. Its jobs then run after other jobs in the queue. A low-priority pipeline costs less.

Response 201

{
  "id": "f1a2…-uuid",
  "status": "running",
  "initial_jobs": ["job_abc", "job_def"]
}

A refusal leaves nothing behind. Everything that can refuse (structure, zones, cost, room in the 5 concurrent jobs) is checked before anything is written. A refused request returns 400 with {"detail": "..."} and creates neither a pipeline nor a job: there is no pipeline left in failed to clean up. The message states the cause and what to do, for example:

• Too many jobs in progress: 5 pending or running on this account, 5 at most, and this pipeline starts 1. Wait for a job to finish or cancel one, then try again.
• Unrecognised zone for country FRA: « Pariss » (did you mean « Paris »). Fix the zone, then try again.
• Estimated cost too high: up to 7.5 EF for this pipeline, against a limit of 5 EF on this account. Reduce the zones, queries or steps, or raise the limit in Settings.

429: creation quota reached (30 creations per 10 minutes and 300 per 24 hours per account, jobs and pipelines together, refused attempts included). Wait for the Retry-After header. See Limits.

To avoid refusals, check the definition with POST /api/pipelines/validate first.

GET /api/pipelines

List the caller's pipelines (most recent first, capped at 50).

Response 200

[
  {
    "id": "f1a2…",
    "name": "Dentists Paris",
    "status": "running",
    "created_at": "2026-05-27 10:14:02",
    "completed_at": null,
    "nodes_count": 3,
    "done_count": 1,
    "results_count": 187,
    "progress_pct": 42
  }
]

status is one of pending | running | done | failed | cancelled. nodes_count is derived from the stored definition. done_count is the number of stages already finished, results_count the rows aggregated across all stages so far, and progress_pct (0–100) a duration-weighted completion estimate: transform stages (filter, sort) count far less than scraping/enrichment stages, and the in-flight stage contributes its real sub-progress. The same progress_pct is also returned by GET /api/pipelines/{id}.

GET /api/pipelines/{id}

Return a single pipeline with its definition and the jobs spawned so far.

Response 200

{
  "id": "f1a2…",
  "user_id": 42,
  "name": "Dentists Paris",
  "definition": { "nodes": [...], "edges": [...] },
  "status": "running",
  "created_at": "2026-05-27 10:14:02",
  "completed_at": null,
  "progress_pct": 42,
  "jobs": [
    {
      "id": "job_abc",
      "job_type": "scrap",
      "status": "done",
      "pipeline_node_id": "n1",
      "results_count": 187,
      "error_message": null,
      "created_at": "2026-05-27 10:14:02",
      "completed_at": "2026-05-27 10:18:55"
    }
  ],
  "output_job": {
    "id": "job_xyz",
    "job_type": "verify_emails",
    "results_count": 142,
    "status": "done",
    "download_available": true
  }
}

output_job is the pipeline's final dataset: the output of the most-downstream stage that has actually produced rows (a pipeline filters/reduces, it does not sum; output_job.results_count therefore matches the headline count, not the sum of all stages). Download it via GET /api/jobs/{id}/download using output_job.id, in csv / xlsx / json. It is null while the pipeline has produced nothing downloadable yet, and download_available reflects whether a CSV (final or partial: so it works for running/stopped pipelines too) is still on disk and unexpired.

PATCH /api/pipelines/{id}

Not implemented. The current API does not expose graph mutation after creation; clone the pipeline by re-issuing POST /api/pipelines with an updated definition. Returns 405 Method Not Allowed.

DELETE /api/pipelines/{id}

Not implemented. Pipelines are immutable once created; deletion will be added once retention policy is defined. Returns 405 Method Not Allowed.

POST /api/pipelines/{id}/run

Not implemented. Pipelines start automatically when created via POST /api/pipelines; there is no separate run endpoint. To re-execute an existing graph, post it again as a new pipeline.

GET /api/pipelines/{id}/nodes/{node_id}/input-columns

Inspect the schema of the CSV that will feed a given node. Useful for building filter UIs.

Behaviour. The endpoint locates the node's most recent predecessor job. If the predecessor is not yet done, the response carries an empty columns list and a reason code. Otherwise the predecessor's output CSV is read (up to 5000 rows) and each column is profiled for type, fill rate, and sample values.

Response 200: predecessor done

{
  "columns": [
    {
      "name": "telephone",
      "type": "phone",
      "fill_rate": 0.92,
      "sample_values": ["+33 1 23 45 67 89", "0612345678"],
      "distinct_count": null
    },
    {
      "name": "categorie",
      "type": "category",
      "fill_rate": 1.0,
      "sample_values": ["dentiste", "orthodontiste"],
      "distinct_count": 4,
      "distinct_values": ["dentiste", "endodontiste", "orthodontiste", "stomatologue"]
    }
  ],
  "row_count": 187,
  "predecessor_job_id": "job_abc"
}

type is one of phone | email | url | number | category | text. A column is tagged category only if it has between 1 and 200 distinct non-empty values; otherwise it falls back to text. A typed verdict requires ≥ 80% of non-empty values to match the corresponding pattern.

Response 200: no usable input

{ "columns": [], "reason": "no_predecessor" }

POST /api/pipelines/{id}/nodes/{node_id}/filter-preview

Apply a set of filter rules in memory against the upstream CSV and return the match count plus a small sample. No job is created; no state is mutated.

The target node must be of type filter. The body uses the same rules shape that filter nodes persist in their config.rules; previews are computed by the same function the worker uses at execution time, so the count is authoritative for the data inspected.

Request body

{
  "rules": {
    "logic": "AND",
    "conditions": [
      {"column": "fill_rate", "op": ">=", "value": 0.5},
      {"column": "categorie", "op": "in", "value": ["dentiste", "orthodontiste"]}
    ]
  }
}

The exact rule grammar is defined by the filter module (see Filter module).

Response 200

{
  "total": 187,
  "matched": 73,
  "samples": [
    {"nom": "Cabinet Dupont", "telephone": "0123456789", "categorie": "dentiste"}
  ],
  "predecessor_job_id": "job_abc",
  "fieldnames": ["nom", "telephone", "categorie", "site_web"],
  "capped": false
}

samples contains up to 5 matched rows with empty fields stripped. capped is true when the upstream CSV exceeded the 5000-row preview limit: in that case total reflects only the inspected window, but the matched/total ratio remains representative.

When the predecessor is not ready, the response is the same {total, matched, samples, reason} skeleton with all counts at 0. Possible reason codes mirror the input-columns endpoint: no_predecessor, no_data_yet, no_csv_found.

Specific causes: 400 target node is not of type filter, or rule application raised; 500 CSV could not be read.

Lifecycle summary

1. POST /api/pipelines validates the graph, the zones, the cost and the room in the concurrent-job quota, then persists the pipeline as running and spawns one job per root node. A refused request persists nothing.
2. As each job reaches done, the worker reads its CSV, transforms rows for the successor's input type, and creates the next job. Empty outputs short-circuit the branch.
3. When a job is cancelled after it started (or fails), the pipeline is paused on that stage and stays running until it is continued from that stage or cancelled as a whole. When a job is cancelled before it ever started (pending, no result), the pipeline is closed as cancelled, unless another of its stages is still pending or running.
4. When every reachable stage is terminal and resolved, the pipeline is finalized as done. A whole-pipeline cancellation finalizes it as cancelled; an expired job finalizes it as failed.