Skip to main content

​
Creating deployment triggers

To enable the simple configuration of event-driven deployments, Prefect provides deployment triggers—a shorthand for creating automations that are linked to specific deployments to run them based on the presence or absence of events.
Deployment triggers are a special case of automations where the configured action is always running a deployment.
Trigger definitions for deployments are supported in prefect.yaml, .serve, and .deploy. At deployment time, specified trigger definitions create linked automations triggered by events matching your chosen grammar. Each trigger definition may include a jinja template to render the triggering event as the parameters of your deployment’s flow run.

​
Define triggers in prefect.yaml

You can include a list of triggers on any deployment in a prefect.yaml file: 
deployments:
  - name: my-deployment
    entrypoint: path/to/flow.py:decorated_fn
    work_pool:
      name: my-work-pool
    triggers:
      - type: event
        enabled: true
        match:
          prefect.resource.id: my.external.resource
        expect:
          - external.resource.pinged
        parameters:
          param_1: "{{ event }}"

​
Scheduling delayed deployment runs

You can configure deployment triggers to run after a specified delay using the schedule_after field. This is useful for implementing opt-out workflows or providing review windows before automated actions: 
deployments:
  - name: campaign-execution
    entrypoint: path/to/campaign.py:execute_campaign
    work_pool:
      name: my-work-pool
    triggers:
      - type: event
        enabled: true
        match:
          prefect.resource.id: marketing.campaign
        expect:
          - campaign.proposal.created
        schedule_after: "PT2H"  # Wait 2 hours before execution
        parameters:
          campaign_id: "{{ event.resource.id }}"
The schedule_after field accepts:
  • ISO 8601 durations (e.g., “PT2H” for 2 hours, “PT30M” for 30 minutes)
  • Seconds as integer (e.g., 7200 for 2 hours)
  • Python timedelta objects in code
This deployment creates a flow run when an external.resource.pinged event and an external.resource.replied event have been seen from my.external.resource: 
deployments:
  - name: my-deployment
    entrypoint: path/to/flow.py:decorated_fn
    work_pool:
      name: my-work-pool
    triggers:
      - type: compound
        require: all
        parameters:
          param_1: "{{ event }}"
        schedule_after: "PT1H"  # Wait 1 hour before running
        triggers:
          - type: event
            match:
              prefect.resource.id: my.external.resource
            expect:
              - external.resource.pinged
          - type: event
            match:
              prefect.resource.id: my.external.resource
            expect:
              - external.resource.replied

​
Define deployment triggers using Terraform

You can also set up a deployment trigger via Terraform resources, specifically via the prefect_automation resource. 
resource "prefect_deployment" "my_deployment" {
  name            = "my-deployment"
  work_pool_name  = "my-work-pool"
  work_queue_name = "default"
  entrypoint      = "path/to/flow.py:decorated_fn"
}

resource "prefect_automation" "event_trigger" {
  name = "my-automation"
  trigger = {
    event = {
      posture   = "Reactive"
      expect    = ["external.resource.pinged"]
      threshold = 1
      within    = 0
    }
  }
  actions = [
    {
      type          = "run-deployment"
      source        = "selected"
      deployment_id = prefect_deployment.my_deployment.id
      parameters = jsonencode({
        "param_1" : "{{ event }}"
      })
    },
  ]
}

​
Define triggers in .serve and .deploy

To create deployments with triggers in Python, the trigger types DeploymentEventTrigger, DeploymentMetricTrigger, DeploymentCompoundTrigger, and DeploymentSequenceTrigger can be imported from prefect.events: 
from datetime import timedelta
from prefect import flow
from prefect.events import DeploymentEventTrigger


@flow(log_prints=True)
def decorated_fn(param_1: str):
    print(param_1)


if __name__=="__main__":
    decorated_fn.serve(
        name="my-deployment",
        triggers=[
            DeploymentEventTrigger(
                enabled=True,
                match={"prefect.resource.id": "my.external.resource"},
                expect=["external.resource.pinged"],
                schedule_after=timedelta(hours=1),  # Wait 1 hour before running
                parameters={
                    "param_1": "{{ event }}",
                },
            )
        ],
    )
As with prior examples, you must supply composite triggers with a list of underlying triggers: 
from datetime import timedelta
from prefect import flow
from prefect.events import DeploymentCompoundTrigger


@flow(log_prints=True)
def decorated_fn(param_1: str):
    print(param_1)


if __name__=="__main__":
    decorated_fn.deploy(
        name="my-deployment",
        image="my-image-registry/my-image:my-tag",
        triggers=[
            DeploymentCompoundTrigger(
                enabled=True,
                name="my-compound-trigger",
                require="all",
                schedule_after=timedelta(minutes=30),  # Wait 30 minutes
                triggers=[
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.pinged"],
                    },
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.replied"],
                    },
                ],
                parameters={
                    "param_1": "{{ event }}",
                },
            )
        ],
        work_pool_name="my-work-pool",
    )

​
Pass triggers to prefect deploy

You can pass one or more --trigger arguments to prefect deploy as either a JSON string or a path to a .yaml or .json file. 
# Pass a trigger as a JSON string
prefect deploy -n test-deployment \
  --trigger '{
    "enabled": true,
    "match": {
      "prefect.resource.id": "prefect.flow-run.*"
    },
    "expect": ["prefect.flow-run.Completed"]
  }'

# Pass a trigger using a JSON/YAML file
prefect deploy -n test-deployment --trigger triggers.yaml
prefect deploy -n test-deployment --trigger my_stuff/triggers.json
For example, a triggers.yaml file could have many triggers defined: 
triggers:
  - enabled: true
    match:
      prefect.resource.id: my.external.resource
    expect:
      - external.resource.pinged
    parameters:
      param_1: "{{ event }}"
  - enabled: true
    match:
      prefect.resource.id: my.other.external.resource
    expect:
      - some.other.event
    parameters:
      param_1: "{{ event }}"
Both of the above triggers would be attached to test-deployment after running prefect deploy.
Triggers passed to prefect deploy will override any triggers defined in prefect.yamlWhile you can define triggers in prefect.yaml for a given deployment, triggers passed to prefect deploy take precedence over those defined in prefect.yaml.
Note that deployment triggers contribute to the total number of automations in your workspace.

​
Further reading

For more on the concepts behind deployment triggers, see the Automations concepts page.