# Defining Flows — Create sequences of checkpointed steps.

After you [create your first workflow](/workflows-tutorial), 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.

**Tab: TypeScript**

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.

**Tab: Python**

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:

**Tab: TypeScript**

```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:

1. We import the `flow` function and `FlowContext` type from the Render SDK.
2. 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](#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).
3. We call `ctx.step()` to run each step and obtain its return value.

**Tab: Python**

```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:

1. We import the `Workflows` class and `FlowContext` type from the Render SDK, then initialize a `Workflows` object as `app`.
2. 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](#configuring-flow-behavior)).
    - The flow function must take a `FlowContext` object as its first argument (Render provides this object to each flow run automatically).
3. We call `ctx.step()` to run each step and obtain its return value.
4. 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:

**Tab: TypeScript**

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

```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
  }
)
```

```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.

**Tab: Python**

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

```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
```

```python
from render import Workflows
from text_flows import app as text_app

app = Workflows.from_workflows(text_app) # highlight-line

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()`](/workflows-sdk-python#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](#configuring-step-behavior).

## Function arguments

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

**Tab: TypeScript**

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

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

**Tab: Python**

```python
from render import FlowContext, Workflows

app = Workflows()

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

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:

**Tab: TypeScript**

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

**Tab: Python**

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

You provide values for these non-`ctx` arguments whenever you [trigger a flow run](/workflows-running).

You can set default values for arguments:

**Tab: TypeScript**

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

**Tab: Python**

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

### 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:

**Tab: TypeScript**

```typescript
const formatMessage = flow(
  {
    name: 'formatMessage',
    plan: '2c-4g' // highlight-line
  },
  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])
  }
)
```

**Tab: Python**

```python
@app.flow(
  plan="2c-4g" # highlight-line
)
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`*](/workflows-limits#the-flex-compute-plan) |
| 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](/workflows-limits#compute-plans).

### 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*.

**Tab: TypeScript**

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

```typescript
const formatMessage = flow(
  {
    name: 'formatMessage',
    timeoutSeconds: 86400 // 24 hours in seconds // highlight-line
  },
  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])
  }
)
```

**Tab: Python**

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

```python
@app.flow(
  timeout_seconds=86400 # 24 hours in seconds # highlight-line
)
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`](/workflows-sdk-python#constructor) 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](#basic-example) runs two steps to format a message:

**Tab: TypeScript**

```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()`](/workflows-sdk-typescript#flowcontext-step).

**Tab: Python**

```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()`](/workflows-sdk-python#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](/workflows-limits#additional-limits).

### 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:

**Tab: TypeScript**

```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)
      }
    })
  }
)
```

**Tab: Python**

```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:

**Tab: TypeScript**

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)) // highlight-line
}

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

**Tab: Python**

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: # highlight-line
    return list(pool.map(fetch_record, record_ids)) # highlight-line

@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](#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](https://nodejs.org/api/worker_threads.html), [Python process pools](https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor), 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](#timeout)
- Uncaught errors in the flow function
- [Determinism errors](#determinism-and-side-effects)
- Errors from exceeding a [flow limit](/workflows-limits#additional-limits)

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

---

##### Appendix: Glossary definitions

###### flow

A function that executes a durable sequence of *steps* on its own *instance*. Belongs to a *workflow*.

If a flow's *run* is interrupted, it can reuse the results from completed steps when it replays.

Related article: https://render.com/docs/workflows-defining-flows.md

###### step

A function representing one unit of work within a *flow*.

Results of completed steps are saved for reuse if a flow's *run* is interrupted and replays.

Related article: https://render.com/docs/workflows-defining-flows.md

###### compute plan

Specifies the CPU and RAM available to your service's *instances*.

Common compute plans for a new web service include:

- *`free`*: 0.1 CPU / 512 MB RAM
- *`0.5c-512mb`*: 0.5 CPU / 512 MB RAM
- *`1c-2g`*: 1 CPU / 2 GB RAM

Compute plans were previously known as *instance types*.

Related article: https://render.com/docs/compute-plans.md

###### run

A single execution of a *task* or *flow*. Belongs to a *workflow*.

A run spins up in its own *instance*, executes, returns a value, and is deprovisioned.

Related article: https://render.com/docs/workflows-running.md