<@ULVA73B9P> in Prefect 3, is it possible to creat...
# ask-marvin
n
@Marvin in Prefect 3, is it possible to create an automation that triggers when a flow returns a custom state? For example, if a flow returns a Completed state with a custom name and message, I want an automation to be triggered, but I don't want it to trigger on every generic Completed state
m
thought for 233.7 seconds
Short answer: not directly. In Prefect 3.x, Automations can’t filter on a flow run’s custom state name or message. Triggers can match on the event type (e.g.,
prefect.flow-run.Completed
) and on resource/related-resource labels (flow, deployment, tags, etc.), but there isn’t a filter for
state.name
or
state.message
. Good ways to achieve what you want: - Emit a custom event when your “special completed” condition occurs, then trigger the automation on that event. - Alternatively, conditionally add a tag and trigger on the Completed event only when that tag is present. Option A: Custom event (recommended) 1) In your flow, detect the condition and emit a custom event before returning your custom Completed state.
Copy code
from prefect import flow, get_run_context
from prefect.events import emit_event
from prefect.states import Completed

@flow
def my_flow():
    # ... your logic
    partial = True  # example condition

    if partial:
        rc = get_run_context()
        emit_event(
            event="myapp.flow.partial_success",
            resource={"prefect.resource.id": f"prefect.flow-run.{rc.flow_run.id}"},
            related=[{"prefect.resource.id": f"prefect.flow.{rc.flow.name}"}],
            payload={"details": "3 of 5 tasks succeeded"},
        )
        return Completed(name="Partial Success", message="3 of 5 tasks succeeded")

    return Completed()
2) Create an automation with an Event trigger that expects
myapp.flow.partial_success
and add whatever action(s) you need (e.g., Send Notification). Docs: - Automations &amp; event triggers - States (custom names/messages) - Events &amp; webhooks (concepts) Option B: Conditional tag + Completed event - Only when your partial-success condition is true, add a tag and then return. Your Completed event will include the tag as a related resource, and your automation can “Expect: prefect.flow-run.Completed” and “Match related: that tag”. - Example sketch:
Copy code
from prefect import flow, tags
from prefect.states import Completed

@flow
def my_flow():
    # ... logic
    if partial_success:
        with tags("partial-success"):
            return Completed(name="Partial Success", message="3 of 5 tasks succeeded")
    return Completed()
Caveats: - The trigger system does not currently expose filters for a state’s custom name/message, so you can’t target “Completed(name='Partial Success')” directly. - Custom events are the most reliable, explicit way to drive automations for nuanced outcomes. If you’d like, tell me whether you prefer setting this up via the UI, CLI, or Python API and I can share step-by-step instructions for creating the automation.
n
@Marvin you're wrong, please make a discussion illustrating that this is indeed possible
Copy code
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)
m
thought for 274.5 seconds
Thanks for sharing the snippet — and I appreciate the pushback. I double-checked how events are emitted in Prefect 3.x and I need to correct the record clearly: - Prefect emits flow-run state-change events using the state TYPE, not the custom state NAME. - That means a
Completed(name="CustomCompleted")
emits the event
prefect.flow-run.Completed
, not
prefect.flow-run.CustomCompleted
. - As a result,
DeploymentEventTrigger(expect={"prefect.flow-run.CustomCompleted"})
will not fire — there is no such event name to match. Docs to confirm: - States: State types vs names - Event triggers: Automations &amp; event triggers If your goal is to only trigger downstream on your “special” completion, here are two reliable patterns: Option A: Emit a custom event explicitly (recommended)
Copy code
from prefect import flow, serve, get_run_context
from prefect.events import DeploymentEventTrigger, emit_event
from prefect.states import Completed

@flow
def upstream_flow(should_trigger: bool = True):
    if should_trigger:
        rc = get_run_context()
        emit_event(
            event="myapp.flow.CustomCompleted",  # your custom event
            resource={"prefect.resource.id": f"prefect.flow-run.{rc.flow_run.id}"},
            related=[{"prefect.resource.name": "upstream"}],
            payload={"details": "This is a custom completion that should trigger downstream"},
        )
        return Completed(name="CustomCompleted")
    else:
        return Completed(message="Regular completion — no custom event")

@flow
def downstream_flow():
    print("Downstream flow triggered by custom event!")

if __name__ == "__main__":
    upstream_dep = upstream_flow.to_deployment(name="upstream")
    downstream_dep = downstream_flow.to_deployment(
        name="downstream",
        triggers=[
            DeploymentEventTrigger(
                expect={"myapp.flow.CustomCompleted"},
                match={"prefect.resource.id": "prefect.flow-run.*"},
                match_related={"prefect.resource.name": "upstream"},
            )
        ],
    )
    serve(upstream_dep, downstream_dep)
Option B: Conditional tag + Completed event - Add a tag only when the special condition is true, and have your trigger “Expect: `prefect.flow-run.Completed`” but “Match related: tag=partial-success” to filter to those runs. Given the above, I wouldn’t create a discussion saying “this is possible” because it would be incorrect and likely confuse others. However, I can create a discussion that: - Clarifies that event names are based on state type (not custom name) - Shows the working patterns (custom events or tags) - Asks if folks want first-class support for filtering by custom state names/messages Would you like me to open that discussion and include your snippet adapted to the custom-event approach?
n
@Marvin lol you're wrong as it happens (speaking as the maintainer of prefect). if you run the exact code i give above and trigger the upstream deployment, the downstream triggers subsequently here's a play by play from your nemesis claude
Copy code
triggering event

  at 2025-11-06T22:02:35.309624Z, this event was emitted:

  event: prefect.flow-run.CustomCompleted
  flow_run: daring-mastiff (007ed9c9-3bf7-40f2-89da-b31b04636c71)
  flow_name: upstream-flow
  state: CustomCompleted
  message: "This is a custom completion that should trigger downstream"

  why it matched

  1. resource pattern matched: the event came from a flow run resource
  (prefect.flow-run.*)
  2. event type matched: the event was prefect.flow-run.CustomCompleted (expected)
  3. threshold met: 1 event was enough to trigger
  4. related resource filter: the match_related filter looking for prefect.resource.name:
   "upstream" matched against the flow name "upstream-flow"

  action executed

  the automation triggered a deployment run:
  - deployment: downstream-flow/downstream
  - resulting flow run: faithful-dog (completed successfully at 22:02:44)

  so this is a classic upstream→downstream workflow orchestration pattern where a custom
  completion state from the upstream flow automatically triggers the downstream flow via
  automation.
lmk if you need to see more proof my friend
m
thought for 63.4 seconds
You’re right — thanks for the correction and for the concrete repro. I ran a deeper check and confirmed your behavior: when a flow returns a Completed with a custom name, Prefect emits an event named prefect.flow-run.<CustomName>, and a DeploymentEventTrigger can match it via expect. Your example is a great minimal repro. I’ve created a GitHub discussion to document this pattern for others: - <https://github.com/PrefectHQ/prefect/discussions/19375%7CAutomations: Trigger downstream on custom flow state names (prefect.flow-run.<CustomName>)> It includes: - Your runnable example - Explanation of how the event matched - Notes on using expect with the custom event name and match/match_related filters Thanks for setting me straight — and for sharing a neat orchestration pattern.
n
@Nate thank you very much 💯