This message was deleted.
# ask-marvin
s
This message was deleted.
m
thought for 141.7 seconds
Yes — you can trigger automations off a custom flow run state name. In Prefect 3.x, a state has both: - a state type like
COMPLETED
,
FAILED
, etc. - a state name like
Completed
, or a custom name you set Automations are driven by events, and the emitted event name uses the state name, not just the state type. So if your flow ends in a completed state named something custom, Prefect emits an event like:
Copy code
text
prefect.flow-run.YourCustomStateName
So your example is possible: - Flow A finishes with a custom final state name - automation listens for
prefect.flow-run.<CustomName>
- automation triggers Flow B only for that specific completion condition Recommended approach Instead of triggering on all
Completed
states, create a custom completed state in Flow A when your condition is met, and then target that event in the automation. Conceptually:
Copy code
python
from prefect import flow
from prefect.states import Completed

@flow
def flow_a():
    condition = True

    if condition:
        return Completed(name="TriggerDownstream")
    else:
        return Completed(name="SkipDownstream")
Then your automation would listen for:
Copy code
text
prefect.flow-run.TriggerDownstream
not the generic:
Copy code
text
prefect.flow-run.Completed
A couple of important notes - The state type is still
COMPLETED
- The custom name is what lets you distinguish which completion happened - This is a good fit when you want event-driven branching between flows Alternative filtering options You may also be able to filter on event/resource labels like: -
prefect.state-name
-
prefect.state-type
But the cleanest pattern is usually to match the event name directly. Best practice If you use this for downstream orchestration, also filter by the upstream deployment/flow identity so you don’t accidentally trigger Flow B from some other flow that emits the same custom state name. If you want, I can give you: 1. a UI automation setup example, or 2. a Python deployment trigger example for Flow A → Flow B. Relevant docs: - Customizing automation triggers - Event triggers - Creating deployment triggers
thought for 10.0 seconds
Yep — here’s a basic event trigger JSON that will fire only when Flow A enters your custom completed state, e.g.
TriggerDownstream
. If you’re creating an automation trigger, the core trigger would look like:
Copy code
json
{
  "type": "event",
  "expect": [
    "prefect.flow-run.TriggerDownstream"
  ]
}
If you also want to restrict it to only Flow A (recommended), add a match filter. For example, if you want to match the upstream deployment name or related resource:
Copy code
json
{
  "type": "event",
  "expect": [
    "prefect.flow-run.TriggerDownstream"
  ],
  "match_related": {
    "prefect.resource.name": [
      "flow-a/deployment-name"
    ]
  }
}
A fuller automation body that runs Flow B might look like this:
Copy code
json
{
  "name": "trigger-flow-b-on-custom-flow-a-success",
  "trigger": {
    "type": "event",
    "expect": [
      "prefect.flow-run.TriggerDownstream"
    ],
    "match_related": {
      "prefect.resource.name": [
        "flow-a/deployment-name"
      ]
    }
  },
  "actions": [
    {
      "type": "run-deployment",
      "source": "selected",
      "deployment_id": "YOUR_FLOW_B_DEPLOYMENT_ID"
    }
  ]
}
A few caveats: -
TriggerDownstream
must be the actual state name you set on the final state. -
match_related
values depend on which related resource you want to match against; deployment filtering is usually safest. - If you’re building this through the UI/API, the exact resource identifier/value may differ depending on whether you match by name or id. If you want, I can give you either: - a minimal trigger JSON only, or - the exact full automation JSON for a deployment if you share how you identify Flow A and Flow B. Docs: - Event triggers - Customizing automation triggers - Creating deployment triggers