This Prefect tutorial for beginners covers installation, writing your first task and flow, retries, scheduling, and deploying with work pools.

If you already know what Prefect is and why it appeals to Python-first teams, this tutorial is where you actually build something with it. We’re going to install Prefect, write a real flow with multiple tasks, watch it run in the UI, add retries, and finish by scheduling it to run automatically. By the end, you’ll have enough hands-on experience to start applying Prefect to your own pipelines.

If you haven’t read the conceptual overview yet, our What is Prefect guide covers tasks, flows, deployments, and work pools in more depth than we’ll have room for here. This tutorial assumes you’re roughly familiar with those terms and focuses on actually writing code.

What You’ll Need

Like most hands-on guides, this Prefect tutorial keeps setup light. You’ll need:

• Python 3.9 or later
• Basic comfort with the command line
• About 30 minutes

No prior Prefect experience is required, and everything in this tutorial runs locally, no cloud account needed.

Step 1: Installing Prefect

Start with a clean project folder and virtual environment:

mkdir prefect-tutorial
cd prefect-tutorial
python3 -m venv venv
source venv/bin/activate
pip install prefect httpx

We’re also installing httpx here since our example pipeline is going to fetch data from a public API.

Step 2: Writing Your First Task

We’re going to build a small pipeline that pulls basic statistics for a GitHub repository, filters out anything with no stars, and prints a summary. Create a file called tasks.py:

from prefect import task
import httpx

@task(retries=3, retry_delay_seconds=5, log_prints=True)
def fetch_repo_stats(repo: str) -> dict:
    url = f"https://api.github.com/repos/{repo}"
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    data = response.json()
    print(f"Fetched stats for {repo}")
    return {
        "name": data["full_name"],
        "stars": data["stargazers_count"],
        "forks": data["forks_count"],
        "open_issues": data["open_issues_count"],
    }

@task
def filter_low_activity(repo_stats: dict, min_stars: int = 0) -> dict | None:
    if repo_stats["stars"] < min_stars: return None return repo_stats @task def format_summary(repo_stats: dict) -> str:
    return (
        f"{repo_stats['name']}: {repo_stats['stars']} stars, "
        f"{repo_stats['forks']} forks, {repo_stats['open_issues']} open issues"
    )

A few things worth noticing already, since this is a lot of orchestration behavior packed into a small amount of code. The retries=3, retry_delay_seconds=5 arguments on fetch_repo_stats mean that if the GitHub API call fails or times out, Prefect will automatically retry it up to three times, waiting five seconds between attempts, without you writing a single line of retry logic. The log_prints=True argument tells Prefect to capture any print() statements inside the task and route them into its own logging system, so they show up in the UI alongside everything else.

Step 3: Writing Your First Flow

Now create flow.py, which ties these tasks together:

from prefect import flow
from tasks import fetch_repo_stats, filter_low_activity, format_summary

@flow(name="github-repo-summary", log_prints=True)
def repo_summary_flow(repo: str = "PrefectHQ/prefect", min_stars: int = 0):
    stats = fetch_repo_stats(repo)
    filtered = filter_low_activity(stats, min_stars)

    if filtered is None:
        print(f"{repo} did not meet the minimum star threshold, skipping.")
        return None

    summary = format_summary(filtered)
    print(summary)
    return summary

if __name__ == "__main__":
    repo_summary_flow()

This is worth pausing on, since it’s the clearest illustration of what makes Prefect’s model distinctive. Notice the if filtered is None: check, that’s ordinary Python control flow, deciding at runtime whether to continue the pipeline or stop early. Many orchestration tools would require you to express this kind of conditional branching through a special API or configuration syntax. Here, it’s just an if statement, because a flow is just a Python function.

Run it directly:

python flow.py

You should see log output showing the task executing, the print statements captured by Prefect’s logging, and the final summary printed to your terminal.

Step 4: Viewing the Run in Prefect’s UI

Everything so far ran locally without any UI involved, but Prefect was tracking it the whole time. Start the local Prefect server:

prefect server start

This starts Prefect’s API and UI, available at http://localhost:4200. Leave this running in its own terminal window, then open a new terminal, reactivate your virtual environment, and run the flow again:

python flow.py

Now open the UI in your browser and look at the “Flow Runs” section. You’ll see github-repo-summary listed, and clicking into it shows you the full execution timeline: each task, its state (completed, in the case of a successful run), the logs captured from log_prints, and exactly how long each step took. This is the same visibility you’d want when debugging a failed run days or weeks later, and you’re getting it automatically, without configuring any separate logging or monitoring tool.

Step 5: Simulating and Watching a Retry

To see the retry behavior actually happen, let’s temporarily break the request on purpose. Edit tasks.py and change the URL to something that will fail:

url = f"https://api.github.com/repos/{repo}/this-will-fail"

Run the flow again:

python flow.py

Watch your terminal output, or the UI, and you’ll see Prefect attempt the task, fail, wait five seconds, and retry, up to three times total, before finally marking the task (and the flow) as failed. This is exactly the behavior you’d otherwise write by hand with a try/except loop and a time.sleep() call, handled here entirely through the retries and retry_delay_seconds arguments you set earlier.

Revert your change back to the working URL before continuing.

Step 6: Adding Parameters

Our flow already accepts repo and min_stars as parameters, which means you can run it against a different repository without touching the code:

if __name__ == "__main__":
    repo_summary_flow(repo="python/cpython", min_stars=1000)

Try running it with a few different repository names and star thresholds. This is the same idea as passing arguments to any Python function, but Prefect tracks each set of parameters as part of that specific flow run’s record, so you can look back later and see exactly what inputs produced a given result.

Step 7: Creating a Deployment

This is the step most Prefect tutorial content skips over too quickly, and it’s arguably the most important one for real use. Running python flow.py manually works for development, but real pipelines need to run on a schedule without a human triggering them. This is where deployments come in. Add this to the bottom of flow.py:

if __name__ == "__main__":
    repo_summary_flow.serve(
        name="daily-repo-summary",
        cron="0 9 * * *",
        parameters={"repo": "PrefectHQ/prefect", "min_stars": 0},
    )

Run this script:

python flow.py

Instead of running once and exiting, this now starts a long-running process that stays alive, waiting for its schedule (0 9 * * *, every day at 9 a.m.) to trigger a run automatically. Check the UI’s “Deployments” tab, and you’ll see daily-repo-summary listed, along with its next scheduled run time.

You can also trigger a deployment manually, without waiting for its schedule, either from the UI by clicking “Run” on the deployment, or from the command line:

prefect deployment run 'github-repo-summary/daily-repo-summary'

Step 8: Understanding Work Pools (Briefly)

For this tutorial, .serve() handles execution directly within the same process, which is perfectly fine for learning and small-scale use. As pipelines grow, teams typically move to work pools, which let flow runs execute in separate, scalable infrastructure (Docker containers, Kubernetes pods, or cloud compute) rather than a single long-running local process.

Setting up a work pool looks like this:

prefect work-pool create my-docker-pool --type docker
prefect worker start --pool my-docker-pool

A worker polls the pool for scheduled runs and executes them in the configured infrastructure. For a first project, .serve() is the simpler starting point, and moving to a work pool later doesn’t require rewriting your flow or task code, only how it’s deployed and executed.

Step 8b: A Full Docker Work Pool Walkthrough

If you want to see the full path from local development to containerized deployment, here’s what that actually looks like in practice. This section is optional for a first read-through, but worth returning to once you’re ready to move a pipeline beyond a single machine.

First, create a Dockerfile alongside your project:

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

And a matching requirements.txt:

prefect
httpx

With Docker running locally, create the work pool and confirm it’s registered:

prefect work-pool create my-docker-pool --type docker
prefect work-pool ls

Next, define a deployment that targets this pool instead of running through .serve(). This is typically done with a prefect.yaml file, which Prefect can generate for you:

prefect deploy

Running this command walks you through an interactive setup: which flow to deploy, which work pool to target, and what schedule to attach. Behind the scenes, it builds a Docker image containing your flow code and pushes it wherever your work pool is configured to pull from (a local Docker daemon, or a container registry for remote infrastructure).

Once deployed, start a worker to actually execute runs assigned to this pool:

prefect worker start --pool my-docker-pool

The worker stays running, polling for scheduled or manually triggered runs, and spins up a fresh container for each execution. This is the same underlying model whether you’re running a single Docker worker on your laptop or a fleet of workers on Kubernetes in production, only the infrastructure target changes, not your flow code.

The practical upshot: everything you wrote earlier in this tutorial, the tasks, the flow, the retries, ports directly into this containerized setup without modification. That portability between local development and production infrastructure is one of the more genuinely useful aspects of Prefect’s deployment model.

Step 9: Using Blocks for Credentials and Configuration

Real pipelines almost always need credentials, an API key, a database password, cloud storage access, and hardcoding those directly in your flow code is a bad habit worth avoiding from the start. Prefect’s blocks system exists specifically for this.

Here’s how you’d store and retrieve a secret using a block:

from prefect.blocks.system import Secret

# Run once, typically outside your flow code, to save the secret
Secret(value="my-actual-api-key").save(name="github-api-key", overwrite=True)

Then, inside your flow or task code, retrieve it at runtime:

from prefect.blocks.system import Secret

@task
def fetch_repo_stats_authenticated(repo: str) -> dict:
    api_key = Secret.load("github-api-key").get()
    headers = {"Authorization": f"token {api_key}"}
    url = f"https://api.github.com/repos/{repo}"
    response = httpx.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()

The credential itself is stored securely through Prefect’s backend (local server or Prefect Cloud, depending on your setup) rather than sitting in plain text in your codebase or a .env file that might accidentally get committed to version control. Blocks aren’t limited to secrets either, database connection details, cloud storage configuration, and Slack webhook URLs are all commonly stored the same way, letting you reference them by name across multiple flows without duplicating configuration.

Step 10: Setting Up Automations for Failure Notifications

Watching the UI manually to see if a scheduled pipeline failed doesn’t scale past a handful of flows. Prefect’s automations let you react to events, like a flow run failing, automatically.

From the UI, under the “Automations” section, you can create a rule with a trigger (“when a flow run enters a Failed state”) and an action (“send a Slack notification” or “send an email”). This can also be configured through code:

from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import SendNotification

failure_automation = Automation(
    name="notify-on-repo-summary-failure",
    trigger=EventTrigger(
        expect=["prefect.flow-run.Failed"],
        match_related={"prefect.resource.name": "github-repo-summary"},
    ),
    actions=[SendNotification(subject="Pipeline failed", body="github-repo-summary failed. Check the UI for details.")],
)

failure_automation.create()

Once this is in place, you’ll find out about a broken pipeline the moment it breaks, rather than discovering it hours later because someone noticed a report was missing. This is a small amount of setup that pays for itself the first time a real production failure happens overnight.

Step 11: Adding a Subflow

As pipelines grow, it’s useful to break them into smaller, reusable flows. Here’s a quick example extending our project with a second flow that summarizes multiple repositories by calling our existing flow as a subflow:

@flow(name="multi-repo-summary")
def multi_repo_summary_flow(repos: list[str]):
    results = []
    for repo in repos:
        result = repo_summary_flow(repo)
        if result:
            results.append(result)
    return results

if __name__ == "__main__":
    multi_repo_summary_flow(["PrefectHQ/prefect", "python/cpython", "pandas-dev/pandas"])

Each call to repo_summary_flow inside this loop shows up as its own nested flow run in the UI, linked to the parent multi_repo_summary_flow run that triggered it. This nested visibility is one of the more useful patterns for building larger pipelines out of smaller, independently testable pieces, the same instinct you’d already have for structuring any reasonably sized Python codebase.

Step 12: Testing Your Flows and Tasks

Because tasks and flows are ultimately just decorated Python functions, testing them follows patterns most Python developers already know, without needing a running Prefect server at all for basic unit tests.

Testing a task directly, calling it like the plain function it is underneath the decorator:

def test_filter_low_activity_excludes_low_star_repos():
    stats = {"name": "test/repo", "stars": 5, "forks": 1, "open_issues": 0}
    result = filter_low_activity.fn(stats, min_stars=10)
    assert result is None

Notice the .fn attribute, this accesses the original, undecorated function directly, letting you test the underlying logic without going through Prefect’s orchestration layer at all, which keeps unit tests fast and independent of any running infrastructure.

For testing a full flow’s behavior, including how tasks interact, Prefect provides a test harness that runs everything synchronously and in an isolated context:

from prefect.testing.utilities import prefect_test_harness

def test_repo_summary_flow_runs_successfully():
    with prefect_test_harness():
        result = repo_summary_flow(repo="PrefectHQ/prefect", min_stars=0)
        assert result is not None
        assert "stars" in result

This pattern is particularly useful in CI pipelines, where you want confidence that a flow’s logic works correctly before it’s deployed, without needing a live Prefect server or real external API calls in cases where you’ve mocked out the network-dependent tasks.

Common Beginner Mistakes

A handful of mistakes come up repeatedly for people new to Prefect.

Forgetting log_prints=True. Without it, print() statements inside tasks and flows won’t show up in Prefect’s captured logs, which can make debugging confusing until you notice the missing flag.

Not setting timeouts or retries on tasks that call external services. Any task making a network call should generally have a reasonable retry policy, since transient failures (a flaky API, a brief network blip) are common and exactly what Prefect’s retry mechanism exists to handle gracefully.

Confusing running a script directly with creating a deployment. Running python flow.py executes the flow once, immediately. Only .serve() (or a work pool deployment) actually registers a schedule that triggers runs automatically going forward.

Mismatched parameter names between a deployment and the flow function. If you pass a parameter name in .serve() that doesn’t match the flow function’s actual argument names, you’ll get an error at trigger time rather than at definition time, so double-check these carefully.

Leaving the local Prefect server assumption unclear. If you’re using Prefect Cloud instead of a local server, make sure you’ve authenticated with prefect cloud login first, or your flow runs won’t show up where you expect.

Best Practices Once You’re Past the Basics

• Keep tasks small and focused, similar to good general Python function design, since smaller tasks are easier to retry safely and test independently.
• Use blocks for credentials and configuration rather than hardcoding API keys or connection strings directly in your flow code.
• Write real unit tests for your tasks, calling them directly as plain functions in your test suite, since that’s exactly what they are underneath the decorator.
• Use automations for failure notifications rather than manually checking the UI, so you find out about a broken pipeline from a Slack message, not from someone asking why yesterday’s report is missing.
• Start with .serve() and move to work pools only when you actually need the scaling they provide; don’t add infrastructure complexity before your pipeline’s actual load requires it.

Frequently Asked Questions

Do I need Docker to follow this tutorial? No. Everything here runs as a local Python process, no containers required. Docker becomes relevant later if you move to work-pool-based execution for production scaling.

What’s the difference between running python flow.py and using .serve()? Running the script directly executes the flow once and exits. .serve() starts a long-running process that stays alive, listening for its schedule to trigger automatic runs, without you manually invoking it each time.

Can I use this same tutorial with Prefect Cloud instead of a local server? Yes. Run prefect cloud login to authenticate, and your flow runs will report to Prefect Cloud’s hosted UI instead of a local server, with no other changes needed to your task or flow code.

Why didn’t my scheduled deployment run on time? The most common cause is the process running .serve() not staying alive, if that terminal window closes or the process stops, scheduled runs won’t trigger. For production use, this is exactly why teams move to work pools with dedicated, persistent workers instead.

Is this tutorial’s code production-ready? It’s close in structure, but a real production pipeline would typically add more robust error handling, secrets management through blocks, and a work-pool-based deployment rather than .serve() running in a single terminal. This tutorial is meant to build your foundational understanding first.

How do I store secrets safely instead of hardcoding them? Use Prefect’s blocks system, covered in Step 9 above. Storing an API key or database password as a Secret block keeps it out of your codebase entirely, retrieved at runtime rather than committed to version control where it could accidentally be exposed.

What happens if two scheduled runs try to start at the same time? Prefect handles concurrent flow runs independently by default, each gets its own execution and its own entry in the UI. If you need to limit how many instances of a flow run simultaneously, task and flow-level concurrency limits can be configured to prevent resource contention, useful when a pipeline shouldn’t run twice at once against the same data.

Can I trigger a flow from an external event instead of a schedule? Yes. Beyond cron-based scheduling, Prefect supports event-driven triggers through its automations system, letting a flow run start in response to a webhook, another flow’s completion, or a custom event your own code emits into Prefect’s event system.

Do I need to learn Docker before I can deploy anything with Prefect? No. .serve(), covered in Step 7, deploys and schedules a flow without any containerization at all. Docker becomes relevant specifically when you want the scaling and isolation benefits of work-pool-based execution, which is a later step, not a prerequisite for getting started.

What You Built: A Quick Recap

Before wrapping up, it’s worth pausing on just how much ground this tutorial actually covered, since it’s easy to lose track moving step by step.

You started with a single task making an authenticated-free API call, then added automatic retries with a single decorator argument. You wrapped that task in a flow using ordinary Python control flow to decide whether to continue or stop early, no special orchestration syntax required. You watched that execution play out visually in Prefect’s UI, including a deliberately broken run so you could see retry behavior happen in real time rather than just reading about it.

From there, you moved from manual execution to a scheduled deployment, first with the simple .serve() approach, then with a full Docker-based work pool walkthrough showing how the exact same flow code ports into containerized, production-style infrastructure without modification. You added credential management through blocks, failure notifications through automations, and composed a larger pipeline out of a subflow. Finally, you wrote both a fast unit test against a task’s raw logic and a full flow-level test using Prefect’s test harness.

That’s the complete arc from “here’s a Python function” to “here’s a monitored, retried, scheduled, tested, and deployable pipeline”, and notably, at no point did any of it require learning a new definition language or rewriting how you’d normally structure Python code.

Wrapping Up

That’s the full arc of this Prefect tutorial for beginners. You’ve now built a real Prefect pipeline from scratch: tasks with automatic retries, a flow with runtime conditional logic, a subflow composing smaller pieces together, and a scheduled deployment triggering runs automatically. You’ve also seen the UI’s observability firsthand, including watching a retry happen in real time.

The core idea worth carrying forward is how little of this required learning something entirely new. Every piece, @task, @flow, retries, parameters, subflows, mapped onto Python concepts you likely already understood. That’s the whole premise behind Prefect’s design, and it’s why teams already comfortable in Python often find it the fastest orchestration tool to actually become productive with.

For the conceptual deep dive behind everything covered here, revisit our What is Prefect guide, and if you’re evaluating orchestration tools more broadly, our guides on What is Temporal, What is Kestra, and our Dagster tutorial for beginners cover the rest of the landscape.

For deeper reference as you keep building, the official Prefect documentation covers work pools, blocks, and Prefect Cloud configuration in far more depth than a single tutorial can.