Got it! Here’s how to create a custom trigger in Python for a Prefect 3.x flow and run it with serve. In 3.x, triggers are event-driven automations you attach to a deployment via

to_deployment(..., triggers=[...])

, then run with

serve(...)

. There isn’t an

@trigger

decorator. Summary of trigger types you can use: - Event-based:

DeploymentEventTrigger

(works in OSS and Cloud) - Metric-based:

DeploymentMetricTrigger

(Cloud-only) - Composite:

DeploymentCompoundTrigger

,

DeploymentSequenceTrigger

Docs: - Concepts: Event triggers - How to: Creating deployment triggers with serve/deploy - API: Deployment trigger classes Example: Trigger a flow when a custom event is emitted 1) Define the flow and attach a custom event trigger

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger, ResourceSpecification

@flow(name="process-dataset")
def process_dataset(name: str):
    print(f"Processing dataset: {name}")

# Fire when our app emits a custom event for a specific dataset
data_ready_trigger = DeploymentEventTrigger(
    expect={"my.app.DataReady"},  # your custom event name
    # Only trigger for this specific dataset resource
    match=ResourceSpecification.model_validate({
        "prefect.resource.id": "my.app.dataset:users"
    }),
    # Optional: require N events within a time window
    # threshold=1,
    # within=60
)

deployment = process_dataset.to_deployment(
    name="process-users-on-data-ready",
    parameters={"name": "users"},
    triggers=[data_ready_trigger],
)

if __name__ == "__main__":
    serve(deployment)

2) Emit your custom event from anywhere in your app

from prefect.events import emit_event

# Emit when data is ready
emit_event(
    event="my.app.DataReady",
    resource={"prefect.resource.id": "my.app.dataset:users"},
    payload={"rows": 123}  # optional context
)

That’s it—when

emit_event

publishes

my.app.DataReady

for

my.app.dataset:users

, Prefect will trigger your deployment and run the flow. Notes and variations - You can filter by related resources instead using

match_related=ResourceSpecification.model_validate({...})

. - You can make the trigger proactive (fires when an expected event is missing within a window) by setting

posture="Proactive"

and

within

(min 10s). - To combine conditions (e.g., two different events must occur), use

DeploymentCompoundTrigger

or

DeploymentSequenceTrigger

. - Metric triggers are available in Prefect Cloud: use

DeploymentMetricTrigger

with a

MetricTriggerQuery

. Want me to tailor this to your use case? Let me know: - The event you want to react to (Prefect built-in like

prefect.flow-run.Completed

or a custom one) - Any resource labels to match on - Whether you’re on Prefect Cloud or open-source (for metric triggers)

Thanks for the details — that helps. The error you’re seeing typically happens when the wrong trigger class is passed into

to_deployment(..., triggers=[...])

. In Prefect 3.x you should pass the Deployment-scoped trigger classes (e.g.,

DeploymentEventTrigger

), not the Automation-scoped ones (e.g.,

EventTrigger

). The Automation triggers expect a

name

and other fields and can cause Pydantic validation issues in the RunnerDeployment model. Two reliable fixes: 1) Use the Deployment trigger classes - Import from

prefect.events

and use the Deployment variants:

DeploymentEventTrigger

,

DeploymentCompoundTrigger

,

DeploymentSequenceTrigger

,

DeploymentMetricTrigger

(Cloud). - Example:

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger, ResourceSpecification

@flow
def process_dataset(name: str):
    print(f"Processing dataset: {name}")

trigger = DeploymentEventTrigger(
    # name is optional, but setting one avoids older-version edge cases
    name="data-ready-users",
    expect={"my.app.DataReady"},
    match=ResourceSpecification.model_validate({
        "prefect.resource.id": "my.app.dataset:users"
    }),
)

deployment = process_dataset.to_deployment(
    name="process-users-on-data-ready",
    parameters={"name": "users"},
    triggers=[trigger],
)

serve(deployment)

2) If you must use the Automation trigger classes, include a

name

- Not recommended here, but if you imported

EventTrigger

instead of

DeploymentEventTrigger

, give it a

name

explicitly to satisfy the validator. Why it worked from the UI - The UI creates the automation/trigger with the correct schema and includes required fields (like

name

) automatically, which is why attaching it there succeeds. Can you share: - Your Prefect version (

prefect version

) - The exact trigger import and construction you used If you’re on an older 3.x release, please try upgrading to the latest 3.x as well. Docs for reference: - Event trigger concepts: Concepts: Event triggers - How to attach triggers with serve/deploy: Creating deployment triggers

You can set dynamic tags at run time using the

prefect.tags

context manager. Build the tags from your flow parameters and wrap the flow body so the flow run (and any tasks inside) inherit them. Example:

from prefect import flow, tags
from prefect.context import get_run_context

@flow
def process_dataset(dataset: str, priority: int = 1):
    # compute tags from parameters
    run_tags = {f"dataset:{dataset}", f"priority:{priority}"}

    # apply tags to this flow run and all nested tasks
    with tags(*run_tags):
        # optional: verify what's on the run
        rc = get_run_context()
        print(f"Current run tags: {rc.tags}")

        # your logic here
        ...

if __name__ == "__main__":
    process_dataset("users", priority=5)

Notes: - Place

with tags(...)

at the top of the flow so the flow run itself gets the tags. - Tags must be strings. Use simple, stable tokens like

dataset:users

. - Static tags on a deployment (

.to_deployment(tags=[...])

) are fixed for all runs; use the

tags(...)

context for per-run, parameter-driven tags. Both sets will merge at runtime. - You can also scope tags to specific tasks by wrapping just those task calls in a

with tags(...)

block. If you’re using triggers that match tags, dynamic tags will show up as related resources like

prefect.tag.dataset:users

, which you can target in a trigger’s

match_related

.

Great callout. To modify the parent flow run’s tags themselves (not just tags applied to child tasks), update the running flow run via the Prefect client. In Prefect 3.x you can PATCH the current flow run with new tags. Key points: - Updating tags this way replaces the run’s tag list, so include existing tags when you add new ones. - Works from within the flow body at any time. Example (sync flow):

from prefect import flow, get_client
from prefect.context import get_run_context

@flow
def process_dataset(dataset: str, priority: int = 1):
    # Build new tags
    new_tags = {f"dataset:{dataset}", f"priority:{priority}"}

    # Get current run and existing tags
    rc = get_run_context()
    flow_run_id = rc.flow_run.id
    existing = set(rc.tags or [])

    # Update the flow run's tags (replace entire list)
    client = get_client(sync_client=True)
    client.update_flow_run(flow_run_id, tags=list(existing | new_tags))

    # Optional: confirm
    print(f"Updated parent flow run tags to: {existing | new_tags}")

    # ... rest of your flow logic ...

if __name__ == "__main__":
    process_dataset("users", priority=5)

Async variant:

from prefect import flow, get_client
from prefect.context import get_run_context
import asyncio

@flow
async def process_dataset_async(dataset: str):
    rc = get_run_context()
    flow_run_id = rc.flow_run.id
    existing = set(rc.tags or [])
    new_tags = {f"dataset:{dataset}"}

    async with get_client() as client:
        await client.update_flow_run(flow_run_id, tags=list(existing | new_tags))

This directly modifies the parent flow run’s tags, so triggers, searches, and UI filters see the updated tags immediately.