Great question. Your automation will fire when it sees 3 events named

animal.walked

that also include

genus=Marmota

and

species=monax

. Since your action is

CancelFlowRun()

, the event’s primary resource must identify the flow run you want to cancel. Here are two reliable ways to emit matching events: 1) From Python (inside a flow) — cancels the current flow run

from prefect import flow
from prefect.events import emit_event
from prefect.runtime import flow_run
from datetime import timedelta
from prefect.automations import Automation, EventTrigger, Posture, CancelFlowRun

# Tip: set an explicit 'within' window since threshold > 1
automation = Automation(
    name="woodchonk",
    trigger=EventTrigger(
        expect={"animal.walked"},
        match={"genus": "Marmota", "species": "monax"},
        posture=Posture.Reactive,
        threshold=3,
        within=timedelta(minutes=5),
        # Optional: count per flow run instead of globally
        # for_each=["prefect.resource.id"],
    ),
    actions=[CancelFlowRun()],
).create()

@flow
def walker():
    # Emit 3 matching events that target THIS flow run
    for i in range(3):
        emit_event(
            event="animal.walked",
            resource={
                "prefect.resource.id": f"prefect.flow-run.{flow_run.id()}",
                "genus": "Marmota",
                "species": "monax",
            },
            payload={"step": i + 1},
        )

if __name__ == "__main__":
    walker()

2) Via REST (any client) — cancels the referenced flow run Repeat this POST 3 times (within your chosen window). Use the target flow run id in the resource id.

curl -X POST "$PREFECT_API_URL/api/events" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -d '[
    {
      "event": "animal.walked",
      "resource": {
        "prefect.resource.id": "prefect.flow-run:REPLACE-WITH-FLOW-RUN-UUID",
        "genus": "Marmota",
        "species": "monax"
      },
      "payload": {"note": "walked a little"}
    }
  ]'

Important notes: - The

CancelFlowRun

action needs the event’s primary resource to be a flow run:

prefect.resource.id = "prefect.flow-run.<uuid>"

. You can still include your custom labels (genus/species) alongside it. - Because you set

threshold=3

, make sure the events arrive within a time window. I recommend setting

within=timedelta(...)

explicitly, as shown, to avoid ambiguity. - To verify events are arriving, run:

prefect events stream

Docs: - Events overview - EventTrigger matching and thresholds - POST /api/events Quick check: Do you want to cancel the same flow run that emits the events, or a different run? And should the “3 events” be counted per run (use

for_each=["prefect.resource.id"]

) or across all runs?

Absolutely—here’s how it works and why the primary resource matters. - Each event has a single “primary resource” described by labels in its

resource

dict. The key label is

prefect.resource.id

, which uniquely identifies what the event is “about.” - Automation actions operate on the primary resource of the event that caused the trigger to fire. - Flow-run actions (like

CancelFlowRun

) need that primary resource to be a flow run. Otherwise, they don’t know which run to act on. What to set for a flow run resource id - Use this exact format for the primary resource id: -

prefect.resource.id = "prefect.flow-run.<uuid>"

- Note: it’s a dot before the UUID, not a colon. Sorry for the earlier inconsistency. - You can get the current run’s UUID inside a flow with

prefect.runtime.flow_run.id()

. Why not rely on “related” resources? - When you emit events inside a flow, Prefect will automatically attach the current flow run as a related resource. However,

CancelFlowRun

targets the event’s primary resource—not a related one. So you must make the flow run the primary resource if you want it canceled. Putting it together - To cancel the current run after 3 matching events:

from prefect import flow
from prefect.events import emit_event
from prefect.runtime import flow_run
from datetime import timedelta
from prefect.automations import Automation, EventTrigger, Posture, CancelFlowRun

Automation(
    name="woodchonk",
    trigger=EventTrigger(
        expect={"animal.walked"},
        match={"genus": "Marmota", "species": "monax"},
        posture=Posture.Reactive,
        threshold=3,
        within=timedelta(minutes=5),
        # Strongly recommended so counting is per run:
        for_each=["prefect.resource.id"],
    ),
    actions=[CancelFlowRun()],
).create()

@flow
def walker():
    for i in range(3):
        emit_event(
            event="animal.walked",
            resource={
                "prefect.resource.id": f"prefect.flow-run.{flow_run.id()}",
                "genus": "Marmota",
                "species": "monax",
            },
            payload={"step": i + 1},
        )

walker()

Corrected REST example (emit this 3 times within your window):

curl -X POST "$PREFECT_API_URL/api/events" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -d '[
    {
      "event": "animal.walked",
      "resource": {
        "prefect.resource.id": "prefect.flow-run.<FLOW-RUN-UUID>",
        "genus": "Marmota",
        "species": "monax"
      },
      "payload": {"note": "walked a little"}
    }
  ]'

Tips - If you want to cancel a different flow run, put that run’s UUID in the primary resource id instead. - Without

for_each=["prefect.resource.id"]

, the 3 events are counted globally; the action will apply to the primary resource of the event that actually fires the trigger. Adding

for_each

scopes counting per run, which is usually what you want here. References: - Events overview - EventTrigger matching and thresholds