> For the complete documentation index, see [llms.txt](https://docs.contextual.io/documentation-and-resources/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.contextual.io/documentation-and-resources/components-and-data/flows/node-reference/stream/source-from.md).

# Source From

Turns a list of items into a lazy source that source-aware downstream nodes can process one item at a time.

Most nodes handle one message at a time: one message in, one message out. Source From also sends one Node-RED message, but attaches a lazy source at `msg._contextual.source`. Source-aware nodes transform or inspect each source item when a *Sink Collect* consumes the source.

Source handling is implemented per node, not by palette category. Active source-aware nodes include the Stream nodes documented in this section, Contextual record and attachment nodes, Contextual request nodes, AI Generate, JSONata, and Log Tap. Nodes without explicit source handling do not automatically run once per source item; they receive the carrier message through normal Node-RED processing.

Use a **Source From** node when one message arrives with an iterable value whose items should be processed individually by source-aware nodes. It can also parse JSON Lines content or a zip archive into source items.

Source From is not the only way a stream begins. Some nodes open one themselves — [a *Query Object*](https://docs.contextual.io/documentation-and-resources/components-and-data/flows/node-reference/object/query-object) node with **Unfold all pages to source** enabled emits its records as a stream directly, and its label in the editor changes to \[Source] Query Objects. Where an upstream node already produces a stream, a Source From node is not needed. Use Source From when the list arrives as an ordinary message property.

A stream that is opened must also be closed. Every source stream needs a *Sink Collect* node downstream of it — see *Every source stream must reach a Sink Collect* below.

To configure a **Source From** node in the *Flow Editor*, drag it from the **Stream** group in the Palette to the Workspace canvas, then double-click the node to open and configure its behavior.

## Choosing between Source From, Loop, and Split/Join

Source From is one of three ways to process a list one item at a time in a Flow. Which one to reach for depends on what you already have and what you need back.

* **A plain list you want to process item-by-item, then optionally collect a result** — use Source From with the other Stream nodes (*Filter*, *Batch*, *Map*, and *Sink Collect*). No feedback wiring is needed: each item flows downstream as its own message, and the editor enforces that the stream closes, reporting a lint error if a stream is opened but no *Sink Collect* is reachable.
* **A fixed number of iterations, a condition to loop until, or enumerating something that isn't already a flat list** (an object's keys, a Map, a Set) — use a *Loop* node instead. A Loop node requires you to wire its iteration output back into itself, and it supports running while or until a condition is true rather than only draining a list once.
* **A one-off split and rejoin of an array or string using the classic Node-RED building blocks** — use *Split* and *Join*. These tag each message with `msg.parts` rather than opening an explicit stream, which is lighter-weight for a single split/transform/rejoin, but nothing in the Flow Editor currently flags a Split whose messages are never rejoined the way an unclosed source stream is flagged — that mismatch surfaces at runtime instead.

As a rule of thumb: reach for Source From when you want a source pipeline without feedback wiring. The editor's lint check verifies that at least one *Sink Collect* is reachable downstream, but it does not guarantee that every branch consumes the source. Reach for Loop when iteration depends on a condition rather than a fixed list, and Split/Join for ordinary message fan-out and reassembly.

## Properties

* **Name** — optional display name for the node in the editor. By default the node shows a label of "Source From"; the value you provide here replaces it, making your flow easier to read.
* **Input** — the `msg` property from which the stream is created. The editor accepts only message-property paths. It defaults to `msg.payload`.
* **Parser** — how the **Input** value is decoded into items before streaming. Defaults to None.
  * *None* — the **Input** value is already a list of items and is streamed as-is
  * *JSONL* — the **Input** value is JSON Lines content: newline-delimited JSON, one object per line. Each line becomes one item. Intended for raw content read from elsewhere, such as a .jsonl file retrieved with a Read Cloud node or a streaming API response, rather than a value entered in the editor.
  * *Unzip* — the **Input** value is a zip archive. Each file entry becomes one `[path, Buffer]` item; non-file entries are skipped

## Setting the Input path

With Parser set to None, **Input** must point to a Node `Readable`, iterable, or async iterable. For a wrapper object, point Input at the iterable property itself.

A plain non-iterable object causes Source From to report an error and send no output; it is not treated as a single item.

Upstream nodes rarely place a bare list at msg.payload. A *Query Object* node with a **Page Size** of 2 or more writes its results as {items: \[...]}, so the correct **Input** is payload.items, not payload. To confirm exactly where the list sits, before wiring Source From, place a *Log* node after the upstream node and read the Debug panel.

## What the stream carries

Each item is emitted as its own message with the item written to msg.payload. Items are passed through unchanged; if the source list contains records, each message carries the complete record, including any platform-added properties such as \_metaData. Source From does not reshape or filter the items.

Items are emitted in the order they appear in the **Input** list, and that order is preserved through to the collected output.

## Every source stream must reach a Sink Collect

A node that opens a source stream must have a path to a *Sink Collect* node downstream of it. This is enforced by the editor, not at runtime: if no such path exists, the node that opened the stream is marked with an error and the flow reports a lint error.

```
Lint Errors:

- Does not reach one of the following nodes: 'sink-collect'
```

The rule applies to any node that opens a stream, not only Source From. A *Query Object* node with **Unfold all pages to source** enabled reports the same error on itself when nothing downstream collects its stream.

Because it is a lint error rather than a runtime failure, you find out while wiring the flow rather than when a request hangs. Add the Sink Collect node and the error clears.

## Concurrency

Items are emitted one at a time, in list order, and nodes inside the stream run once per item.

Source From does not expose any setting to process items concurrently, and nodes placed inside the stream do not gain one by being there. A Map node wired between a stream source and a Sink Collect shows only its usual properties.

The **Source** tab that Object nodes such as *Put Object* and *Query Object* provide is a feature of those nodes, not of the stream itself, and is not granted to arbitrary nodes by placing them inside a source stream. On a node that has it, **Parallelism** controls how many stream items that node processes at once:

* *None* processes items sequentially, one at a time.
* *Ordered* processes up to **Max** items concurrently, but still emits results in the order the items arrived.
* *Unordered* processes up to **Max** items concurrently and emits each result as soon as it finishes, so a slow item can be overtaken by faster ones behind it.

**Max** sets the concurrency ceiling for Ordered and Unordered; it has no effect when Parallelism is None. These settings only take effect when the node is receiving a source stream from an upstream node — for example, downstream of a Source From node — not when the node runs against a single message.

## Example

A weather application stores readings as records of the **weather app** Object Type. The `/report` endpoint returns up to 10 matching readings from one query page in a single response, processing each record individually on the way.

```
[GET] /report → Query Objects → Source From → Sink Collect → HTTP Response
```

### Configure the Query Objects node

Set **Type Id** to weather app, **Query** to {} to match all records, **Page Size** to 10, and enable **Include Total**.

With a page size of 2 or more, the node outputs an object containing an items array:

```json
{
  "items": [
    {
      "temperature": 12,
      "stationId": "Dallas",
      "id": "jbKBEsasrH3Tjqy6oXeWbA",
      "datetime": "2026-08-05T17:02:00.850Z",
      "_metaData": {}
    },
    {
      "temperature": 13,
      "stationId": "Irving",
      "id": "oQsSt3BfTSAwRcAXpdBmNt",
      "datetime": "2026-08-05T17:02:08.808Z",
      "_metaData": {}
    }
  ],
  "totalCount": 10
}
```

### Configure the Source From node

Set **Input** to payload.items and leave **Parser** as None.

The array is nested under items, so payload alone would stream the wrapper object as a single item.

### Add a Sink Collect node

For this example, leave **Input** and **Output** at their `msg.payload` defaults. Sink Collect can read from one message property and write the collected array to another; `msg.payload` is not required. Using the defaults here closes the stream and clears the lint error on Source From.

### Confirm the split

Place a Log node between Source From and Sink Collect, then request /report. The Debug panel shows one entry per record, each carrying a single reading:

```
{ temperature: 12, stationId: "Dallas", id: "jbKBEsasrH3Tjqy6oXeWbA", datetime: "2026-08-05T17:02:00.850Z", _metaData: object }

{ temperature: 13, stationId: "Irving", id: "oQsSt3BfTSAwRcAXpdBmNt", datetime: "2026-08-05T17:02:08.808Z", _metaData: object }

{ temperature: 14, stationId: "Frisco", id: "52KsVsLoz8opZ6K6UtaxE9", datetime: "2026-08-05T17:02:14.947Z", _metaData: object }
```

Ten records produce ten entries. A Log node placed after Sink Collect shows a single entry containing all ten.

Any node placed between Source From and Sink Collect runs once per record, so a Map or HTTP Request node in that position operates on one reading at a time rather than on the whole list.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation by asking a question.

Perform an HTTP GET request on the following URL with the `ask` and `goal` query parameters:

```
GET https://docs.contextual.io/documentation-and-resources/components-and-data/flows/node-reference/stream/source-from.md?ask=<question>&goal=<user_goal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is what the user is ultimately trying to achieve, the reason they need the answer. Sharing it helps GitBook give you a better, more relevant answer. A goal is most helpful when it describes the outcome the user wants rather than restating the question. For example, with `ask=how do I create an API token`, a goal like `automate deployments from our CI pipeline` lets GitBook tailor the answer to that use case.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.