I don’t see the prior question in this thread. Could you paste the exact question (or a brief summary) here? If it’s about Prefect, a few details that will help me answer quickly: - Prefect version (I’ll assume 3.x unless you say otherwise) - Where you’re running (e.g., AWS CodeBuild/CI) - Python version - Any error message/logs and a minimal code snippet or CLI command you ran Once I have that, I’ll jump right in.

Short answer: You can’t use state.data in an Automation template. In Prefect 3.x, state.data is intended for results and is not persisted on the event that Automations use. The Automation system reconstructs the state from event labels and only includes name, type, message, and timestamp — not data — so

{{ flow_run.state.data }}

will be None. Why you’re seeing this: - When a state is sent to the API, non-result values in

state.data

are dropped (only result records are persisted). - The event emitted for

prefect.flow-run.<StateName>

only carries basic state fields (name, type, message, timestamp). Automations build the Jinja context from this event, so there’s no

data

to template. Ways to attach metadata that Automations can use: 1) Emit a custom event with a payload (recommended when metadata is produced at runtime) - Emit your own event with the flow run as the resource and put your metadata in the event payload. Then trigger your Automation on that custom event and reference

event.payload

. Example:

from prefect import flow
from prefect.states import Completed
from prefect.events import emit_event
from prefect.runtime import flow_run

@flow
def my_flow():
    # ... do work ...
    some_id = "abc-123"

    # Emit a custom event tied to the current flow run
    emit_event(
        event="myapp.CustomCompleted",
        resource={"prefect.resource.id": f"prefect.flow-run.{flow_run.id()}"},
        payload={"id": some_id, "note": "A message"},
    )

    # Keep your custom-named state if you like
    return Completed(name="CustomCompleted", message="A message")

Automation trigger (event): - match:

{"prefect.resource.id": "prefect.flow-run.*"}

- expect:

["myapp.CustomCompleted"]

Webhook body template can then use:

State message: {{ flow_run.state.message }}
Custom id: {{ event.payload.id }}
Flow run {{ flow.name }}/{{ flow_run.name }} observed in state `{{ flow_run.state.name }}` at {{ flow_run.state.timestamp }}.
Flow run URL: {{ flow_run|ui_url }}

2) Pass the metadata as a flow parameter (if known at start) - Put the

id

in a parameter and access it with

{{ flow_run.parameters.id }}

in your template. 3) Put small details in the state message (quick and dirty) - Encode the id in

message="A message (id=abc-123)"

and read it from

{{ flow_run.state.message }}

. This is unstructured and subject to length limits, so use sparingly. Notes: -

flow_run.state.state_details

is available in templates, but it does not include arbitrary metadata. - Persisted results (via result storage) are not surfaced to Automations/templates either. If you share your current Automation JSON, I can tweak it to listen for the custom event and update your webhook body template accordingly.

Yes. If the flow returned a value and you had result persistence enabled (e.g.,

persist_result=True

on the flow and S3 result storage configured), you can fetch the flow run’s state by ID and materialize the result. Prefect will automatically download and deserialize it from S3. Async example:

import asyncio
from uuid import UUID
from prefect.client.orchestration import get_client

async def get_flow_result(flow_run_id: str):
    async with get_client() as client:
        flow_run = await client.read_flow_run(UUID(flow_run_id))
        state = flow_run.state

        # Will raise if the run failed; set raise_on_failure=False to get the exception object instead
        result = await state.aresult(raise_on_failure=True, retry_result_failure=True)
        return result

# Usage
# asyncio.run(get_flow_result("YOUR-FLOW-RUN-ID"))

Sync example:

from uuid import UUID
from prefect.client.orchestration import get_client

def get_flow_result_sync(flow_run_id: str):
    with get_client(sync_client=True) as client:
        flow_run = client.read_flow_run(UUID(flow_run_id))
        state = flow_run.state
        return state.result(raise_on_failure=True, retry_result_failure=True)

# Usage
# get_flow_result_sync("YOUR-FLOW-RUN-ID")

Notes and gotchas: - The flow must be configured to persist results (e.g.,

@flow(persist_result=True, result_storage="s3-bucket/<your-block>")

or via the default result storage setting). - If the run is not in a final state, `state.result()`/`aresult()` will raise because the result isn’t ready yet. - If the run failed,

state.result()

raises by default; use

raise_on_failure=False

to get back the exception object instead. - Your environment must be able to load the S3 block referenced by the state (it contains the S3 location and credentials) or otherwise have valid AWS credentials to read from the bucket. To quickly inspect a run (state, metadata) from the CLI:

prefect flow-run inspect <FLOW_RUN_ID>

This shows state details, but materializing the actual Python result requires the Python API shown above.