Pipeline reference - VectorShift

Lifecycle

`new`

Pipeline.new(name: str, nodes: Optional[list[Node]] = None) -> Pipeline

async Pipeline.anew(name: str, nodes: Optional[list[Node]] = None)

Create a new pipeline with the specified parameters. Parameters

name

str

required

The name of the pipeline.

nodes

Optional[list[Node]]

default:"None"

List of nodes to include in the pipeline.

Returns

returns

Pipeline

A new Pipeline instance.

`save`

Pipeline.save(deploy: bool = True, bump: Optional[BumpLevel] = None, description: Optional[str] = None)

async Pipeline.asave(deploy: bool = True, bump: Optional[BumpLevel] = None, description: Optional[str] = None)

Save the pipeline with its current configuration. Updates the pipeline in the database with the current nodes, inputs and outputs configuration.

`deploy=True` vs `deploy=False` — the branch model

Think of a pipeline like a Git repo:

Working tree — the Pipeline object in your Python process. Every pipeline.add(...) mutation lives here.
Main branch — the branch_id on the server. save(deploy=False) writes your working tree to this branch as a new commit. Anything reading by branch_id (the platform editor, ad-hoc pipeline.run(...) from this SDK process) immediately sees the change.
Deployed version — what callers of the published pipeline get. Chatbots, integrations, scheduled runs, and the platform’s “Run published” button all read this version. save(deploy=True) (the default) promotes the current main-branch state to be the new deployed version.

bump attaches a semantic-version tag ("major", "minor", or "patch") to the deploy. Consumers can then pin to that version via Pipeline.fetch(id=..., version="1.2.0") instead of always tracking latest. In practice:

Iterating on a draft? save(deploy=False) — keeps the deployed version stable while you tweak.
Ready to ship? save(deploy=True) (default) — optionally with bump="patch" + description="..." for a labelled release.

Parameters

deploy

bool

default:"True"

When True, promotes the saved state to be the new deployed version (the one chatbots, integrations, and the “Run published” path will hit). When False, only updates the working/main branch — the deployed version is untouched until the next save(deploy=True).

bump

Optional[BumpLevel]

default:"None"

Semantic-version bump level: "major", "minor", or "patch". Only meaningful with deploy=True. See BumpLevel.

description

Optional[str]

default:"None"

Updates the pipeline description; also used as the changelog entry when bumping a version.

Returns

returns

dict: A dictionary containing the status of the save operation.

Raises

Exception

exception

If the pipeline update fails.

`fetch`

Pipeline.fetch(id: Optional[str] = None, name: Optional[str] = None, branch_id: Optional[str] = None, version: Optional[str] = None, username: Optional[str] = None, org_name: Optional[str] = None) -> Pipeline

async Pipeline.afetch(id: Optional[str] = None, name: Optional[str] = None, branch_id: Optional[str] = None, version: Optional[str] = None, username: Optional[str] = None, org_name: Optional[str] = None) -> Pipeline

Fetches an existing pipeline by id or name. branch_id and version only apply when fetching by id. Parameters

Optional[str]

default:"None"

The unique identifier of the pipeline to fetch.

name

Optional[str]

default:"None"

The name of the pipeline to fetch. When multiple pipelines share the same name, the most recently modified one is returned.

branch_id

Optional[str]

default:"None"

Fetch the pipeline for a specific branch (id-fetch only).

version

Optional[str]

default:"None"

Semantic version to fetch (e.g. "1.2.0"), or "latest" (id-fetch only).

username

Optional[str]

default:"None"

The username of the pipeline owner.

org_name

Optional[str]

default:"None"

The organization name of the pipeline owner.

Returns

returns

Pipeline

Pipeline: The fetched Pipeline instance.

Raises

ValueError

exception

If neither id nor name is provided.

`list`

Pipeline.list(folder_id: Optional[str] = None, include_shared: Optional[bool] = None, verbose: Optional[bool] = None, offset: Optional[int] = None, limit: Optional[int] = None) -> "list[Pipeline]"

async Pipeline.alist(folder_id: Optional[str] = None, include_shared: Optional[bool] = None, verbose: Optional[bool] = None, offset: Optional[int] = None, limit: Optional[int] = None) -> "list[Pipeline]"

List pipelines. Parameters

folder_id

Optional[str]

default:"None"

Filter pipelines by folder.

include_shared

Optional[bool]

default:"None"

Include shared pipelines.

verbose

Optional[bool]

default:"None"

Include full pipeline details.

offset

Optional[int]

default:"None"

Pagination offset.

limit

Optional[int]

default:"None"

Maximum number of pipelines to return.

Returns

returns

list[Pipeline]

A list of Pipeline instances.

`delete`

Pipeline.delete()

async Pipeline.adelete()

Deletes an existing pipeline. Returns

returns

dict: A dictionary containing the status of the deletion operation.

Raises

Exception

exception

If the pipeline couldn’t be deleted.

`duplicate`

Pipeline.duplicate(name: Optional[str] = None, description: Optional[str] = None, folder_id: Optional[str] = None) -> Pipeline

async Pipeline.aduplicate(name: Optional[str] = None, description: Optional[str] = None, folder_id: Optional[str] = None) -> Pipeline

Duplicate this pipeline. Parameters

name

Optional[str]

default:"None"

Name for the duplicate. Auto-generated if omitted.

description

Optional[str]

default:"None"

Description for the duplicate.

folder_id

Optional[str]

default:"None"

Folder to place the duplicate in.

Returns

returns

Pipeline

A new Pipeline instance for the duplicated pipeline.

Building

`add`

Pipeline.add -> NodeAdder  # property

Fluent builder for adding nodes to this pipeline. Usage

inp = pipeline.add(name="query").input(input_type="string")
llm = pipeline.add(name="llm").llm(provider="openai", model="gpt-4o", prompt=inp.text)

Returns

returns

NodeAdder

`add_batch`

Pipeline.add_batch -> BatchNodeAdder  # property

Fluent builder that adds nodes with execution_mode='batch'. Usage

llm = pipeline.add_batch.llm(provider="openai", model="gpt-5.1", prompt=inp.text)
# equivalent to: pipeline.add.llm(..., execution_mode="batch")

Returns

returns

BatchNodeAdder

Node catalogue

pipeline.add(name=).<node_type>(...) exposes every node type as a typed builder method. There are ~500 node types in total. The biggest categories:

Category	Representative builders
I/O	`input`, `output`, `text`, `file`, `file_to_text`, `file_save`
LLM & AI	`llm`, `ai_routing`, `ai_filter_list`, `ai_operations`, `ai_text_to_image`, `ai_image_to_text`, `ai_text_to_speech`, `ai_speech_to_text`, `ai_fill_pdf`, `categorizer`
Knowledge & retrieval	`knowledge_base`, `knowledge_base_loader`, `url_loader`, `arxiv`, `wikipedia`, `deep_research`, `semantic_search`, `parallel_ai_search`
Control flow	`condition`, `ai_routing`, `merge`, `combine_text`, `pipeline` (sub-pipeline), `agent` (agent-as-tool)
Transforms	`text_formatter`, `combine_text`, `split_text`, `transformation` (custom Python), `generate_chart`
Data sources	`read_json_values`, `csv_reader`, `dataframe_`, `excel_`, `append_files`, `file_save`
Integrations	`gmail`, `slack`, `notion`, `airtable`, `salesforce`, `hubspot`, … (every connected service exposes a node)
Interaction	`chat_*`, `browser_extension`, `api`

Each builder returns a typed node object with attributes (.text, .response, .results, .path_0, etc.) you wire into downstream nodes. The full surface ships as a .pyi stub (vectorshift/pipeline/node_adder.pyi) — your editor’s autocomplete is the canonical browseable catalogue.

inp = pipeline.add(name="q").input(input_type="string")
llm = pipeline.add(name="llm").llm(
    provider="openai", model="gpt-5.1", prompt=inp.text,
)
pipeline.add(name="out").output(output_type="string", value=llm.response)

For runnable examples covering each category, see the pipeline examples gallery — the sidebar groups them by purpose (Core, Logic & routing, Composition, Streaming, Background, Integrations, Lifecycle).

`add_node`

Pipeline.add_node(node: Node) -> None

Add a node to the pipeline. If a node with the same name already exists, it will be replaced in-place. Parameters

node

Node

required

The node to add to the pipeline.

`remove_node`

Pipeline.remove_node(node_id: Optional[str] = None, node_name: Optional[str] = None) -> dict[str, Any]

async Pipeline.aremove_node(node_id: Optional[str] = None, node_name: Optional[str] = None) -> dict[str, Any]

Remove a node (and its connected edges) from the pipeline. Provide either node_id or node_name. Parameters

node_id

Optional[str]

default:"None"

ID of the node to remove.

node_name

Optional[str]

default:"None"

Name of the node to remove.

Returns

returns

dict[str, Any]

\{"status": "success", "node_id": "\<resolved_node_id>"\}

Running

`run`

Pipeline.run(inputs: dict[str, Any], stream: bool = False, stream_all_outputs: bool = False, session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, send_intermediate_results: Optional[bool] = None) -> Union[dict[str, Any], Generator]

async Pipeline.arun(inputs: dict[str, Any], stream: bool = False, stream_all_outputs: bool = False, session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, send_intermediate_results: Optional[bool] = None) -> Union[dict[str, Any], AsyncGenerator]

Run the pipeline with the specified inputs. Parameters

inputs

dict[str, Any]

required

Dictionary of input nodes -> input values for the pipeline. eg: {“input_node”: “Hello, world!”}

stream

bool

default:"False"

Whether to stream the response. (Set true only when pipeline has an output node with a streaming llm input)

stream_all_outputs

bool

default:"False"

Whether to stream all outputs as they arrive.

session_id

Optional[str]

default:"None"

Optional session ID for run grouping and tracing. Groups multiple runs under the same session for analytics and observability. Note: this does not provide conversational memory to LLM nodes — each run’s LLM nodes only see the current inputs.

node_input_overrides

Optional[dict[str, Any]]

default:"None"

Optional per-node input overrides.

send_intermediate_results

Optional[bool]

default:"None"

Whether to send intermediate results (default: True on server).

Returns

returns

Union[dict[str, Any], Generator]

Union[dict[str, Any], Generator]: A dictionary containing pipeline outputs and run_id. If stream is True, returns a generator that yields response chunks.

Raises

Exception

exception

If the pipeline execution fails.

`bulk_run`

Pipeline.bulk_run(inputs: list[dict[str, Any]]) -> dict[str, Any]

async Pipeline.abulk_run(inputs: list[dict[str, Any]]) -> dict[str, Any]

Run the pipeline with a list of specified inputs. Parameters

inputs

list[dict[str, Any]]

required

List of dictionaries of input values for the pipeline.

Returns

returns

dict[str, Any]

A single dictionary with two keys: run_outputs — a list with one entry per input set, each containing that run’s outputs — and status, the overall bulk-run status. (Note: this is one dict wrapping all runs, not a list.)

Raises

Exception

exception

If the pipeline execution fails.

`run_status`

Pipeline.run_status(run_id: str) -> RunStatus

async Pipeline.arun_status(run_id: str) -> RunStatus

Check the status of an async pipeline run. Parameters

run_id

str

required

The run/task ID returned by :meth:start.

Returns

returns

RunStatus

A :class:RunStatus dict with keys task_id, status, and optionally error or result. See RunStatus.

Background runs

`start`

Pipeline.start(inputs: dict[str, Any], session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, send_intermediate_results: Optional[bool] = None, webhook_url: Optional[str] = None, version: Optional[str] = None, trace_id: Optional[str] = None) -> RunHandler

async Pipeline.astart(inputs: dict[str, Any], session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, send_intermediate_results: Optional[bool] = None, webhook_url: Optional[str] = None, version: Optional[str] = None, trace_id: Optional[str] = None) -> RunHandler

Start pipeline in background, return a RunHandler. Parameters

inputs

dict[str, Any]

required

Dictionary of input values for the pipeline.

session_id

Optional[str]

default:"None"

node_input_overrides

Optional[dict[str, Any]]

default:"None"

Optional per-node input overrides.

send_intermediate_results

Optional[bool]

default:"None"

Whether to send intermediate results.

webhook_url

Optional[str]

default:"None"

URL to call when the run completes.

version

Optional[str]

default:"None"

Pipeline version to run.

trace_id

Optional[str]

default:"None"

Trace ID for observability.

Returns

returns

RunHandler

A :class:RunHandler for the background run.

`terminate`

Pipeline.terminate(run_id: str) -> dict[str, Any]

async Pipeline.aterminate(run_id: str) -> dict[str, Any]

Terminate an active pipeline run. Parameters

run_id

str

required

The pipeline run ID to terminate.

Returns

returns

dict[str, Any]

\{"status": "success"\}

Streaming

`stream`

Pipeline.stream(inputs: dict[str, Any], stream_all_outputs: bool = False, session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, version: Optional[str] = None) -> Generator[StreamChunk, None, None]

Pipeline.astream(inputs: dict[str, Any], stream_all_outputs: bool = False, session_id: Optional[str] = None, node_input_overrides: Optional[dict[str, Any]] = None, version: Optional[str] = None) -> AsyncGenerator[StreamChunk, None]

Run the pipeline with streaming, yielding StreamChunk objects. Parameters

inputs

dict[str, Any]

required

Dictionary of input values for the pipeline.

stream_all_outputs

bool

default:"False"

Whether to stream all outputs as they arrive.

session_id

Optional[str]

default:"None"

node_input_overrides

Optional[dict[str, Any]]

default:"None"

Optional per-node input overrides.

version

Optional[str]

default:"None"

Pipeline version to run.

Returns

returns

Generator[StreamChunk, None, None]

A generator of :class:StreamChunk objects.

Pipeline.share(user_id: Optional[str] = None, org_id: Optional[str] = None, role: "Literal[viewer, editor]" = viewer, name: Optional[str] = None) -> dict[str, Any]

async Pipeline.ashare(user_id: Optional[str] = None, org_id: Optional[str] = None, role: "Literal[viewer, editor]" = viewer, name: Optional[str] = None) -> dict[str, Any]

Share this pipeline with a user or organization. At least one of user_id or org_id is required. Parameters

user_id

Optional[str]

default:"None"

User ID to share with.

org_id

Optional[str]

default:"None"

Organization ID to share with.

role

Literal[viewer, editor]

default:"'viewer'"

Permission role ("viewer" or "editor").

name

Optional[str]

default:"None"

Display name for the shared entity.

Returns

returns

dict[str, Any]

\{"status": "success"\}

`unshare`

Pipeline.unshare(user_id: Optional[str] = None, org_id: Optional[str] = None) -> dict[str, Any]

async Pipeline.aunshare(user_id: Optional[str] = None, org_id: Optional[str] = None) -> dict[str, Any]

Remove sharing for this pipeline. At least one of user_id or org_id is required. Parameters

user_id

Optional[str]

default:"None"

User ID to unshare.

org_id

Optional[str]

default:"None"

Organization ID to unshare.

Returns

returns

dict[str, Any]

\{"status": "success"\}

`publish`

Pipeline.publish(title: str, description: str = '', tags: Optional[list[str]] = None, internal: bool = False) -> dict[str, Any]

async Pipeline.apublish(title: str, description: str = '', tags: Optional[list[str]] = None, internal: bool = False) -> dict[str, Any]

Publish this pipeline to the marketplace. Parameters

title

str

required

Marketplace listing title.

description

str

default:"''"

Marketplace listing description.

`unpublish`

Pipeline.unpublish(marketplace_id: str) -> dict[str, Any]

async Pipeline.aunpublish(marketplace_id: str) -> dict[str, Any]

Remove this pipeline from the marketplace. Parameters

marketplace_id

str

required

The marketplace listing ID to remove.

Returns

returns

dict[str, Any]

\{"status": "success"\}

`move_to_folder`

Pipeline.move_to_folder(folder_id: str) -> dict[str, Any]

async Pipeline.amove_to_folder(folder_id: str) -> dict[str, Any]

Move the pipeline to a different folder. Parameters

folder_id

str

required

Target folder ID.

Returns

returns

dict[str, Any]

\{"status": "success"\}

Versioning

`revert`

Pipeline.revert(version: str) -> dict[str, Any]

async Pipeline.arevert(version: str) -> dict[str, Any]

Revert pipeline to a specific version. .. note:

The local `Pipeline` object is **not** updated after this call.
Call `Pipeline.fetch(id=...)` to get the reverted state.

Parameters

version

str

required

Semantic version string, e.g. "1.2.3".

Returns

returns

dict[str, Any]

\{"status": "success"\}

Serialization

`to_dict`

Pipeline.to_dict() -> dict

Convert Pipeline to ObjectInfo dict format for use in other nodes. Returns

returns

dict

`from_json`

Pipeline.from_json(data: dict) -> Pipeline

Parameters

data

dict

required

Returns

returns

Pipeline

`serialize_inputs`

Pipeline.serialize_inputs(inputs: dict[str, Any]) -> dict[str, Any]

Parameters

inputs

dict[str, Any]

required

Returns

returns

dict[str, Any]

Types

Configuration objects, response shapes, and enums used by the methods above.

`BumpLevel`

Members

PATCH = "patch"
MINOR = "minor"
MAJOR = "major"

`RunResult`

Response from :meth:Pipeline.run / :meth:Pipeline.arun. Fields

run_id

str

required

outputs

dict[str, Any]

required

trace_id

str

required

`RunStatus`

Response from :meth:Pipeline.run_status / :meth:RunHandler.run_status. Fields

task_id

str

required

status

Literal[in_progress, completed, failed]

required

error

str

required

result

dict[str, Any]

required

`StartResult`

Response from :meth:Pipeline.start background run initiation. Fields

task_id

str

required

trace_id

str

required

​Lifecycle

​new

​save

​deploy=True vs deploy=False — the branch model

​fetch

​list

​delete

​duplicate

​Building

​add

​add_batch

​Node catalogue

​add_node

​remove_node

​Running

​run

​bulk_run

​run_status

​Background runs

​start

​terminate

​Streaming

​stream

​Sharing & publishing

​share

​unshare

​publish

​unpublish

​move_to_folder

​Versioning

​revert

​Serialization

​to_dict

​from_json

​serialize_inputs

​Types

​BumpLevel

​RunResult

​RunStatus

​StartResult

Lifecycle

`new`

`save`

`deploy=True` vs `deploy=False` — the branch model

`fetch`

`list`

`delete`

`duplicate`

Building

`add`

`add_batch`

Node catalogue

`add_node`

`remove_node`

Running

`run`

`bulk_run`

`run_status`

Background runs

`start`

`terminate`

Streaming

`stream`

Sharing & publishing

`share`

`unshare`

`publish`

`unpublish`

`move_to_folder`

Versioning

`revert`

Serialization

`to_dict`

`from_json`

`serialize_inputs`

Types

`BumpLevel`

`RunResult`

`RunStatus`

`StartResult`