hey @Marvin what is the preferred way to broadcast events in Prefect? For instance, I have a flow that untars tar files that land in an SFTP, I would like to broadcast a metadata about that load. What features of Prefect are available to me? From my knowledge, I'm thinking automations, the

from prefect.input.run_input import receive_input

api, and also polling the prefect server for certain flow runs are solutions. What do you think?

✅ thought for 403.2 seconds

emit_event

is probably what you want, marvin should hopefully say as much

is it possible to pause a flow until an event occurs?

Great question — and you’re on the right track. In Prefect 3.x, the preferred way to “broadcast” metadata that other parts of your system can react to is: - Broadcast with events: use

prefect.events.emit_event()

to publish a custom event with a payload and resource labels - React with Automations: use an Event Trigger to kick off downstream deployments, send notifications/webhooks, etc. - Store rich metadata on the run: use artifacts (tables/markdown/links/progress) for UI visibility - Use Run Input for point-to-point messaging: only when you want to send typed data to a specific flow run (not a broadcast) - Avoid polling: prefer event subscribers or automations over polling the API A quick mapping to your ideas - Automations: yes — best for reacting to events across flows and for integrations (Slack, webhooks, run deployments). - `receive_input`: it’s for targeted, typed messaging to a specific flow run; it is not a broadcast and doesn’t trigger automations. - Polling Prefect Server: works but generally not recommended when events and subscribers exist. Recommended pattern for your SFTP untar flow 1) Emit a custom event with your load’s metadata

from prefect import flow, runtime
from prefect.events import emit_event

@flow
def extract_tar(file_path: str, sftp_host: str):
    # ... untar, compute metadata ...
    meta = {"file": file_path, "rows": 12345, "bytes": 987654321}

    emit_event(
        event="data.ingest.sftp.tar_extracted",
        resource={"prefect.resource.id": f"sftp://{sftp_host}{file_path}"},
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{runtime.flow_run.id}",
                "prefect.resource.role": "producer",
            }
        ],
        payload=meta,
    )

2) React with an Automation (UI recommended) - Create an Event Trigger that matches

data.ingest.sftp.tar_extracted

- Add actions like: - Run a downstream deployment (pass

{{ event.payload.* }}

parameters) - Send a Slack/email notification - Call a webhook with the event payload Docs for this flow: - Events concept and API: Events • emit_event() - Automations and triggers: Automations • Event triggers - Passing event payloads to flows: Pass event payloads to flows - Chaining deployments with events: Chaining deployments Optional: subscribe to events programmatically If you want a service or a separate flow to listen in real-time without polling:

import asyncio
from prefect.events import get_events_subscriber
from prefect.events.filters import EventFilter, EventNameFilter

async def main():
    f = EventFilter(event=EventNameFilter(prefix=["data.ingest.sftp.tar_extracted"]))
    async with get_events_subscriber(filter=f) as sub:
        async for event in sub:
            print("Got event:", event.resource, event.payload)

asyncio.run(main())

- Subscribers API: get_events_subscriber - Event filters: Event filters Include rich metadata in the UI with artifacts Artifacts are great for surfacing details of the load on the flow run page (but they do not trigger automations): ``` from prefect.artifacts import create_table_artifact, create_markdown_artifact create_table_artifact( table=[{"member": "file1.csv", "rows": 5000}, {"member": "file2.csv", "rows": 7345}], key="untar-members", description="Files extracted from tar" ) create_markdown_artifact( markdown=f"# SFTP load\nFile: {file_path}\nRows: {meta['rows']}\nBytes: {meta['bytes']}",

yea you should be able to use the suspend and resume actions