Defining Flows

Create sequences of checkpointed steps.

After you create your first workflow, you can start defining flows for it. This article describes supported syntax and configuration options.

First: Install the Render SDK

The Render SDK is currently available for TypeScript and Python.

SDKs for additional languages are planned for future releases.

The Render SDK is required to define and register flows.

From your TypeScript project directory:

shell

$ npm install @renderinc/sdk@latest

(Or pnpm install, bun add, etc.)

If you already have the SDK installed, make sure you're using version 1.7.0 or later:

shell

$ npm install @renderinc/sdk@latest

After installing, make sure @renderinc/sdk is listed as a dependency in your package.json file at version ^1.7.0 or later.

From your Python project directory:

shell

$ pip install render

If you already have the SDK installed, make sure you're using version 1.7.0 or later:

shell

$ pip install --upgrade render

After installing, make sure to add render>=1.7.0 as a dependency in your application's requirements.txt, pyproject.toml, or equivalent.

Basic example

Let's start with a "minimum viable workflow" that defines a single flow. This flow formats a message in two steps:

index.ts

typescript

import { flow, type FlowContext } from '@renderinc/sdk/workflows'

const formatMessage = flow(
  { name: 'formatMessage' },
  async function formatMessage(ctx: FlowContext, message: string): Promise<string> {
    const trimmed = await ctx.step('trim', (text: string) => text.trim(), [message])
    const uppercase = await ctx.step('uppercase', (text: string) => text.toUpperCase(), [trimmed])
    return uppercase
  }
)

This includes everything required to define a workflow:

We import the flow function and FlowContext type from the Render SDK.
We call flow(...) to register a function as a flow (in this case, formatMessage).
- The first parameter to flow(...) is an object containing configuration options (only name is required). For other supported options, see Configuring flow behavior.
- The second parameter is the flow function itself. This function must take a FlowContext object as its first argument (Render provides this object to each flow run automatically).
We call ctx.step() to run each step and obtain its return value.

main.py

python

from render import FlowContext, Workflows

app = Workflows()

@app.flow
def format_message(ctx: FlowContext, message: str) -> str:
  trimmed = ctx.step("trim", lambda text: text.strip(), message)
  uppercase = ctx.step("uppercase", lambda text: text.upper(), trimmed)
  return uppercase

if __name__ == "__main__":
  app.start()

This includes everything required to define a workflow:

We import the Workflows class and FlowContext type from the Render SDK, then initialize a Workflows object as app.
We apply the @app.flow decorator to register a function as a flow (in this case, format_message).
- The decorator accepts a number of optional parameters (see Configuring flow behavior).
- The flow function must take a FlowContext object as its first argument (Render provides this object to each flow run automatically).
We call ctx.step() to run each step and obtain its return value.
We call app.start() in our code's entry point.
- On Render, this is what kicks off the flow registration process and the execution of each run.

In Python, flow functions and step functions must be synchronous (def, not async def). Each call to ctx.step() returns when the step completes.

Organizing definitions

You can define your workflow's flows across multiple files in your project repo:

Define and export flows from individual files, then import those files from your workflow's entry point:

text-flows.ts

typescript

import { flow, type FlowContext } from '@renderinc/sdk/workflows'

export const formatMessage = flow(
  { name: 'formatMessage' },
  async function formatMessage(ctx: FlowContext, message: string): Promise<string> {
    const trimmed = await ctx.step('trim', (text: string) => text.trim(), [message])
    const uppercase = await ctx.step('uppercase', (text: string) => text.toUpperCase(), [trimmed])
    return uppercase
  }
)

index.ts

typescript

import './text-flows'

By setting your workflow's start command to run the JS output of index.ts, you ensure that the imported definitions are registered. Add imports for any other files that define flows or tasks.

Initialize a Workflows object in each file where you define flows, then incorporate those objects into the one in your workflow's entry point:

text_flows.py

python

from render import FlowContext, Workflows

app = Workflows()

@app.flow
def format_message(ctx: FlowContext, message: str) -> str:
  trimmed = ctx.step("trim", lambda text: text.strip(), message)
  uppercase = ctx.step("uppercase", lambda text: text.upper(), trimmed)
  return uppercase

main.py

python

from render import Workflows
from text_flows import app as text_app

app = Workflows.from_workflows(text_app) 

if __name__ == "__main__":
  app.start()

To register all of your flows and tasks, your workflow's entry point (commonly main.py) imports and incorporates the Workflows apps from each other file using the Workflows.from_workflows() method. Pass additional apps as additional positional arguments.

Naming flows

Tasks and flows share a namespace. Use a name that is unique across all tasks and flows in the workflow. In TypeScript, registering another task or flow with the same name replaces the earlier definition. In Python, duplicate names cause a registration error.

Step names do not register separate definitions. You provide them when running steps.

Function arguments

Every flow function must take a FlowContext object as its first argument:

typescript

import { flow, type FlowContext } from '@renderinc/sdk/workflows'

const myFlow = flow(
  { name: 'myFlow' },
  async function myFlow(ctx: FlowContext): Promise<number> { 
    // ...
  }
)

python

from render import FlowContext, Workflows

app = Workflows()

@app.flow
def my_flow(ctx: FlowContext) -> int: 
  # ...

Render passes this object to each flow run automatically.

A flow function can define any number of additional arguments. This example flow takes three additional arguments of different types:

typescript

const myFlow = flow(
  { name: 'myFlow' },
  async function myFlow(ctx: FlowContext, a: number, b: string, c: boolean): Promise<number> { 
    // ...
  }
)

python

@app.flow
def my_flow(ctx: FlowContext, arg1: int, arg2: str, arg3: bool) -> int: 
  # ...

You provide values for these non-ctx arguments whenever you trigger a flow run.

You can set default values for arguments:

typescript

const myFlow = flow(
  { name: 'myFlow' },
  async function myFlow(ctx: FlowContext, arg1: number = 3): Promise<number> { 
    // ...
  }
)

python

@app.flow
def my_flow(ctx: FlowContext, arg1: int = 3) -> int: 
  # ...

Argument format requirements

A flow's argument types (and its return type) must be JSON-serializable.
- Argument values are passed to a flow run's instance as JSON, and a flow run's result is returned as JSON.
The combined size of all arguments passed to a single flow run cannot exceed 4 MB.
- Otherwise, the request fails with an error.

Configuring flow behavior

Compute plan

By default, flow runs execute on the Flex compute plan (up to 1 CPU, up to 4 GB RAM).

You can specify a different compute plan to use for a given flow with the following syntax:

typescript

const formatMessage = flow(
  {
    name: 'formatMessage',
    plan: '2c-4g' 
  },
  async function formatMessage(ctx: FlowContext, message: string): Promise<string> {
    const trimmed = await ctx.step('trim', (text: string) => text.trim(), [message])
    return await ctx.step('uppercase', (text: string) => text.toUpperCase(), [trimmed])
  }
)

python

@app.flow(
  plan="2c-4g" 
)
def format_message(ctx: FlowContext, message: str) -> str:
  trimmed = ctx.step("trim", lambda text: text.strip(), message)
  return ctx.step("uppercase", lambda text: text.upper(), trimmed)

The following compute plans are available for flows:

CPU	RAM	Plan ID
Up to 1 CPU	Up to 4 GB	`flex`
2	4 GB	`2c-4g`
2	8 GB	`2c-8g`
4	8 GB	`4c-8g`
4	16 GB	`4c-16g`
8	32 GB	`8c-32g`

Providing starter or standard is equivalent to providing flex.

Because steps execute on the same instance as their flow, you do not specify per-step compute plans.

See pricing details.

Timeout

By default, each execution attempt of a flow run times out after 2 hours if its function hasn't returned yet. You can override this on a per-flow basis to any value between 30 seconds and 24 hours.

Provide your flow's timeout to flow(...) via the timeoutSeconds option:

typescript

const formatMessage = flow(
  {
    name: 'formatMessage',
    timeoutSeconds: 86400 
  },
  async function formatMessage(ctx: FlowContext, message: string): Promise<string> {
    const trimmed = await ctx.step('trim', (text: string) => text.trim(), [message])
    return await ctx.step('uppercase', (text: string) => text.toUpperCase(), [trimmed])
  }
)

Provide your flow's timeout to the @app.flow decorator via the timeout_seconds parameter:

python

@app.flow(
  timeout_seconds=86400 
)
def format_message(ctx: FlowContext, message: str) -> str:
  trimmed = ctx.step("trim", lambda text: text.strip(), message)
  return ctx.step("uppercase", lambda text: text.upper(), trimmed)

Each flow recovery starts a new execution attempt with a fresh timeout. A run's total duration can therefore exceed its configured timeout.

In Python, you can also set default_plan and default_timeout on the Workflows object. Individual flows can override these settings.

Configuring step behavior

Within a flow definition, use the FlowContext object's step method to run a plain function as a step. Steps run inside the flow's process, on the same compute as the flow. They are not registered as separate definitions, but each step and its attempts appear in the flow run's execution viewer.

The basic example runs two steps to format a message:

typescript

const trimmed = await ctx.step('trim', (text: string) => text.trim(), [message])
const uppercase = await ctx.step('uppercase', (text: string) => text.toUpperCase(), [trimmed])

Each call to ctx.step() provides a step name, a function, and an array of arguments to pass to that function.

For all supported options, see FlowContext.step().

python

trimmed = ctx.step("trim", lambda text: text.strip(), message)
uppercase = ctx.step("uppercase", lambda text: text.upper(), trimmed)

Each call to ctx.step() provides a step name, a function, and its positional arguments. The name is required for every step, including lambdas. Keyword arguments are reserved for step options; pass the function's inputs positionally.

For all supported options, see FlowContext.step().

The step's function receives its arguments without a context argument. Pass a plain function, not a registered task or flow definition.

Step arguments and return values must be JSON-serializable. For example, TypeScript-defined steps can't return Date objects, class instances, NaN, or Infinity.

Step results pass through JSON serialization, even during the initial execution.

A step function cannot make calls on the flow's context.

Flows cannot run registered tasks. To share code between a flow and a task, define a plain function to call as a step and call the same function from a registered task.

For step limits, see Limits and pricing.

Retry settings

A step can automatically retry if its initial attempt fails, such as when the step function raises an exception or throws an error.

Each retry reruns the step's function on the flow's instance. If a step reaches its retry limit with no successful attempts, the step is marked as failed. An unhandled step failure fails the flow run.

If a step's return value fails JSON serialization, the step fails without retrying.

Default retry behavior

If you don't customize a step's retry behavior, it does not retry (one total attempt).

Customizing retries

You can customize retry behavior on a per-step basis.

Provide retry settings to ctx.step() with the following syntax:

typescript

import { flow, type FlowContext } from '@renderinc/sdk/workflows'

function flipCoin(): string {
  if (Math.random() < 0.5) {
    throw new Error('Flipped tails! Retrying.')
  }
  return 'Flipped heads!'
}

const flipCoinFlow = flow(
  { name: 'flipCoinFlow' },
  async function flipCoinFlow(ctx: FlowContext): Promise<string> {
    return await ctx.step('flipCoin', flipCoin, [], {
      retry: {
        maxRetries: 3, // Retry up to 3 times (i.e., 4 total attempts)
        waitDurationMs: 1000, // Set a base retry delay of 1 second
        backoffScaling: 1.5 // Increase delay by 50% after each retry (1s, 1.5s, 2.25s)
      }
    })
  }
)

python

from render import FlowContext, Retry, Workflows
import random

app = Workflows()

def flip_coin() -> str:
  if random.random() < 0.5:
    raise Exception("Flipped tails! Retrying.")
  return "Flipped heads!"

@app.flow
def flip_coin_flow(ctx: FlowContext) -> str:
  return ctx.step(
    "flip_coin", flip_coin,
    retry=Retry(
      max_retries=3, # Retry up to 3 times (i.e., 4 total attempts)
      wait_duration_ms=1000, # Set a base retry delay of 1 second
      backoff_scaling=1.5 # Increase delay by 50% after each retry (1s, 1.5s, 2.25s)
    )
  )

This contrived example defines a step that "flips a coin" and raises an exception/error when it "flips tails", causing the attempt to fail. The step retries according to its settings, up to three times.

Parallelizing work within a step

Each ctx.step() call must finish before the next starts:

In TypeScript, await each ctx.step() call before making the next. Concurrent calls, including calls in Promise.all, throw an error.
In Python, each ctx.step() call blocks until the step finishes.

You can still perform concurrent work inside a single step, such as fetching multiple records from an API:

Use Promise.all or a similar concurrency utility inside the step function. In this example, fetchRecord is a plain asynchronous function defined elsewhere that fetches one record:

typescript

import { flow, type FlowContext } from '@renderinc/sdk/workflows'

async function fetchBatch(recordIds: string[]) {
  return await Promise.all(recordIds.map(fetchRecord)) 
}

const fetchRecords = flow(
  { name: 'fetchRecords' },
  async function fetchRecords(ctx: FlowContext, recordIds: string[]) {
    return await ctx.step('fetchBatch', fetchBatch, [recordIds])
  }
)

Use ThreadPoolExecutor to overlap synchronous I/O inside the step function. In this example, fetch_record is a plain synchronous function defined elsewhere that fetches one record:

python

from concurrent.futures import ThreadPoolExecutor
from render import FlowContext, Workflows

app = Workflows()

def fetch_batch(record_ids: list[str]):
  with ThreadPoolExecutor(max_workers=5) as pool: 
    return list(pool.map(fetch_record, record_ids)) 

@app.flow
def fetch_records(ctx: FlowContext, record_ids: list[str]):
  return ctx.step("fetch_batch", fetch_batch, record_ids)

Each batch above is a single step. Render saves the batch's combined result, not the result of each individual request. If the step retries or re-executes after an interruption, the entire batch can run again.

All work within a step shares the flow's compute plan. A plan with more CPU provides more processing capacity for CPU-intensive work across multiple workers or processes. More RAM supports work with larger memory requirements. Your code controls how many workers or threads to create; changing the compute plan does not automatically parallelize your code.

Overlapping I/O with Promise.all or ThreadPoolExecutor does not require multiple CPUs. To use multiple CPUs for CPU-intensive work, consider Node.js worker threads, Python process pools, or libraries that use multiple CPUs.

Determinism and side effects

An interrupted step whose outcome was not recorded can execute again, so side effects should be safe to repeat.

Flow code must be deterministic: it must make the same ctx calls in the same order with the same inputs on replay. Put I/O, random values, and reads of the current time inside steps.

If a step's name, order, or inputs differ from the recorded history, the SDK raises a DeterminismError. This fails the flow run even if you catch the error.

Flow replay

You do not set a retry policy for a flow. If an infrastructure issue (such as instance failure) interrupts a flow run, Render automatically replays the run on a new instance. This executes the flow function from the beginning, using the recorded return values of any successfully completed steps instead of executing those steps again.

Each replay creates a new execution attempt of the original flow run. The initial attempt has "attempt number" 0, the first replay attempt has number 1, and so on. These are separate from a step's retry attempts.

All attempts of a flow run use the workflow version the run started on, even if you deploy a new version.

The following errors fail a flow run without replay:

Out-of-memory (OOM) errors
Timeouts
Uncaught errors in the flow function
Determinism errors
Errors from exceeding a flow limit

Canceling a flow run ends it with a canceled status. Canceled runs do not replay.