Tom Han
10/02/2025, 5:08 AMflow , deployment , flow_run , work_pool , work_queue , and metric ." What exactly can I use?
Can I check the artifact created and get the link to that artifact? Can I get the return value of the flowrun? If I want to emit some information from within the flow, but it is just a number of something (so creating an artifact is a bit too much), what should I do? I just want to include some statistics on the notification I sent via slack webhook.Marvin
10/02/2025, 5:08 AMMarvin
10/02/2025, 5:15 AMflow — the Flow object
- deployment — the Deployment object (if present)
- flow_run — the Flow run that’s relevant to the trigger
- work_pool, work_queue — when applicable
- metric — only when the Automation is triggered by a MetricTrigger
- event — when the Automation is triggered by an EventTrigger; includes event.resource, event.related, and event.payload
What you can reference (practically)
- These objects are Pydantic models with attributes like id, name, and so on; your template can use Jinja paths like {{ flow_run.id }}, {{ flow_run.name }}, {{ flow_run.state.type }}, {{ deployment.name }}, {{ work_pool.name }}, {{ work_queue.name }}.
- For Metric-triggered automations, you can use {{ metric.name }}, {{ metric.value }}, {{ metric.period }} (the exact fields depend on the metric).
- For Event-triggered automations, you can use {{ event.name }}, {{ event.resource }}, and any custom payload you included, e.g. {{ event.payload.records_processed }}.
Can the template access artifacts or a flow’s return value?
- Artifacts: there is no built-in artifact object in the notification template. The template cannot query for “latest artifact for this run” on its own. To include artifact details, a common pattern is to emit a custom event from your flow that includes the artifact metadata (e.g., its ID or a link) in event.payload — then reference {{ event.payload.artifact_url }} in your template.
- Flow return value: the template does not have direct access to the flow’s return value (which is stored in result storage). If you want it in a notification, emit it yourself via a custom event (or log event), or send a notification from within your flow.
Recommended patterns to include custom numbers/stats in Slack notifications
1) Emit a custom event (best for structured payloads)
- From your flow, emit an event with your stats in the payload and optionally point to an artifact you created.
from prefect import flow
from prefect.artifacts import create_markdown_artifact
from prefect.events import emit_event
from prefect.runtime import flow_run
@flow
def my_flow():
# Compute a number you want to surface
records_processed = 123
# Optional: create an artifact for richer context
art = create_markdown_artifact(
markdown=f"# Run Summary\n\nRecords processed: {records_processed}"
)
# Emit a custom event with your number and artifact reference
emit_event(
event="my.app.summary",
resource={
"prefect.resource.id": f"prefect.flow-run.{flow_run.id}",
"prefect.resource.name": flow_run.name,
},
related=[{"prefect.resource.id": f"prefect.artifact.{art.id}"}],
payload={
"records_processed": records_processed,
"artifact_id": str(art.id),
},
)
- Create an Automation with an EventTrigger for my.app.summary, and use a template like:
{{ flow_run.name }} finished in {{ flow_run.state.type }}.
Records processed: {{ event.payload.records_processed }}
Run: <{{ flow_run.ui_url }}|Open in Prefect>
{% if event.payload.artifact_id %}Artifact ID: {{ event.payload.artifact_id }}{% endif %}
2) Use a log-based event (lightweight, no artifacts)
- Log a line with a recognizable marker, then trigger an Automation on Prefect’s log events with a text filter:
from prefect import flow, get_run_logger
@flow
def my_flow():
n = 123
logger = get_run_logger()
<http://logger.info|logger.info>(f"[SUMMARY] records_processed={n}")
- Configure an EventTrigger for prefect.log.write with a text filter like [SUMMARY] and a resource filter for flow runs. Template:
```
{{ flow_run.name }} reported: {{ event.payload.message }}Marvin
10/02/2025, 5:15 AM3) Send a notification from within the flow (bypass automations when desired)
- If you’d rather not wire automations, you can send directly via the Slack webhook block:
from prefect.blocks.notifications import SlackWebhook
from prefect import flow
@flow
def my_flow():
n = 123
slack = SlackWebhook.load("my-slack-webhook")
slack.notify(f"Run { { '{' } }flow_run.name{ { '}' } }} processed {n} records.")
```
Note: the inline {{ }} won’t resolve automatically inside a Python f-string — typically you’ll build the string directly with values, e.g., f"Run {flow_run.name} processed {n}..." if you pull the run name from runtime or from parameters.
About links
- Flow run link: you can use {{ flow_run.ui_url }} in templates to link directly to the run in the UI.
- Artifact links: templates do not get artifacts automatically. If you need a clickable link, either:
- Include a link in your event payload and reference it in the template, or
- Create a “link” artifact (which points to an external URL) and include that URL in your event payload.
References
- Automations (concepts and triggers): Automations
- Events (and event payloads/resources): Events
- Notification blocks (Slack, Teams, etc.): Notification blocks
- Artifacts: Artifacts API
- Objects API (FlowRun, Deployment, etc. — attributes you can reference): Objects API
If you can share which stats you want to include and what should trigger the message (e.g., “on flow success,” “on a custom event,” or “on a metric threshold”), I can sketch the exact event + template setup for your use case.