Great question. In Prefect 3.x, the most common (and reliable) patterns to trigger a flow when a new S3 object is created are: Recommended (no code): S3 → EventBridge → Prefect Cloud Webhook → Automation runs your deployment - What it does: S3 emits an Object Created event → EventBridge forwards it to a Prefect Webhook → the Webhook turns it into a Prefect event → an Event Trigger (Automation) on your deployment runs it, optionally passing the S3 bucket/key as parameters. - Why it’s nice: No Lambda to maintain, event-driven, and easy to template parameters from the incoming event. - High-level steps: 1) Create a Webhook in Prefect Cloud and choose a template (or custom) to map fields from the incoming request into a Prefect event. You can set the event name (e.g.,

aws.s3.object.created

) and shape the payload/resource values you’ll use in the Automation. - Docs: Create a webhook, Webhooks concept 2) In AWS, send S3 object created events to EventBridge (optionally filter by suffix like .csv). - AWS docs: S3 events to EventBridge 3) In EventBridge, create an API Destination that POSTs to your Prefect Webhook URL (you can use a “no auth” Connection since the webhook URL is unguessable; you can rotate it anytime). - AWS docs: API destinations 4) In Prefect, add an Automation to your deployment: - Trigger: Event matches your webhook’s event name (e.g.,

aws.s3.object.created

), optionally filter on

resource.id

or

payload

content. - Action: Run Deployment. Pass parameters via templating, e.g.

bucket: {{ event.payload.bucket }}

,

key: {{ event.payload.key }}

. - Docs: Create deployment triggers, Event triggers concept Code-light alternative: S3 → Lambda → Prefect Webhook - Same as above but with a small Lambda that POSTs the S3 event to your Prefect Webhook (useful if you want to transform the event a bit before Prefect sees it). No Prefect auth needed for the webhook. Direct API approach (more control): S3 → Lambda → Prefect REST API (create a flow run from a deployment) - What it does: Lambda parses the S3 event and directly creates a flow run for a specific deployment, passing the S3 bucket/key as parameters. - When to pick this: You want to skip webhooks/automations and directly target a deployment, or add custom validation/routing in code. - Endpoint (Cloud): - POST https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>/deployments/<deployment_id>/create_flow_run - Auth header:

Authorization: Bearer <PREFECT_API_KEY>

- Endpoint (self-hosted): - POST https://<your-prefect-server>/api/deployments/<deployment_id>/create_flow_run - API reference: Create flow run from deployment - Minimal Python Lambda example (you can swap

requests

for

urllib3

if you don’t package requests): ``` import json, os, requests PREFECT_API_KEY = os.environ["PREFECT_API_KEY"] PREFECT_BASE = os.environ["PREFECT_BASE_URL"] # e.g. "https://api.prefect.cloud/api/accounts/acc_id/workspaces/ws_id" DEPLOYMENT_ID = os.environ["PREFECT_DEPLOYMENT_ID"] def handler(event, context): # S3 event format: event["Records"][0]["s3"]["bucket"]["name"], ["s3"]["object"]["key"] record = event["Records"][0] bucket = record["s3"]["bucket"]["name"] key = record["s3"]["object"]["key"] url = f"{PREFECT_BASE}/deployments/{DEPLOYMENT_ID}/create_flow_run" headers = { "Authorization": f"Bearer {PREFECT_API_KEY}", "Content-Type": "application/json", }

payload = { "parameters": { "bucket": bucket, "key": key, }, # optionally: # "name": f"s3-trigger-{bucket}/{key}", # "tags": ["s3-trigger"], } r = requests.post(url, headers=headers, json=payload, timeout=10) r.raise_for_status() return {"statusCode": 200, "body": json.dumps({"flow_run": r.json()})} ``` Polling fallback (simplest but least event-driven) - Schedule a short, frequent deployment that checks S3 for new files and exits quickly if there’s nothing to process. This is the easiest to ship but can be wasteful (lots of idle runs) and adds latency. Prefer the event-driven patterns above. Which should you choose? - Prefer Webhook + Automation if you want a no-/low-code, secure, and easily observable event-driven setup entirely in Prefect Cloud. - Use the Lambda → Prefect API method if you need custom logic, fan-out across multiple deployments, or conditional routing based on S3 metadata. If you share: - Are you on Prefect Cloud or self-hosting? - Do you want to avoid Lambda (pure EventBridge) or are you comfortable using Lambda? - What parameters does your flow expect? I can tailor a ready-to-copy Webhook template (mapping S3 bucket/key to event payload) and the exact Automation configuration for your deployment. Helpful docs: - Webhooks: Create a webhook, Concepts - Event triggers/automations: Creating deployment triggers, Event triggers - REST: Create flow run from deployment - Blog (event-driven pipelines on AWS): Orchestrating event-driven serverless data pipelines

Nice find — that module is exactly for wiring S3 → EventBridge → Prefect Cloud Webhook. To “extend it for the automation piece,” you’ll add a Prefect Automation on your deployment that listens for the webhook’s event and runs the deployment, passing the S3 bucket/key as parameters. Here’s the end-to-end plan (no Lambda required): 1) Create a Prefect Cloud Webhook - In Prefect Cloud, create a Webhook. Set: - Event name: something like

aws.s3.object.created

- Map fields from the incoming AWS event into the event payload (you’ll use these in your deployment’s parameters). - When S3 sends to EventBridge, the payload shape you’ll receive is the EventBridge event. Typical paths: -

request.body.detail.bucket.name

→ bucket -

request.body.detail.object.key

→ key - In the webhook builder UI, map those to payload fields, e.g.: - payload.bucket = request.body.detail.bucket.name - payload.key = request.body.detail.object.key - Copy the unique Webhook URL; you’ll pass this to the Terraform module. - Docs: Create a webhook, Webhooks concept 2) Use the terraform-prefect-bucket-sensor S3 module - Feed your Prefect Webhook URL and S3 bucket into the module; it will create the EventBridge rule, connection, and API destination to POST events to your webhook. - You can typically add filters (prefix/suffix) to limit which objects trigger the event. - Example usage (variable names may differ slightly per module version; adjust to the module’s README):

module "prefect_bucket_sensor" {
  source = "<http://github.com/PrefectHQ/terraform-prefect-bucket-sensor//modules/s3?ref=vX.Y.Z|github.com/PrefectHQ/terraform-prefect-bucket-sensor//modules/s3?ref=vX.Y.Z>"

  name        = "my-s3-bucket-sensor"
  region      = var.aws_region
  bucket_arn  = aws_s3_bucket.data.arn     # or bucket_name if the module takes that
  webhook_url = var.prefect_webhook_url    # from step 1

  # Optional filters if supported by the module:
  # object_key_prefixes = ["incoming/"]
  # object_key_suffixes = [".csv"]
}

- After apply, S3 Object Created events matching your filters will be POSTed directly to your Prefect Webhook via EventBridge API Destinations (no Lambda to maintain). 3) Add an Automation on your Prefect deployment - In Prefect Cloud, open your deployment → Automations → Add Automation: - Trigger: Event - Event name = the exact name you set on the webhook (e.g.,

aws.s3.object.created

) - Optional filters: match on payload values if you want (e.g., only run for a specific bucket or prefix) - Action: Run Deployment - Parameters: map from the event payload you defined in the webhook, e.g.: -

bucket: {{ event.payload.bucket }}

-

key: {{ event.payload.key }}

- Docs: Creating deployment triggers, Event triggers concept Flow code tip - Make sure your flow accepts parameters that match what you’ll pass:

from prefect import flow

@flow
def process_s3_object(bucket: str, key: str):
    # your logic here
    ...

Sanity checks and gotchas - Ensure S3 is sending events to EventBridge in your account/region (S3 → Properties → Event notifications via EventBridge). - In EventBridge, the API Destination connection can be “none” because the Prefect Webhook URL is already a secret, and you can rotate it anytime in Prefect Cloud. - Use suffix/prefix filters either in the module (if supported) or in the Prefect Automation trigger filters to avoid unnecessary runs. Do you want to manage the Automation/Webhook in Terraform too? If so, I can share a Terraform example using the Prefect Cloud provider, or a small script using the Prefect Cloud API. If you share: - your module version (ref) and - the parameters your flow expects (names/types),

I can tailor the exact webhook template and the Automation parameter mappings.

@Marvin can i do that in terraform

Short answer: Partly. That module will wire S3 → EventBridge → your Prefect Cloud Webhook. The “automation piece” (an Event Trigger that runs your deployment when the webhook fires) currently isn’t exposed as a first‑party Terraform resource in Prefect Cloud, so you have two practical options: Option A — Minimal IaC, use the UI for Prefect pieces - Create the Webhook in Prefect Cloud UI and copy its URL. - Docs: Create a webhook, Webhooks concepts - Pass that URL into the

terraform-prefect-bucket-sensor

S3 module you found so EventBridge sends to your webhook. - On your deployment, add an Automation (Event Trigger) that listens for your webhook’s event and runs the deployment with parameters like: -

bucket: {{ event.payload.bucket }}

-

key: {{ event.payload.key }}

- Docs: Creating deployment triggers, Event triggers Option B — Keep it IaC via Terraform “workarounds” - Webhook: Create once in the UI (there’s not an official Terraform resource for Prefect webhooks today), then keep its URL in Terraform (var/secret) for the S3 module. - Automation: Use Terraform to call Prefect’s REST API to create/update an Automation that runs your deployment on that webhook event. You can do this with a

null_resource

+

local-exec

(curl) or the HTTP provider. Example sketch:

variable "prefect_api_key" {}
variable "prefect_account_id" {}
variable "prefect_workspace_id" {}
variable "prefect_deployment_id" {}
variable "event_name" { default = "aws.s3.object.created" }

locals {
  prefect_base = "<https://api.prefect.cloud/api/accounts/${var.prefect_account_id}/workspaces/${var.prefect_workspace_id}>"
}

resource "null_resource" "create_automation" {
  # change to force update, e.g. when deployment or mappings change
  triggers = {
    deployment_id = var.prefect_deployment_id
    event_name    = var.event_name
  }

  provisioner "local-exec" {
    command = <<-EOT
      curl -sS -X POST \
        -H "Authorization: Bearer ${var.prefect_api_key}" \
        -H "Content-Type: application/json" \
        "${local.prefect_base}/automations" \
        -d '{
          "name": "Run deployment on S3 file drop",
          "enabled": true,
          "trigger": {
            "type": "event",
            "match": { "any": [ { "event": "${var.event_name}" } ] }
          },
          "actions": [
            {
              "type": "run-deployment",
              "deployment_id": "${var.prefect_deployment_id}",
              "parameters": {
                "bucket": "{{ event.payload.bucket }}",
                "key": "{{ event.payload.key }}"
              }
            }
          ]
        }'
    EOT
  }
}

Notes on Option B: - The exact Automations API shape can change; if you want, I can help you test and pin the correct payload for your account/workspace. - Store your API key in a TF variable/secret backend, not in code. - You still need to create the Webhook in the UI (and map

request.body.detail.bucket.name

and

request.body.detail.object.key

into

event.payload.bucket

and

event.payload.key

) and feed its URL into the S3 module. Alternative if you want 100% Terraform without Prefect Webhooks/Automations - Use Terraform to create a Lambda that receives S3 events and calls Prefect’s REST API to create a flow run on your deployment. - Endpoint (Cloud):

POST <https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>/deployments/<deployment_id>/create_flow_run>

- Docs: Create flow run from deployment - This keeps everything under Terraform (AWS infra + Lambda code + Prefect call), at the cost of maintaining a small Lambda.

If you share your preferences: - Do you want to manage the Prefect Automation via Terraform (HTTP/curl) or is UI acceptable? - Do you prefer webhook + automation or the Lambda-direct-to-API approach? I can provide a ready-to-apply Terraform snippet tailored to your account/workspace/deployment IDs and your flow’s parameter names.