---
title: "Stream results with pipeline nodes"
description: "Author a type: pipeline node that emits many output frames for one input, compose pipeline nodes into a pipeline_mode flow, and stream the frames from the CLI and over HTTP as they are produced."
category: guide
surfaces: [cli, http-api]
related: [concepts/execution-model, guides/publish-a-flow, guides/debug-a-flow, guides/configure-an-instance, reference/http-api, reference/axiom-yaml]
last_reviewed: 2026-08-03
---

# Stream results with pipeline nodes

Most nodes answer once: one input message in, one output message out. Some
answers are not one value but a **series** — the chunks of a document, the
tokens of a model's reply, the records in a large file, the pages of an API.
Buffering a series into one response makes the caller wait for the slowest part
of it and puts the whole thing in memory at once.

A **pipeline node** streams instead. It reads a stream of input frames and emits
as many output frames as it likes, and every frame is delivered to the next node
— or to your terminal — the moment it is produced.

Prerequisites: a package you can build and push — see
[Create a node](create-a-node-go.md) for the authoring loop in your language.

## Declare the node

One knob in `axiom.yaml`:

```yaml
nodes:
  - name: Chunk
    type: pipeline          # "" or "unary" is the default
    description: Splits text into word tokens, one frame per word.
    input: TextRequest
    output: TokensResult
```

Scaffold it with the flag and you get a streaming handler and a matching
streaming unit test:

```bash
axiom create node Chunk --type pipeline --input TextRequest --output TokensResult
```

`type: pipeline` composes with everything else a node can declare —
`required_secrets`, `mutation_capable`, `kind: generic`.

## Write the handler

The handler signature is fixed per language and `axiom validate` enforces it. Go
is the odd one out: it takes a receive channel plus an `emit` callback rather
than returning an iterator.

```go
// Go — nodes/chunk.go
func Chunk(ctx context.Context, ax axiom.Context, in <-chan *gen.TextRequest, emit func(*gen.TokensResult) error) error {
    for input := range in {
        for _, w := range strings.Fields(input.GetText()) {
            if err := emit(&gen.TokensResult{Tokens: []string{w}, Count: 1}); err != nil {
                return err
            }
        }
    }
    return nil
}
```

```python
# Python — yield, do not return a list
def chunk(ax: AxiomContext, inputs: Iterator[TextRequest]) -> Iterator[TokensResult]:
    for inp in inputs:
        for word in inp.text.split():
            yield TokensResult(tokens=[word], count=1)
```

```typescript
// TypeScript
export async function* chunk(
  ax: AxiomContext,
  inputs: AsyncIterable<TextRequest>,
): AsyncGenerator<TokensResult> {
  for await (const inp of inputs) {
    for (const word of inp.text.split(/\s+/)) {
      yield new TokensResult({ tokens: [word], count: 1 });
    }
  }
}
```

```rust
// Rust
pub fn chunk<I>(ax: &dyn AxiomContext, inputs: I)
    -> impl Iterator<Item = Result<TokensResult, Box<dyn std::error::Error>>>
where I: Iterator<Item = TextRequest> { /* … */ }
```

```java
// Java
public static Stream<TokensResult> chunk(AxiomContext ax, Iterator<TextRequest> inputs) { /* … */ }
```

```csharp
// C#
public static async IAsyncEnumerable<TokensResult> Chunk(
    IAxiomContext ax,
    IAsyncEnumerable<TextRequest> inputs,
    [EnumeratorCancellation] CancellationToken ct = default) { /* … */ }
```

How many items the input stream yields depends on where the node sits:

- **Start node of a pipeline flow:** exactly one item — the flow's input.
- **Any node further down:** as many frames as its upstream emitted.
- **Direct invoke:** one item for an ordinary invoke, or **one item per line**
  when the caller streams NDJSON into it — see
  [Stream input into a node](#stream-input-into-a-node) below.

So a start node must not assume a single item: the same node reached by direct
invoke can be handed a live, unbounded stream. Write the loop, not the
`inputs[0]`.

## Give the output message its own `is_final` field

When your function returns, the platform appends a transport-level terminal
frame with an **empty payload**. A consumer therefore cannot read the transport
flag as a business signal: by the time it sees "final", the payload is gone.

If "this is the last one" matters to whoever consumes your stream, put the
marker on your own message and set it yourself:

```protobuf
message TokensResult {
  repeated string tokens = 1;
  int32 count = 2;
  // True on the LAST frame of the stream and no other.
  bool is_final = 3;
}
```

## Make every frame idempotent

Delivery is at-least-once, per frame, exactly as it is for a unary call. A
pipeline flow is durable but **not resumable**: its checkpoint is written once
per node at stream completion, so a redelivery re-runs the whole graph and
replays every frame. Upsert on a stable key derived from the frame — never
"append one row per frame".

## Test it before you push

Seed the input, collect **every** emitted frame, and assert the exact count and
order — plus `is_final` on only the last one. This is what
`axiom create node --type pipeline` scaffolds:

```go
func TestChunk(t *testing.T) {
    in := make(chan *gen.TextRequest, 1)
    in <- &gen.TextRequest{Text: "hello world"}
    close(in) // an unclosed channel leaves the node's range loop blocked forever

    var frames []*gen.TokensResult
    err := nodes.Chunk(context.Background(), ax, in, func(f *gen.TokensResult) error {
        frames = append(frames, f)
        return nil
    })
    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    if len(frames) != 2 {
        t.Fatalf("expected 2 frames, got %d", len(frames))
    }
    if frames[0].GetTokens()[0] != "hello" || frames[1].GetTokens()[0] != "world" {
        t.Errorf("frames out of order: %v", frames)
    }
}
```

Then run the real generated service and watch the frames arrive:

```bash
axiom dev
curl -N localhost:8083/nodes/Chunk -d '{"text":"hello world"}'
# data: {"tokens":["hello"],"count":1}
# data: {"tokens":["world"],"count":1}
```

`-N` is load-bearing — without it curl buffers the response and you learn
nothing about *when* each frame appeared.

## Invoke a published pipeline node

`axiom invoke` prints one JSON frame per line, as each arrives:

```bash
axiom invoke me/text-ops/Chunk --input '{"text":"hello world"}'
#   ···  streaming — pipeline node, one JSON frame per line
# {"frame_index":0,"payload":{"tokens":["hello"],"count":1},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"tokens":["world"],"count":1},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true}
```

Frames go to stdout and diagnostics to stderr, so the stream pipes straight into
`jq`. Over raw HTTP the same invoke answers with `Content-Type:
text/event-stream` — see [HTTP API](../reference/http-api.md).

## Stream input into a node

Output streaming is only half of it. The same invoke accepts a **streamed
request body**: NDJSON — one JSON object per LINE — where each line becomes its
own input frame, dispatched as it is read.

```bash
axiom invoke me/text-ops/Chunk --input-stream frames.ndjson
#   ···  streaming input — NDJSON, one frame per line
# {"frame_index":0,"payload":{"tokens":["hello"],"count":1},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"tokens":["world"],"count":1},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true,"input_frames":2}
#   ···  2 input frame(s) sent and accepted
```

`-` reads stdin, so the input can be produced live and never end:

```bash
tail -f events.ndjson | axiom invoke me/tools/StreamRecords --input-stream -
```

That last line — `sent and accepted` — is the CLI reconciling the frames it
wrote against the count the platform reports back. If they ever disagree the
invoke **fails loudly** rather than printing a clean run over a lossy one.

### Under the hood: three requests, not one

`--input-stream` drives three separate HTTP requests: it creates a stream,
attaches the output channel, then uploads frames. Each is **half-duplex** —
bytes flow one way per request.

That shape is not incidental. NDJSON up and Server-Sent Events down on ONE
request is full-duplex HTTP, and an HTTP client — including the reverse proxy in
front of the API — stops writing a request body once the response arrives. The
one-request form therefore truncated the caller's upload the moment the node
answered: 7 frames paced 3s apart delivered 4, with the run still reporting
success. Splitting the legs removes the interaction entirely.

Two consequences worth knowing:

- **The output channel replays from the beginning**, so attaching late — or
  reconnecting after a drop — still gets you every frame from the first.
- **Posting NDJSON straight to the node endpoint still works, but its response
  is buffered**: you get every frame, only after your upload has finished. Use
  the three-leg shape (or just the CLI) when you want output as it happens.

Driving it by hand sets `Content-Type: application/x-ndjson` and streams the
body with `curl -N -X POST -T -` — **not** `--data-binary @-`, which reads all of
stdin into memory before sending and therefore cannot stream at all. See
[HTTP API](../reference/http-api.md#stream-input-into-a-pipeline-node) for the
three legs in full.

**Pipeline nodes only.** A unary node consumes exactly one input, so later
frames would be silently discarded; streaming into one is rejected with a clear
error instead.

**Limits.** 16 MiB per frame, 64 MiB and 100 000 frames per stream, and the
same 5-minute node ceiling as any other invoke. A frame that fails validation is
rejected at that frame — the stream ends with an error naming the frame index,
and output already emitted for earlier frames stands. A streamed direct invoke
is synchronous and is not redelivered: if the connection drops, re-stream.

### Live mode: full duplex, and nothing written down

`--input-stream` is durable bulk ingestion. When what you want is an
**interactive session** — a token relay, a live tail, a dialogue where you send
the next thing based on what came back — use `--live`:

```bash
axiom invoke me/tools/Chat --live -
```

That opens **one WebSocket**. Frames go up and come down on the same connection
at the same time, so you can send frame *k+1* in response to frame *k*'s output.
Under the hood it is the same node stream, the same frame JSON, the same
per-frame validation and reshape — only the client-edge transport differs.

The trade is durability, and it is total:

> **A live session is ephemeral.** No durable execution row, no replay, no
> redelivery, and it is pinned to a single ingress pod — if that pod restarts,
> the session dies. `axiom executions get` will never find it. Recovery means
> reconnecting and re-sending.

Use `--input-stream` when you need a record of what was ingested. Use `--live`
when the value is in the round trip and re-sending is cheap.

**A slow reader loses its session.** The platform buffers only a small bounded
amount of output per live session; a client that stops draining is closed with a
reason rather than buffered indefinitely. Live sessions also cap **output** at
64 MiB and hold one inflight execution slot for their lifetime. The per-frame
16 MiB cap and the 5-minute ceiling are the same as everywhere else.

**`--live` needs a `write`-scoped token,** like any other invoke — running a node
is a write, even though the WebSocket handshake itself is a `GET`.

### Incremental consumption is per-language

Every language receives every frame. What differs is **when** — and it is a
property of the generated node service, not of your handler:

| Language | When your handler sees frame *k* |
|---|---|
| Go | as it arrives |
| Python | as it arrives |
| TypeScript | as it arrives |
| Rust | after the whole input stream ends (frames are buffered first) |
| Java | after the whole input stream ends |
| C# | after the whole input stream ends |

If your node has to react to frame *k* before frame *k+1* is sent — a token
relay, a live tail, an interactive session — write it in Go, Python or
TypeScript. For a bounded batch that simply arrives as N frames, any language
works.

## Compose pipeline nodes into a streaming flow

A flow that should stream sets one top-level field:

```yaml
# doc-chunk-stream.flow.yaml
name: me/doc-chunk-stream
version: 0.1.0
pipeline_mode: true
nodes:
  - alias: chunk
    package: me/text-ops@0.1.0
    node: Chunk
  - alias: prep
    package: me/embed-prep@0.1.0
    node: Prepare
edges:
  - from: chunk
    to: prep
```

Compile it as usual, then run it with `--stream`:

```bash
axiom flow compile doc-chunk-stream.flow.yaml
axiom flow run 01J… --stream -d '{"text":"hello world"}'
# {"tokens":["hello"],"count":1}
# {"tokens":["world"],"count":1}
# ✓ succeeded  (execution 4bf9…, 2 frame(s))
```

`--stream --json` prints the full frame envelope instead — including
`frame_index` and `is_final`, which is what lets you assert the sequence.

**Without `--stream`, a pipeline flow is sent to the unary endpoint**, which
returns a single aggregated result and discards every later frame. The CLI
detects the mismatch and warns loudly; that warning means the output you are
looking at is not the flow's output. Re-run with `--stream`.

The raw HTTP form is `POST /v1/flows/invoke/stream`:

```bash
curl -N -X POST 'https://api.axiomide.com/invocations/v1/flows/invoke/stream' \
  -H "Authorization: Bearer $AXIOM_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"graph_id":"01J…","input":{"text":"hello world"}}'
```

## Mixed graphs are allowed — with one rule

`pipeline_mode: true` does **not** require every node in the flow to be
`type: pipeline`. The pipeline executor drives a unary node over the same stream
using single-frame semantics, so mixing compiles and runs.

The rule that decides your design: **a unary node receives the first frame and
every later frame is discarded.** A unary node is therefore fine as the start of
a pipeline flow, or anywhere upstream of the first real stream — and is silent
data loss anywhere downstream of one.

## Instances of a pipeline generic

A [generic node](configure-an-instance.md) can be `type: pipeline`, and an
Instance you bind to it is an ordinary pipeline node: it streams. The facade
reshape runs **once per frame** — your input facade on the way in, your output
facade on every frame that comes back — so each frame arrives at the caller in
the shape the Instance declares, not the generic's:

```bash
axiom instance create me/docs/Tokens \
  --generic-package me/text-ops --generic-node Split \
  --input-message Doc   --input-field text=string \
  --input-map body=text \
  --output-message Token --output-field position=int64 --output-field value=string \
  --output-map position=index --output-map value=body

axiom invoke me/docs/Tokens --input '{"text":"hello world"}'
# {"frame_index":0,"payload":{"position":"1","value":"seen: hello"},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"position":"2","value":"seen: world"},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true}
```

The same Instance node drops into a `pipeline_mode` flow like any other pipeline
node, and streams there too.

One consequence of the empty terminal frame above: the reshape is applied to
**data** frames only. The platform's terminal frame carries no payload, so there
is nothing there for your declared facade fields to honor — which is another
reason to put your own `is_final` on your own message.

## What a pipeline flow cannot do

These are rejected at compile time, with a diagnostic:

| Not available | Instead |
|---|---|
| `input_facade` / `output_facade` | A pipeline flow's contract is the raw start-node input and the raw terminal-node output, one message per frame |
| `mocks:` | Test with the real nodes — prefer cheap or no-auth nodes while iterating |
| `subflow:` nodes | Inline the subflow's nodes into the pipeline flow |
| A `loop: true` edge touching a pipeline node | Loops may only connect unary nodes |
| Client-**streamed input** into the flow | A flow run takes one stored input (that is what makes it durable and redeliverable). To feed a live client stream, invoke the pipeline NODE directly with NDJSON — see [Stream input into a node](#stream-input-into-a-node) |

Absent at run time: human-in-the-loop pauses, node retry, mutation, gates and
quorum joins. Fan-out and fan-in both work, and fan-in is order-preserving —
frames are merged positionally by their source frame index. Conditions and edge
adapters are evaluated **once per frame**, so a condition can drop individual
frames out of a stream.

Limits: a single frame must be ≤24 MiB, and a node has a 5-minute ceiling for
its whole stream.

## Prove it actually streamed

The execution timeline is the definitive evidence — a flow that silently
degraded to one aggregated result looks completely different there:

```bash
axiom executions get <execution-id>          # the event sequence
axiom executions get <execution-id> --json   # …with each event's full payload
```

Look for per-frame `NODE_FRAME` and `NODE_INPUT_FRAME` events, per-frame
per-edge `EDGE_EVALUATED`, and `NODE_STREAM_COMPLETE` — which replaces
`NODE_COMPLETED` for a pipeline node. The human view lists the events; the
per-event counters (`total_frames`, `total_bytes`, `frames_dropped`) are in the
payload, so use `--json` to read them.

`frames_dropped` is the one to watch. A frame whose outgoing edge's condition or
adapter *throws* is **dropped**: the stream gets shorter and the flow still
succeeds. Each drop is also recorded as an `EDGE_EVALUATED` event carrying the
`frame_index` and an `error` naming the failure (and deliberately no
`condition_pass` — the condition never returned a verdict), so you can see
exactly which frame died on which edge. Assert exact frame counts in your tests
and a non-zero `frames_dropped` will never surprise you in production.

One wrinkle worth knowing: a **unary** node inside a pipeline flow reports
`NODE_STREAM_COMPLETE` before it forwards its single result, so its
`frames_dropped` is always 0. For those nodes the per-drop `EDGE_EVALUATED`
event is the record.

See [Debug a flow](debug-a-flow.md) for the rest of the timeline surfaces.