Claude Code with Prefect: Reliable Workflows in Python
Why Prefect without CLAUDE.md ships pipelines that hang
Prefect is the most ergonomic Python workflow orchestrator in 2026. A flow is a decorated function, a task is a decorated function, and you can take a working Python script and lift it into a scheduled, retrying, observable pipeline by adding two decorators. The speed is the appeal. The risk is that Claude Code does not know which Prefect defaults are the right shape for production and which are convenient for the quickstart. Without explicit constraints, Claude generates flows that work in development against a small dataset, then time out, deadlock, or silently double-run when the scheduler picks them up at 3 AM.
The most common Claude defaults that break a Prefect project: writing a flow with no @task decomposition (every step runs in the flow process with no caching, retry, or observability boundary), omitting retries and retry_delay_seconds from tasks that call external APIs, importing modules at deployment-build time that are not available in the worker's image, using serve for development scripts and shipping the same pattern to production, defaulting to a single Process work pool when the workload needs Docker isolation, and storing secrets in environment variables on the worker instead of Prefect Secret blocks. On top of those, Claude makes a consistent SDK-level mistake: it writes @flow with no name= argument, then the run shows up in the UI under the function name and breaks every dashboard query that filtered by canonical flow names.
This guide covers the CLAUDE.md configuration that locks Claude Code into Prefect's correct model: tasks for retryable external calls, flows as orchestration, deployments as the versioned unit, work pools chosen for the runtime, retries and timeouts declared on every task, and secrets pulled from Prefect blocks at runtime. For the Python language defaults that pair with Prefect, Claude Code with Python covers the project layout and dependency management. If your Prefect flows write to Postgres, Claude Code with PostgreSQL shows the connection pool pattern that prevents idle-connection exhaustion under concurrent flow runs.
The Prefect CLAUDE.md template
The CLAUDE.md at your project root is read at the start of every Claude Code session. For a Prefect integration it needs to declare: the Prefect version, the project layout, the flow and task naming convention, the retry policy, the deployment pattern, the work pool choice, the secret management approach, and the hard rules that block the mistakes Claude makes most often.
# Prefect rules
## Stack
- prefect ^3.x, Python 3.11+
- prefect-docker for Docker-based work pools, or prefect-aws for ECS
- uv or poetry for dependency management (NOT pip + requirements.txt freeze)
- PREFECT_API_URL set in worker environment (NEVER hardcode)
## Project structure
- flows/ , one file per flow (snake_case names)
- tasks/ , shared task definitions (reusable across flows)
- deployments/ , prefect.yaml or programmatic Deployment definitions
- blocks/ , block creation scripts (run once, not in flow files)
- tests/ , pytest tests for tasks (flows can be smoke-tested)
## Flow naming (MANDATORY)
- @flow ALWAYS has name='snake_case_canonical_name'
- The function name is for the file, the flow name is the dashboard identifier
- NEVER rename a deployed flow without considering downstream dashboards and alerts
## Task decomposition (MANDATORY)
- Every external call (HTTP, DB, S3, third-party API) is its OWN @task
- Tasks that do pure computation on data already in memory CAN stay in the flow
- A flow with no @task calls is a smell - lift external calls to tasks
## Retry policy (MANDATORY for tasks that touch the outside world)
- @task(retries=3, retry_delay_seconds=[10, 30, 60]) for HTTP, third-party APIs
- @task(retries=2, retry_delay_seconds=5) for DB tasks (idempotent only)
- @task(retries=0) for non-idempotent tasks (send_email, create_resource) - handle retry in idempotency key
- Timeouts: @task(timeout_seconds=300) for external calls, NEVER unbounded
## Deployment pattern
- Production flows ALWAYS deploy via prefect.yaml or Python Deployment.build_from_flow
- NEVER ship .serve() patterns to production (they keep the process alive)
- Deployments declare: work pool, work queue, parameters schema, tags, schedule
- Worker images MUST contain the same package set the flow imports
## Work pool selection
- Process pool: local dev, simple Python scripts, no external runtime needed
- Docker pool: production default, isolated runtime, pinned image
- Kubernetes pool: large workloads, parallel fan-out, autoscaling
- ECS / Cloud Run pool: managed cloud, no cluster to operate
## Secrets (MANDATORY)
- ALWAYS use Prefect Secret blocks for API keys
- NEVER hardcode in flow files
- NEVER read from os.environ in flow code (use a Secret block in the flow)
- Block creation lives in blocks/ scripts, run once per environment
## Hard rules
- NEVER define @flow without name= argument
- NEVER call an external API directly in a flow body without a @task wrapper
- NEVER skip retries on HTTP calls
- NEVER deploy from a notebook or a REPL
- NEVER use Prefect 1.x patterns (Parameters, with Flow():, this is Prefect 3 only)
- NEVER ship .serve() to production - use prefect deployment apply
- ALWAYS pin the worker image and the flow code together
Four rules here prevent the majority of broken Prefect projects Claude generates without them.
The task decomposition rule is the difference between a Prefect flow and a Python script wrapped in @flow. Tasks are the unit of caching, retry, mapping, and observability. A flow that calls requests.get() directly in the function body cannot retry just the HTTP call when the third party is flaky, it has to retry the whole flow. Claude defaults to writing one big function under @flow because the decorator works that way and the tutorial examples often do exactly that. The CLAUDE.md rule forces external calls into tasks so the retry and timeout boundary is at the right level.
The retry policy rule prevents flows that fail forever on transient errors. Prefect tasks do not retry by default. A 503 from a third-party API kills the task and the flow. The rule "every task that touches the outside world has retries=3" with an exponential delay list is the right default for almost every external call. Non-idempotent tasks (send an email, create a billing charge) need a different pattern: handle retry through an idempotency key in the call itself, not via the task retry mechanism.
The deployment rule prevents the .serve() trap. The Prefect quickstart shows flow.serve(name='...') for local development because it spins up an inline scheduler in the same process as the flow code. It is excellent for trying Prefect. It is terrible for production: the process is the scheduler, the worker, and the flow runtime all at once, and a single error in any of those takes down the deployment. The rule "production flows deploy via prefect.yaml" forces the correct separation: deployments are declarative, workers run in separate processes, and the API is the coordination layer.
The secrets rule prevents accidental key exposure. Claude defaults to os.environ['API_KEY'] because that is how every other Python script reads secrets. In Prefect, that pattern means the secret has to be in the worker's environment, which means it sits in process memory, log lines, and crash dumps. Prefect Secret blocks store the secret in the Prefect API encrypted, and Secret.load('api-key').get() retrieves it inside the task. The block is named, versioned, and auditable.
Install and project setup
Install Prefect with uv (the recommended package manager in 2026):
uv add prefect
uv add prefect-docker # if using Docker work pool
Configure the Prefect API URL. For Prefect Cloud:
prefect cloud login
For a self-hosted Prefect server:
prefect config set PREFECT_API_URL=http://your-prefect-server:4200/api
Initialise the project:
prefect init
This creates prefect.yaml at the project root, which is where deployment definitions live.
Writing a flow with task decomposition
A flow that does work and survives transient failures. The shape Claude needs to learn:
# flows/sync_orders.py
from datetime import datetime
from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret
import httpx
import asyncpg
@task(
name="fetch_orders_page",
retries=3,
retry_delay_seconds=[10, 30, 60],
timeout_seconds=60,
)
async def fetch_orders_page(api_url: str, api_key: str, page: int) -> dict:
logger = get_run_logger()
logger.info(f"Fetching orders page {page}")
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(
f"{api_url}/orders",
params={"page": page, "limit": 100},
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status()
return response.json()
@task(
name="upsert_orders",
retries=2,
retry_delay_seconds=5,
timeout_seconds=120,
)
async def upsert_orders(dsn: str, orders: list[dict]) -> int:
if not orders:
return 0
conn = await asyncpg.connect(dsn)
try:
await conn.executemany(
"""
INSERT INTO orders (id, customer_id, total, status, created_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET
status = EXCLUDED.status,
total = EXCLUDED.total,
updated_at = now()
""",
[
(o["id"], o["customer_id"], o["total"], o["status"], o["created_at"])
for o in orders
],
)
finally:
await conn.close()
return len(orders)
@flow(
name="sync_orders_hourly",
log_prints=True,
timeout_seconds=1800,
)
async def sync_orders_hourly() -> int:
logger = get_run_logger()
api_key = await Secret.load("orders-api-key")
db_dsn = await Secret.load("orders-db-dsn")
total = 0
for page in range(1, 11):
data = await fetch_orders_page(
api_url="https://api.example.com/v1",
api_key=api_key.get(),
page=page,
)
if not data.get("orders"):
logger.info(f"No more orders at page {page}, stopping")
break
total += await upsert_orders(dsn=db_dsn.get(), orders=data["orders"])
logger.info(f"Synced {total} orders")
return total
if __name__ == "__main__":
import asyncio
asyncio.run(sync_orders_hourly())
Five things in this flow that Claude does not generate without CLAUDE.md rules: each task has name=, retries=, retry_delay_seconds=, and timeout_seconds=; the flow has name= (the canonical dashboard identifier); secrets are loaded via Secret blocks; the HTTP and DB calls are in their own tasks; the flow body coordinates rather than doing the work itself.
Add a concrete flow shape to CLAUDE.md:
## Flow shape (CONCRETE - copy this)
@flow(name='snake_case_name', log_prints=True, timeout_seconds=<seconds>)
async def my_flow():
# 1. Load secrets via Secret blocks (NOT os.environ)
# 2. Call tasks, not external APIs directly
# 3. Return a result (Prefect logs it)
@task(name='snake_case_name', retries=3, retry_delay_seconds=[10, 30, 60], timeout_seconds=60)
async def task_that_calls_external_api(...):
# External call here
# Raise on failure - Prefect retries based on @task config
Deployments and prefect.yaml
A deployment is the versioned, schedulable instance of a flow. The deployment binds a flow to a work pool, a schedule, parameters, and tags. The deployment is the thing that runs at 3 AM, not the flow file directly.
A prefect.yaml for the orders sync flow:
name: orders-pipeline
prefect-version: 3.0.0
build:
- prefect_docker.deployments.steps.build_docker_image:
id: build-image
requires: prefect-docker
image_name: ghcr.io/yourorg/orders-pipeline
tag: '{{ get_git_commit_sha() }}'
dockerfile: auto
push:
- prefect_docker.deployments.steps.push_docker_image:
requires: prefect-docker
image_name: ghcr.io/yourorg/orders-pipeline
tag: '{{ build-image.tag }}'
pull:
- prefect.deployments.steps.set_working_directory:
directory: /opt/prefect/orders-pipeline
deployments:
- name: sync-orders-hourly
entrypoint: flows/sync_orders.py:sync_orders_hourly
work_pool:
name: production-docker-pool
job_variables:
image: '{{ build-image.image }}'
schedules:
- cron: '0 * * * *'
timezone: UTC
tags:
- production
- orders
parameters: {}
Deploy with:
prefect deploy
Three things the YAML does right that Claude often skips. The build step pins the Docker image to the git commit SHA, so a deployment always runs the exact code version that was deployed. The pull step sets the working directory, so the flow can resolve relative imports. The tags include production, which downstream alerting filters on.
Add a deployment checklist to CLAUDE.md:
## Deployment checklist
1. prefect.yaml at project root
2. Build step pins image tag to git commit SHA
3. Each deployment names: entrypoint, work_pool, schedules, tags, parameters
4. Tags include environment ('production', 'staging')
5. Worker image contains the same Python package set the flow imports
6. Deploy: prefect deploy <deployment-name>
7. Verify: prefect deployment ls and prefect work-pool inspect <pool-name>
NEVER deploy from a notebook. NEVER ship .serve() patterns to production.
Work pools and workers
A work pool defines where flow runs execute. A worker polls the pool and picks up runs. The work pool type determines the runtime: Process, Docker, Kubernetes, ECS, Cloud Run.
Creating a Docker work pool:
prefect work-pool create production-docker-pool --type docker
Setting the default image and other job variables:
prefect work-pool update production-docker-pool \
--base-job-template /path/to/template.json
Running a worker that polls the pool:
prefect worker start --pool production-docker-pool --name worker-01
In production, run the worker as a systemd service or a container in your cluster. A worker per pool is a sensible starting point, with horizontal scaling as flow run volume grows.
Add a work pool guidance section to CLAUDE.md:
## Work pool selection guide
- Single flow, runs locally: process pool, no worker, prefect deployment serve
- Small production team, scheduled flows: docker pool, one worker as systemd service
- Multiple flows, fan-out work: kubernetes pool, autoscaling worker deployment
- Managed cloud preference: ECS or Cloud Run pool, no cluster to operate
- Long-running streaming work: kubernetes pool with separate worker deployment
NEVER mix process and docker workers on the same pool - the runtime mismatch causes random failures
Caching tasks with cache_key_fn
Prefect tasks can cache results based on a deterministic key. The cache hits avoid re-running expensive work when inputs have not changed. This is the right pattern for tasks that fetch reference data, compute expensive aggregations, or transform stable inputs.
from datetime import timedelta
from prefect import task
from prefect.tasks import task_input_hash
@task(
name="load_product_catalogue",
cache_key_fn=task_input_hash,
cache_expiration=timedelta(hours=6),
)
async def load_product_catalogue(category: str, region: str) -> list[dict]:
# expensive fetch and transform
...
The task_input_hash function hashes the task arguments to produce the cache key. Two calls with the same category and region within 6 hours return the cached result. Two calls with different arguments compute fresh.
For tasks where the inputs include sensitive values you do not want hashed into the key, supply a custom cache_key_fn:
def catalogue_cache_key(context, parameters):
return f"catalogue:{parameters['category']}:{parameters['region']}"
@task(
cache_key_fn=catalogue_cache_key,
cache_expiration=timedelta(hours=6),
)
async def load_product_catalogue(category: str, region: str, _secret_token: str) -> list[dict]:
...
The custom function excludes _secret_token from the key, so the cache hits whether the token rotates or not.
Add a caching guidance section to CLAUDE.md:
## Task caching
- Use cache_key_fn=task_input_hash for tasks with deterministic outputs
- Set cache_expiration based on data freshness needs
- For tasks with sensitive args, write a custom cache_key_fn that excludes them
- NEVER cache tasks with side effects (writes, sends, charges)
- NEVER cache tasks where the output depends on external state Prefect cannot see
Common Claude Code mistakes with Prefect
Six patterns Claude generates incorrectly without CLAUDE.md constraints.
1. No name= on @flow
Claude generates: @flow with no argument. The flow name in the dashboard defaults to the function name.
Correct pattern: @flow(name='canonical_snake_case'). The name is stable across function renames.
2. External calls in flow body
Claude generates: httpx.get(...) directly in the @flow function.
Correct pattern: the HTTP call lives in a @task so it can retry independently.
3. Missing retries
Claude generates: @task with no retries=. The first transient failure kills the task.
Correct pattern: @task(retries=3, retry_delay_seconds=[10, 30, 60]) for any external call.
4. os.environ for secrets
Claude generates: api_key = os.environ['API_KEY'] inside the flow.
Correct pattern: api_key_block = await Secret.load('api-key') then api_key_block.get().
5. .serve() shipped to production
Claude generates: if __name__ == '__main__': my_flow.serve(name='production').
Correct pattern: deploy via prefect.yaml, run a separate worker process, the flow file only contains asyncio.run(my_flow()) for local testing.
6. Prefect 1.x syntax
Claude generates: with Flow('my-flow') as flow: or Parameter('input') from Prefect 1.x.
Correct pattern: Prefect 3.x decorators only. @flow and @task, no context manager, parameters are normal function arguments.
Add this list to CLAUDE.md with the corrected form for each. Claude has seen a lot of Prefect 1.x in training data and reaches for it unless told otherwise.
Permission hooks for Prefect CLI
A Prefect project accumulates CLI invocations: deploy, work pool create, secret block apply, deployment runs. Some are local dev. Some change production schedules and config. Permission hooks gate the destructive ones.
In .claude/settings.local.json:
{
"permissions": {
"allow": [
"Bash(prefect deployment ls*)",
"Bash(prefect work-pool ls*)",
"Bash(prefect flow-run ls*)",
"Bash(prefect block ls*)",
"Bash(prefect worker start*)"
],
"deny": [
"Bash(prefect deploy*)",
"Bash(prefect deployment delete*)",
"Bash(prefect work-pool delete*)",
"Bash(prefect block delete*)"
]
}
}
Listing deployments, work pools, and flow runs is safe and read-only. Deploying, deleting deployments, deleting work pools, or removing secret blocks need explicit confirmation. The deny list forces Claude to surface those operations as prompts rather than running them as part of an automated workflow.
Building Prefect projects that run at 3 AM without alerts
The Prefect CLAUDE.md in this guide produces orchestration code where every flow has a stable canonical name, external calls live in retryable tasks, secrets come from Prefect Secret blocks, deployments are declarative via prefect.yaml, work pools are chosen for the runtime, and tasks have explicit retries and timeouts.
The underlying principle is the same as any orchestrator with Claude Code. Prefect without a CLAUDE.md produces flows that work in development and break in production: tasks that hang because they have no timeout, flows that fail forever because retries were not declared, deployments that drift because the worker image lags the flow code, and secrets that leak through os.environ. The CLAUDE.md template removes each failure mode by making the correct pattern the only pattern Claude can generate.
For deeper Python project layout patterns, Claude Code with Python covers the dependency management discipline that keeps worker images aligned with flow code. Claudify includes a Prefect-specific CLAUDE.md template with the flow naming convention, task decomposition rules, retry policy matrix, deployment checklist, work pool selection guide, Secret block pattern, and all six common-mistake rules pre-configured.
Get Claudify. Ship Prefect pipelines that finish what they start.
More like this
Ready to upgrade your Claude Code setup?
Get Claudify
