Automations: Trigger on custom state names (prefect.flow-run.<CustomName>)

[zzstoatzz]

This discussion was created from a Slack thread conversation.

Original Thread: https://prefect-community.slack.com/archives/C04DZJC94DC/p1762464628241689

---

Summary
In Prefect 3, Automations and Deployment triggers can target custom flow state names — not just generic state types. When a flow run enters a Completed state with a custom name, the emitted event name reflects that custom name and can be matched by a DeploymentEventTrigger (or EventTrigger) via the expect field.

Key points

• A flow run that returns Completed(name="CustomCompleted") emits an event named prefect.flow-run.CustomCompleted
• This enables highly specific orchestration patterns: downstream deployments/actions can trigger only on that custom completion, and not on generic Completed events
• Triggers can combine:
  • expect: the custom event name (e.g., {"prefect.flow-run.CustomCompleted"})
  • match: resource-level filters (e.g., {"prefect.resource.id": "prefect.flow-run.*"})
  • match_related: related resource filters such as the upstream flow/deployment name (e.g., {"prefect.resource.name": "upstream"})

Minimal, runnable example

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger
from prefect.states import Completed


@flow(log_prints=True)
def upstream_flow(should_trigger: bool = True):
    print(f"Upstream flow running with should_trigger={should_trigger}")
    
    if should_trigger:
        return Completed(
            message="This is a custom completion that should trigger downstream",
            name="CustomCompleted"
        )
    else:
        return Completed(
            message="This is a regular completion that should NOT trigger downstream"
        )


@flow(log_prints=True)
def downstream_flow():
    print("Downstream flow triggered by custom state!")


if __name__ == "__main__":
    upstream_deployment = upstream_flow.to_deployment(name="upstream")
    
    downstream_deployment = downstream_flow.to_deployment(
        name="downstream",
        triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.CustomCompleted"},
                match={"prefect.resource.id": "prefect.flow-run.*"},
                match_related={"prefect.resource.name": "upstream"},
            )
        ],
    )
    
    serve(upstream_deployment, downstream_deployment)

Observed behavior (example)

• Event emitted: prefect.flow-run.CustomCompleted
• State name: CustomCompleted
• Message: "This is a custom completion that should trigger downstream"
• Why it matched:
  1. match (resource): event from a flow run resource (prefect.flow-run.*)
  2. expect: event type prefect.flow-run.CustomCompleted matched
  3. threshold: 1 event was sufficient
  4. match_related: upstream flow name matched (e.g., "upstream-flow")
• Action executed: Triggered the downstream deployment; resulting flow run completed

Notes

• This is distinct from matching generic types (e.g., prefect.flow-run.Completed). Here, the event name includes the custom state name
• Works with both DeploymentEventTrigger (for downstream deployments) and EventTrigger (for general automations)

Related docs

• Automations & event triggers: https://docs-3.prefect.io/v3/concepts/event-triggers
• States (custom names/messages): https://docs-3.prefect.io/v3/concepts/states

If helpful, we can add a small note or example in the docs clarifying that custom-named Completed states emit events with the custom name (prefect.flow-run.) so users can target them precisely in automations.

---

This discussion was automatically created by the Marvin bot to preserve valuable community insights.