Great questions — and welcome! Here’s how this maps to Prefect 3.x and AWS Lambda. High level - Prefect can absolutely orchestrate flows that fan out to AWS Lambda and pass results between steps. - There is no native “Lambda worker” or job template that runs each Prefect task as a Lambda. Tasks run inside your flow’s execution environment (e.g., ECS/Fargate), and from there you invoke Lambda. - The two common patterns are: 1) Synchronous Lambda calls per Prefect task (simplest). 2) Asynchronous Lambda calls where Prefect polls for completion (more scalable/robust for longer work or higher concurrency). Is there a Lambda-specific worker? - Not today. Prefect 3.x supports ECS, Kubernetes, Docker, Cloud Run, Azure Container Instances, etc., but not Lambda as a work pool type. - Recommended AWS compute for running Prefect flows is ECS/Fargate (either push work pools or ECS workers) and then invoke Lambda from tasks when needed. - Docs: Work pools, AWS ECS guide, prefect-aws integration How to invoke Lambda from a Prefect task - Synchronous (RequestResponse): the task blocks until the Lambda finishes or hits its timeout (max 15 minutes). - Asynchronous (Event): returns immediately; you then poll for completion (e.g., check S3/DynamoDB) or use a callback/destination. Example: synchronous invocation and passing results to next task

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_aws.credentials import AwsCredentials
import json

aws = AwsCredentials.load("my-aws-creds")  # create this block in the UI first

@task
def invoke_lambda(function_name: str, payload: dict, region: str = "us-east-1"):
    session = aws.get_boto3_session()
    client = session.client("lambda", region_name=region)
    resp = client.invoke(
        FunctionName=function_name,
        InvocationType="RequestResponse",  # waits for the result
        Payload=json.dumps(payload).encode(),
    )
    if resp.get("FunctionError"):
        raise RuntimeError(f"Lambda error: {resp['FunctionError']}")
    body = resp["Payload"].read()
    return json.loads(body)

@flow(task_runner=ConcurrentTaskRunner(max_workers=64))
def orchestrate(items: list[dict]):
    # fan out to Lambda, then pass results to next step
    results = [invoke_lambda.submit("my-first-lambda", item) for item in items]
    outputs = [r.result() for r in results]  # resolve futures
    # now call the next Lambda with previous results
    next_results = [invoke_lambda.submit("my-second-lambda", out) for out in outputs]
    return [nr.result() for nr in next_results]

Example: async invocation + polling S3 for results - Have your Lambda write its output to S3 with a correlation key (avoids Lambda payload size limits and lets you decouple timings). ``` from prefect import flow, task from prefect.task_runners import ConcurrentTaskRunner from prefect_aws.credentials import AwsCredentials import boto3, json, time aws = AwsCredentials.load("my-aws-creds") @task def start_lambda(function_name: str, payload: dict, region: str = "us-east-1"): # include a unique result_key in payload; Lambda writes to s3://bucket/result_key.json session = aws.get_boto3_session() client = session.client("lambda", region_name=region) client.invoke( FunctionName=function_name, InvocationType="Event", # fire-and-forget Payload=json.dumps(payload).encode(), ) return payload["result_key"] @task(retries=60, retry_delay_seconds=10) def wait_for_s3_result(bucket: str, key: str, region: str = "us-east-1"): session = aws.get_boto3_session() s3 = session.client("s3", region_name=region) obj = s3.get_object(Bucket=bucket, Key=key) return json.loads(obj["Body"].read()) @flow(task_runner=ConcurrentTaskRunner(max_workers=64))

def orchestrate(items: list[dict], bucket: str): keys = [start_lambda.submit("my-lambda", item) for item in items] results = [wait_for_s3_result.submit(bucket, k.result()) for k in keys] return [r.result() for r in results] ``` Do workers wait the full 15 minutes? - Yes, for synchronous invocation (

InvocationType="RequestResponse"

) the Prefect task waits until the Lambda returns or times out. That ties up one worker thread. You can: - Increase the task runner’s thread pool (

ConcurrentTaskRunner(max_workers=...)

). - Use async invocation + polling (shown above) to decouple long waits and improve throughput. - Add

timeout_seconds

and retries to tasks for resilience. Scalability for 10s–100s of concurrent flows - Yes — this is a common pattern. Key considerations: - Use ECS/Fargate work pools to scale out flow execution. With ECS “Push” work pools in Prefect Cloud, you don’t need to run a persistent worker process. - Control concurrency and blast radius with Prefect’s global concurrency limits and rate limits (useful to respect AWS quotas and avoid thundering herds). - Docs: Global concurrency limits, Rate limits - Prefer passing pointers (e.g., S3 keys) instead of large in-memory payloads between tasks/Lambdas. - Use S3 for result storage to make fan-out/fan-in robust and avoid payload size limits. - Docs: Results and result storage - Ensure your AWS Lambda concurrency and S3/DynamoDB quotas match your target scale. “Can Prefect deploy each task as a Lambda?” - Not out-of-the-box. Prefect executes tasks in the flow’s environment. If your goal is “serverless per task,” common alternatives are: - Keep Prefect on ECS/Fargate and have tasks invoke Lambda (recommended). - Or move the step-by-step Lambda orchestration into AWS Step Functions and have Prefect kick off Step Functions executions (via boto3) and monitor them. - For batch-style jobs that can run longer than Lambda’s 15-minute limit, consider AWS Batch; prefect-aws provides helpers for Batch. - Docs: prefect-aws index A few best practices - Set task timeouts and retries for Lambda calls. - Use correlation IDs + S3/DynamoDB to pass results between steps when using async invocations. - Use Prefect concurrency limits to cap concurrent Lambda invocations by name/region. - Consider ECS Fargate for steps that exceed Lambda’s limits or need larger dependencies. If you can share: - Are you on Prefect Cloud or self-hosted Prefect Server? - Are your Lambda steps under 15 minutes and small/medium payloads? - Is ECS/Fargate acceptable for running the flow(s)? I can tailor a concrete deployment plan (including work pool setup and example deployments) based on those answers. Helpful links - Work pools overview: Work pools - AWS ECS guide: Run Prefect on AWS ECS/Fargate - prefect-aws integration: prefect-aws docs - Results and storage: Results - Concurrency and rate limits: Global concurrency, Rate limits

Thanks for the extra context — this helps a ton. Short answer to your thread/“blocking” concern: if you run each flow on ECS/Fargate (recommended), each flow run gets its own container. A task that waits on a Lambda in one flow only ties up a thread inside that flow’s container. It won’t block other flows because they’re running in separate ECS tasks/containers. A solid starter architecture - Start with Prefect Cloud and an ECS “push” work pool. Each flow run becomes an ECS Fargate task — nice isolation and easy horizontal scale. Later, if you self-host, you can switch to an ECS worker while keeping the same flow code. - Work pools: Work pools overview - ECS guide: Run Prefect on AWS ECS/Fargate - In your flow code, invoke Lambda from tasks. For <14 minute functions you can: - Use synchronous RequestResponse (simple), or - Fire-and-forget (Event) + poll S3/DynamoDB and use task retries to avoid holding a thread between checks. Synchronous Lambda (waits, with a longer HTTP read timeout)

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_aws.credentials import AwsCredentials
from botocore.config import Config
import json

aws = AwsCredentials.load("my-aws-creds")  # create this block in the UI

@task
def invoke_lambda(function_name: str, payload: dict, region: str = "us-east-1"):
    session = aws.get_boto3_session()
    cfg = Config(read_timeout=900, retries={'max_attempts': 3})  # up to 15 min
    client = session.client("lambda", region_name=region, config=cfg)
    resp = client.invoke(
        FunctionName=function_name,
        InvocationType="RequestResponse",
        Payload=json.dumps(payload).encode(),
    )
    if resp.get("FunctionError"):
        raise RuntimeError(f"Lambda error: {resp['FunctionError']}")
    return json.loads(resp["Payload"].read())

@flow(task_runner=ConcurrentTaskRunner(max_workers=64))
def orchestrate(items: list[dict]):
    futures = [invoke_lambda.submit("my-first-lambda", it) for it in items]
    results = [f.result() for f in futures]  # pass to next step, etc.
    return results

Notes: - Boto’s default read timeout (~60s) is shorter than your Lambda duration; set

read_timeout

high enough. - If you expect larger outputs, write to S3 in Lambda and return just an S3 key. Asynchronous Lambda + polling that doesn’t hog threads - Have Lambda write results to S3 (e.g., `s3://bucket/run-&lt;id&gt;.json`). - Use task retries with delay to “reschedule” the check — your flow container is idle between attempts.

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_aws.credentials import AwsCredentials
import json, time, uuid

aws = AwsCredentials.load("my-aws-creds")

@task
def start_lambda(function_name: str, base_payload: dict, bucket: str, region: str = "us-east-1"):
    result_key = f"results/{uuid.uuid4()}.json"
    payload = {**base_payload, "result_bucket": bucket, "result_key": result_key}
    session = aws.get_boto3_session()
    client = session.client("lambda", region_name=region)
    client.invoke(
        FunctionName=function_name,
        InvocationType="Event",
        Payload=json.dumps(payload).encode(),
    )
    return result_key

@task(retries=60, retry_delay_seconds=10)
def wait_for_s3_result(bucket: str, key: str, region: str = "us-east-1"):
    session = aws.get_boto3_session()
    s3 = session.client("s3", region_name=region)
    obj = s3.get_object(Bucket=bucket, Key=key)  # raises if not yet available
    return json.loads(obj["Body"].read())

@flow(task_runner=ConcurrentTaskRunner(max_workers=64))
def orchestrate(items: list[dict], bucket: str):
    keys = [start_lambda.submit("my-lambda", item, bucket) for item in items]
    results = [wait_for_s3_result.submit(bucket, k.result()) for k in keys]
    return [r.result() for r in results]

This pattern scales well: your flow container does brief checks, sleeps via rescheduling between checks, and won’t pin a thread for 10+ minutes. Concurrency and rate control - To cap total concurrent Lambda calls across all flows: - Easiest: tag-based concurrency limits. Tag your Lambda task and set a limit via CLI. - Docs: Tag-based concurrency limits - Example:

# in code
      from prefect import task

      @task(tags=["lambda-my-first"])
      def invoke_lambda(...):
          ...

prefect concurrency-limit create lambda-my-first 50

- You can also add per-deployment or per-work-queue limits in Cloud to keep your ECS cost and API usage predictable. - If you need true rate limiting (X per second), we can layer that in too; happy to sketch it once we know your target rates. - Docs: Global concurrency limits, Rate limits About threads and “blocking” - With ECS/Fargate work pools: - Each flow run = its own ECS task/container, so one flow’s long wait does not block other flows. - Inside a single flow container, synchronous Lambda calls use threads from the task runner. For I/O-bound calls, a larger thread pool works well. - Set via

ConcurrentTaskRunner(max_workers=...)

or env var

PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS

. - If you expect many long waits, prefer the async + S3 polling pattern above — with retries/reschedules it frees up threads between checks. Cloud vs self-hosted - Cloud is the fastest path to value; you get push work pools, hosted API, and observability out of the box. - If you later self-host, you’ll run the Prefect server plus an ECS worker. Flow code and your Lambda orchestration pattern stay the same. - Work pools overview: Work pools A few final recommendations - Use S3 for passing medium/large payloads and for durable results between Lambda steps: Results and storage - Set reasonable task

timeout_seconds

and retries around Lambda calls. - Confirm AWS Lambda concurrency and S3/Dynamo quotas are aligned with your peak parallelism. If you can share: - Approx peak concurrent flow runs and Lambda invocations? - Typical payload sizes and where you’d like to store results (S3 vs inline)? - Region/VPC constraints? I can turn that into a concrete deployment plan (ECS work pool sizing, concurrency settings, and a deployment template) tailored to your scale.

generally speaking

How does this transfer to Azure/GCP infrastructure?

we support most of the same things in each big cloud provider's integration library, but aws and gcp might be slightly more fully featured than azure circumstantially feel free to make an enhancement request for anything you expect to see but don't

Great questions — here’s how this maps out in Prefect 3.x. Does the ECS worker run continuously and poll? - Yes. In the pull model, a Prefect worker is a lightweight, long-lived process that polls your Prefect API (Cloud or self-hosted) for new flow runs. - An ECS worker can run anywhere it can submit ECS tasks (an EC2 instance, a container running in ECS/Fargate, or any machine with AWS creds and network access to ECS). It polls the Prefect API, and when work is available it launches ECS tasks, waits for them to finish, then keeps polling. - Key knobs: polling frequency and heartbeat are configurable via env vars like

PREFECT_WORKER_QUERY_SECONDS

and

PREFECT_WORKER_HEARTBEAT_SECONDS

. - Docs: Workers, Work pools, ECS guide Cloud equivalents for Workers / Work Pools You can do this two ways in every cloud: pull (run a worker) or push (serverless, no worker). - AWS - Pull: ECS work pool + ecs worker (runs continuously and polls) - Push: ecs:push work pool (Prefect submits tasks to ECS; no worker required) - Also: Kubernetes worker (EKS), Docker worker, Process worker - GCP - Pull: cloud-run-v2 work pool + worker (runs continuously and polls; launches Cloud Run jobs) - Push: cloud-run:push work pool (Prefect submits Cloud Run jobs; no worker) - Also: kubernetes worker (GKE), vertex-ai worker for Vertex jobs - Azure - Pull: azure-container-instance (ACI) work pool + worker (polls; launches ACI containers) - Push: azure-container-instance:push work pool (Prefect submits ACI jobs; no worker) - Also: kubernetes worker (AKS) Serverless vs always-on - Pull pools: require an always-on worker process (think: a tiny service) that polls for new work. - Start one with:

prefect worker start --pool <your-pull-pool-name>

- Push pools: no worker to run; Prefect Cloud provisions + submits jobs to the cloud service on demand. Scale-to-zero when idle. Great for cloud-agnostic, low-ops setups. - You can also auto-provision cloud infra for push pools:

prefect work-pool provision-infra <your-push-pool-name>

- Docs: Serverless (push) guide Cloud-agnostic patterns - Keep your flow code portable and avoid cloud-specific code in flows (e.g., use

prefect-aws

,

prefect-gcp

,

prefect-azure

tasks behind configuration when needed). - Create one work pool per target cloud and deploy the same flow to each by targeting a different pool. - Use job variables to swap images, env, and resource settings per cloud (ECR vs GAR vs ACR, etc.). - Example flow deployment (Prefect 3.x pattern, avoiding deprecated 2.x commands):

from prefect import flow

@flow
def my_flow():
    ...

# Deploy to AWS ECS (pull or push pool)
my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/my_flow.py:my_flow",
    name="my-flow"
).deploy(work_pool_name="aws-ecs-pull")

# Deploy the same flow to GCP Cloud Run
my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/my_flow.py:my_flow",
    name="my-flow-gcp"
).deploy(work_pool_name="gcp-cloud-run-push")

CLI quick-start examples - Create a work pool (examples; use the type that matches your target):

prefect work-pool create "aws-ecs-pull" --type ecs
prefect work-pool create "aws-ecs-push" --type ecs:push
prefect work-pool create "gcp-cloud-run" --type cloud-run-v2
prefect work-pool create "gcp-cloud-run-push" --type cloud-run:push
prefect work-pool create "azure-aci" --type azure-container-instance
prefect work-pool create "azure-aci-push" --type azure-container-instance:push

- If you’re unsure of a type’s job template, ask Prefect for it:

prefect work-pool get-default-base-job-template --type <type>

- Start a worker for any pull pool: ```

prefect worker start --pool <pool-name> ``` When to choose which - Prefer push (serverless) if you want zero idle cost and minimal ongoing ops. - Choose pull if you need fine-grained control, private networking, custom auth flows, or to run workers inside specific environments. Good references - Workers: Workers - Work pools: Work pools - Serverless/push: Serverless deployments - ECS: AWS ECS guide - Kubernetes (works across EKS/GKE/AKS): Kubernetes guide A couple clarifying questions so I can tailor a concrete recommendation: - Do you want to avoid running a long-lived worker (i.e., prefer push/serverless), or is a small always-on worker OK? - Which GCP/Azure services are in-scope for you (Cloud Run vs GKE; ACI vs AKS)? - Any constraints like private networking-only or strict IAM/service account boundaries?

@Marvin can explain push pools where you don't have to run the worker yourself if you don't need full control of the worker and are okay storing cloud creds in prefect cloud?

Absolutely. In Prefect 3.x, a “push” work pool lets Prefect Cloud submit your flow runs directly to your cloud provider without you running a long-lived worker. High-level: - What it is: Prefect “pushes” jobs to your cloud (ECS, Cloud Run, ACI, etc.) when a flow run is ready. No polling worker; zero idle cost. - Where it works: AWS ECS (

ecs:push

), GCP Cloud Run (

cloud-run:push

), Azure Container Instances (

azure-container-instance:push

), plus Modal/Coiled. - Credentials: You store cloud credentials in Prefect Cloud as blocks (AWS/GCP/Azure). Prefect uses those only to submit/run jobs. Blocks are encrypted at rest and in transit, and you scope them with least-privilege IAM/service accounts. How it works at run time - You create a push work pool and configure its base job template (image, resources, env, etc.). - When a deployment is triggered, Prefect Cloud uses the pool’s credentials to create a job in your provider (e.g., an ECS task, a Cloud Run job, an ACI container group). - Your container runs your flow, talks back to the Prefect API for state/logs, then exits. No worker process is required. Typical setup (example: ECS push) 1) Create a push pool and (optionally) auto-provision infra:

prefect work-pool create aws-ecs-push --type ecs:push
prefect work-pool provision-infra aws-ecs-push

2) Inspect/adjust the default job template (set image, CPU/mem, env, etc.):

prefect work-pool get-default-base-job-template --type ecs:push

3) Deploy your flow to the push pool:

from prefect import flow

@flow
def my_flow():
    ...

my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/my_flow.py:my_flow",
    name="my-flow"
).deploy(work_pool_name="aws-ecs-push")

Notes: - Ensure the pool’s job variables specify a container image (or provide one during deploy). The image should be accessible to your cloud runtime (ECR/Artifact Registry/ACR). - You can repeat the same pattern with

cloud-run:push

and

azure-container-instance:push

pools to stay cloud-agnostic. Security and trade-offs - Storing creds in Prefect Cloud: Blocks are encrypted and access-controlled. Use the narrowest IAM/service-account scope possible; short-lived/assume-role patterns are supported. - Networking: Your job’s container must reach the Prefect API for state/logs. No inbound traffic from Prefect to your VPC is required. - Trade-offs vs. pull workers: - Push: zero idle infra, simpler ops, great for serverless. Fewer knobs than running your own worker. - Pull: full control over runtime and networking (e.g., private-only clusters), but you run a small always-on worker. Good references - Work pools: Concepts: Work pools - Serverless/push: How-to: Serverless (push) deployments - ECS specifics: AWS ECS guide - Workers (for comparison): Concepts: Workers If you tell me which provider you want to start with (ECS, Cloud Run, or ACI), I can share a minimal job template and exact variable settings to get your first push pool running.