Defining Workflow Tasks

Specify the tasks that your apps can run.

Render Workflows are in limited early access.

During the early access period, the Workflows API and Python SDK might introduce breaking changes.

After you create your first workflow, you can start defining your own tasks. This article describes supported syntax and configuration options.

Basic example

The Workflows SDK is currently available only for Python.

SDKs for other languages are coming soon. It is not currently possible to define tasks in languages besides Python.

Define a task in your workflow service by applying the @task decorator to any function, like so:

python

from render_sdk.workflows import task, start

@task 
def calculate_square(a: int) -> int:
  return a * a

if __name__ == "__main__":
  start()

This defines a basic task named calculate_square that takes a single integer argument and returns its square.

For all supported @task options, see the Python SDK reference.

Task arguments

A task's function can define any number of arguments. This example task takes three arguments of different types:

python

@task
def my_task(arg1: int, arg2: str, arg3: bool) -> int:
  # ...

Task arguments are positional. Whenever you run a task, you provide its arguments in a JSON array in the same order as they appear in the task's function signature:

python

started_run = await client.workflows.run_task(
  task_identifier="my-workflow/my-task",
  input_data=[1, "hello", True] 
)

Argument and return types must be JSON-serializable. Your applications provide task arguments in a JSON array via the Render API, and a task's result is also returned as JSON.

All task arguments are required. If you attempt to run a task with missing arguments (or too many arguments), the task will fail. Argument values can be null, as long as your task logic supports this.

Retry logic

Your tasks can automatically retry if a run fails. A task run is considered to have failed if its function raises an exception instead of returning a value.

Retries are useful for tasks that might be affected by temporary failures, such as network errors or timeouts.

Default retry behavior

By default, tasks use the following retry logic:

Retry up to 3 times (i.e., 4 total attempts)
Wait 1 second before attempting the first retry
Double the wait time after each retry (i.e., one second, two seconds, four seconds)

Customizing retries

You can customize retry behavior on a per-task basis:

Provide retry settings to the @task decorator with the following syntax:

python

from render_sdk.workflows import task, Options, Retry
import random

@task(
  options=Options(
    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
      factor=1.5 # Increase delay by 50% after each retry (exp. backoff)
    )
  )
)
def flip_coin() -> str:
  if random.random() < 0.5:
    raise Exception("Flipped tails! Retrying.")
  return "Flipped heads!"

This contrived example defines a task named flip_coin that raises an exception when it "flips tails", causing the run to fail and retry according to its settings.

Running subtasks

A task can straightforwardly run other tasks that are defined in the same workflow. These subtasks each run in their own instance, just like their parent.

When should I run a subtask?

Subtasks are most helpful when different parts of a larger job benefit from long-running, independent compute.

For simple jobs (such as the very basic example below), it's more efficient to perform the entirety of your logic in a single task.

The simple sum_squares task below runs two calculate_square subtasks:

python

from render_sdk.workflows import task
import asyncio

# A task that runs two subtasks
@task
async def sum_squares(a: int, b: int) -> int: # Must be async to await subtasks
  result1, result2 = await asyncio.gather(
    calculate_square(a),
    calculate_square(b)
  )
  return result1 + result2

@task
def calculate_square(a: int) -> int:
  return a * a

When running subtasks:

In most cases, the parent task should be defined as async.
- Otherwise, it can't await the results of its subtasks.
You run a subtask by calling the corresponding function (e.g., calculate_square above).
- However, this call doesn't return the function's defined return value!
- Instead, this kicks off a task run and returns a special TaskInstance object.
- As shown, you can await this object to obtain the subtask's actual return value.
Your task can call functions that are not marked as tasks. These functions run and return as normal (they do not spin up their own task instances).

A task can run another task defined in a different workflow. However:

This requires instead using the Workflows SDK or Render API, as described in Running Workflow Tasks.
This is not tracked as a task/subtask relationship when visualizing task execution in the Render Dashboard.

Parallelizing subtasks

When running subtasks, it's usually helpful to run multiple of them in parallel, such as to chunk a large workload into smaller independent pieces. Common examples include processing batches of images or analyzing different sections of a large document.

Use asyncio.gather to run multiple subtasks in parallel. In the example below, the process_photo_upload task runs a separate process_image subtask for each element in its image_urls argument:

python

from render_sdk.workflows import task
import asyncio

@task
async def process_photo_upload(image_urls: list[str]) -> dict:
  # Process all images in parallel by running a subtask for each
  results = await asyncio.gather( 
    *[process_image(url) for url in image_urls] 
  ) 

  num_successful = sum(1 for r in results if r["success"])
  num_failed = len(results) - num_successful

  return {
    "total": len(image_urls),
    "processed": num_successful,
    "failed": num_failed,
    "results": results
  }

@task
def process_image(image_url: str) -> dict:

  # Image processing logic goes here

  return {
    "url": image_url,
    "thumbnail_url": f"{image_url}_thumb.jpg",
    "success": True
  }

If you don't use asyncio.gather or a similar function, subtasks run serially.

For example:

python

@task
async def sum_squares_slower(a: int, b: int) -> int:
  # ⚠️ Not parallel!
  result1 = await calculate_square(a)
  result2 = await calculate_square(b) # Runs after first subtask completes
  return result1 + result2

@task
def calculate_square(a: int) -> int:
  return a * a

Serial execution is helpful when running a chain of subtasks that depend on each other. However, it dramatically slows execution for subtasks that are completely independent. Parallelize wherever your use case allows.