Prefect tasks help you quickly execute small, discrete units of work. Deferred Prefect tasks run in a background process using a Prefect task worker. Use deferred tasks to move work out of the foreground of your application and distribute concurrent execution across multiple processes or machines.

For example, if you have a web application, deferred tasks allow you to offload processes such as sending emails, processing images, or inserting data into a database.

​
Using deferred tasks

Prefect tasks are Python functions that can be run immediately or deferred for background execution.

Define a task by adding the @task decorator to a Python function, and use the delay method to run the task in the background.

If you schedule the task for background execution, you can run a task worker in a separate process or container to execute the task. This process is similar to a Celery worker or an arq worker.

​
Defining a task

Add the @task decorator to a Python function to define a Prefect task:

from prefect import task

@task
def my_background_task(name: str):
    # Task logic here
    print(f"Hello, {name}!")

​
Calling tasks

You can call a task to run it immediately, or you can defer the task by scheduling it for background execution with Task.delay.

You can submit tasks to a task runner such as Ray or Dask within a workflow, which in Prefect is called a flow. However, this guide focuses on deferring task execution outside of workflows. For example, by calling my_task.delay() within a web application.

However you run a task, Prefect uses your task configuration to manage and control task execution. The following example shows both methods of calling a task and using delay:

# Import the previously-defined task
from myproject.tasks import my_background_task

# Run the task immediately
my_background_task("Joaquim")

# Schedule the task for execution outside of this process
my_background_task.delay("Agrajag")

​
Executing deferred tasks with a task worker

To run tasks in a separate process or container, start a task worker.

The task worker continually receives instructions to execute deferred tasks from Prefect’s API, executes them, and reports the results back to the API.

Task workers only run deferred tasks, not tasks you call directly as normal Python functions.

Run a task worker by passing tasks into the prefect.task_worker.serve() method:

tasks.py
from prefect import task
from prefect.task_worker import serve


@task
def my_background_task(name: str):
    # Task logic here
    print(f"Hello, {name}!")


if __name__ == "__main__":
    # NOTE: The serve() function accepts multiple tasks. The Task worker 
    # will listen for scheduled task runs for all tasks passed in.
    serve(my_background_task)

The task worker begins listening for scheduled tasks. If tasks were scheduled before the task worker started, it will begin processing them.

You can also use the helper CLI command prefect task serve to start a task worker:

prefect task serve my_task.py:my_background_task

​
Guided exploration of deferred tasks and task workers in Prefect

Here are some examples of using deferred tasks and task workers in Prefect.

You will:

  • run a Prefect task in the foreground by calling it
  • start a task worker and defer tasks so that they run in the background
  • create a basic FastAPI application that defers tasks when you hit an endpoint
  • use Docker in two examples that mimic real use cases

One example uses a FastAPI server with multiple microservices and simulates a new user signup workflow. The other example uses a Flask server with Marvin to ask questions of an LLM from the CLI and get back answers.

​
Set up

​
Step 1: Activate a virtual environment

This example uses conda, but any virtual environment manager will work.

conda deactivate
conda create -n python-tasks python=3.12
conda activate python-tasks

​
Step 2: Install Python dependencies

pip install -U prefect marvin fastapi==0.107

​
Step 3: Connect to Prefect Cloud or a self-hosted Prefect server

Use either Prefect Cloud or a self-hosted Prefect server for these examples.

You must have PREFECT_API_URL set to send tasks to task workers.

If you’re using a Prefect server with a SQLite backing database (the default database), save this value to your active Prefect Profile with the following command:

prefect config set PREFECT_API_URL=http://127.0.0.1:4200/api

If using Prefect Cloud, set the PREFECT_API_URL value to the Prefect Cloud API URL and add your API key.

The examples that use docker (examples 4 and 5) use a Prefect server by default. You can switch to Prefect Cloud by changing the PREFECT_API_URL and adding a variable for your API key in the docker-compose.yaml. Or use a Prefect server backed by a PostgreSQL database by setting the PREFECT_API_DATABASE_CONNECTION_URL.

If using Prefect server instead of Prefect Cloud, start your server by running the following command:

prefect server start

​
Step 4: Clone the repository (optional)

Clone the repository to get the code files for the examples:

git clone https://github.com/PrefectHQ/prefect-background-task-examples.git

Move into the directory:

cd prefect-background-task-examples

​
Example 1: Run a Prefect task in the foreground by calling it

Add the @task decorator to any Python function to define a Prefect task.

​
Step 1: Create a file with a task-decorated function

Create a file and save the following code in it, or run the existing file in the basic-examples directory.

greeter.py
from prefect import task 

@task(log_prints=True)
def greet(name: str = "Marvin"):
    print(f"Hello, {name}!")

if __name__ == "__main__":
    greet()

​
Step 2: Run the script in the terminal

python greeter.py

You should see the task run in the terminal. This task runs in the foreground, meaning it is not deferred.

​
Optional

You can see the task run in the UI. If you’re using a self-hosted Prefect server instance, you can also see the task runs in the database.

If you want to inspect the SQLite database, use your favorite interface. DB Browser for SQLite is explained below.

Download it here, if needed. Install it and open it.

Click Connect. Then navigate to your SQLite DB file. It’s located in ~/.prefect directory by default.

Go to the task_run table to see all your task runs there. Scroll down to see your most recent task runs or filter for them.

Hit the refresh button for updates, if needed.

​
Example 2: Start a task worker and run deferred tasks in the background

To run tasks in a separate process or container, start a task worker, similar to how you would run a Celery worker or an arq worker. The task worker continually receives scheduled tasks to execute from Prefect’s API, executes them, and reports the results back to the API. Run a task worker by passing tasks into the prefect.task_worker.serve() method.

​
Step 1: Define the task and task worker in a file

task_worker.py
from prefect import task
from prefect.task_worker import serve


@task
def my_background_task(name: str):
    print(f"Hello, {name}!")


if __name__ == "__main__":
    serve(my_background_task)

​
Step 2: Start the task worker by running the script in the terminal

python task_worker.py

The task worker is waiting for runs of the my_background_task task.

​
Step 3: Create a file and save the following code in it:

task_scheduler.py
from task_worker import my_background_task


if __name__ == "__main__":
    my_background_task.delay("Agrajag")

​
Step 4: Open another terminal and run the script

python task_scheduler.py

The code returns a “future” from the delay method. You can use this object to wait for the task to complete with wait() and to retrieve its result with result(). You can also see the task run’s UUID and other information about the task run.

​
Step 5: See the task run in the UI

Use the task run UUID to see the task run in the UI. The URL will look like this:

http://127.0.0.1:4200/task-runs/task-run/my_task_run_uuid_goes_here

Substitute your UUID at the end of the URL.

​
Step 6: Use multiple task workers to run tasks in parallel

Start another instance of the task worker. In another terminal run:

python task_worker.py

​
Step 7: Send multiple tasks to the task worker

Modify the task_scheduler.py file to send multiple tasks to the task worker with different inputs:

from task_worker import my_background_task

if __name__ == "__main__":
    my_background_task.delay("Ford")
    my_background_task.delay("Prefect")
    my_background_task.delay("Slartibartfast")

Run the file to see the work get distributed across both task workers.

​
Step 8: Shut down the task workers with control + c

This guide showed you how to send tasks to multiple Prefect task workers running in the background. This allows you to observe these tasks executing in parallel and very quickly with web sockets, with no polling required.

See additional examples in the deferred tasks GitHub repository.

Was this page helpful?

Write and run tasksConfigure logging