You’re right to call this out—when you return a custom-named completed state (e.g.,

Completed(name="S3Files")

), the emitted event is

prefect.flow-run.S3Files

, so an automation expecting

prefect.flow-run.Completed

won’t match. To trigger on “any completed flow,” regardless of the state’s name, create an Event-triggered automation that: - expects all flow-run state events with a wildcard, and - filters by the state type label

prefect.state-type = COMPLETED

That way, it will match standard Completed as well as custom-named completed states (and note: it will also include Cached and RolledBack since they are type COMPLETED). Example automation YAML (works in Prefect 3.x):

name: "On any completed flow (regardless of state name)"
description: "Fires whenever a flow run reaches a COMPLETED-type state, even with a custom name"
enabled: true
trigger:
  type: event
  posture: Reactive
  expect:
    - "prefect.flow-run.*"              # catch any flow-run state event
  match:
    "prefect.state-type": "COMPLETED"   # restrict to COMPLETED type only
    "prefect.resource.id": "prefect.flow-run.*"
  threshold: 1
actions:
  - type: send-notification
    subject: "Flow {{ flow_run.name }} completed"
    body: |
      Flow run {{ flow_run.name }} completed with state:
      - name: {{ event.resource['prefect.state-name'] }}
      - type: {{ event.resource['prefect.state-type'] }}
      - id: {{ event.resource['prefect.resource.id'] }}

Create it with the CLI (verified):

prefect automation create --from-file automation.yaml

Notes and tips: - Custom completed states change the event name, not the type.

Completed(name="S3Files")

emits

prefect.flow-run.S3Files

with labels

prefect.state-type=COMPLETED

and

prefect.state-name=S3Files

. - If you only want a subset of COMPLETED-type states (e.g., exclude Cached/RolledBack), say so and we can refine the

match

further. - You can scope to specific flows/deployments/tags using

match_related

(e.g., match a deployment id or a tag). Helpful docs: - Automations in Prefect - Events and event payloads Does this behavior (including Cached and RolledBack as COMPLETED-type) match what you want, or should we exclude those?

Short answer: Yes—this configuration will work. It will match those state types regardless of the state name. Why it works: -

expect: ["prefect.flow-run.*"]

catches any flow-run state event, including custom-named states (e.g.,

prefect.flow-run.S3Files

). -

match

then filters to only the state types you listed.

prefect.state-type

supports list values, so this matches any of: - COMPLETED, SCHEDULED, RUNNING, FAILED, CRASHED - This means even if the state name is custom (e.g.,

Completed(name="S3Files")

→ event

prefect.flow-run.S3Files

), it will still match because the label

prefect.state-type

is

COMPLETED

. A few validation notes: -

within: 0

is valid for Reactive triggers. -

match_related

with

prefect.resource.role: "tag"

and

prefect.resource.id: ["prefect.tag.job"]

is correct to scope to runs related to that tag. - Lists in

match

and

match_related

are supported. Wildcards and negation are supported in label values if you ever need them. If you want to narrow to specific flows/deployments later, you can add additional

match_related

entries (e.g., a deployment id) alongside the tag filter.

Yes—attach your return value via the state’s data field. In Prefect 3.x you can create a custom-named Completed state and include the result like this:

from prefect import flow
from prefect.states import Completed

@flow
def my_flow():
    return Completed(
        name="SomeName",
        message="Finished with a custom name",
        data="asd"  # your result payload goes here
    )

# Programmatic retrieval
state = my_flow(return_state=True)
print(state.name)         # "SomeName"
print(state.type.value)   # "COMPLETED"
print(state.result())     # "asd"

Notes: - Returning a raw value (e.g.,

return "asd"

) auto-wraps it in a standard Completed state named “Completed.” If you want a custom name, explicitly return a State (as above) and pass your result in

data=...

. - To retrieve the result from code, use

state.result()

(or

await state.aresult()

for async). - If you need the result available later via the API/UI (not just in-memory), enable result persistence (e.g.,

@flow(persist_result=True)

or a results block). Ensure your

data

object is serializable by your chosen serializer (pickle by default).