<@ULVA73B9P> i am emiting an event ```{ "id": "0...
# marvin-ai
k
@Marvin i am emiting an event
Copy code
{
  "id": "019956a3-c8dc-7e4f-a713-102ef7258136",
  "event": "axos.gold.table.updated",
  "occurred": "2025-09-17T07:46:38.427Z",
  "payload": {
    "table_name": "gold.invoice_monthly_summary",
and this calls a trigger to update something else, is it possible to pass a value from the event to the automation as a param for the deployment which takes table_name as a payload or is this not possible
m
thought for 246.3 seconds
Yes — in Prefect 3.x you can pass values from the triggering event into a Run Deployment action using Jinja templates. How to reference your event payload - In an Automation action, parameters can be templated with
{{ event.payload.<key> }}
. - For your example, use
{{ event.payload.table_name }}
. If you’re configuring the Automation in the UI - Add an action: Run deployment - In Parameters, add a key
table_name
- Click the template toggle (</>) and set the value to:
Copy code
{{ event.payload.table_name }}
If you’re defining the Automation as JSON (or via API), templated fields use the special JSON form:
Copy code
"parameters": {
  "table_name": {
    "__prefect_kind": "jinja",
    "template": "{{ event.payload.table_name }}"
  }
}
Notes - Make sure your deployment’s flow has a
table_name
parameter in its signature. - You can add simple logic/defaults in the template, e.g.:
Copy code
{{ event.payload.table_name or 'unknown_table' }}
Docs - Using templates in Automations - Available Jinja filters and context If you share how you’re creating the Automation (UI vs code) I can provide the exact steps or a ready-to-paste snippet.
k
@Marvin so the event that is coming through is this:
Copy code
{
  "id": "019956b4-18a6-7c81-bcbf-47a93ad40c0a",
  "event": "axos.gold.table.updated",
  "occurred": "2025-09-17T08:04:27.429Z",
  "payload": {
    "table_name": "gold.invoice_monthly_summary",
    "timestamp": "2025-09-17T20:04:27.428100"
  },
  "received": "2025-09-17T08:04:27.466Z",
  "related": [
    {
      "prefect.resource.id": "prefect.flow-run.37b3953c-33f8-49a3-8905-6b1f7f19ce33",
      "prefect.resource.role": "flow-run",
      "prefect.resource.name": "clever-whippet"
    },
    {
      "prefect.resource.id": "prefect.task-run.019956b3-9734-7fd7-9dad-da2a5a996c9d",
      "prefect.resource.role": "task-run",
      "prefect.resource.name": "InvoiceMonthlySummaryGoldWorker-644"
    },
    {
      "prefect.resource.id": "prefect.flow.8d5bdcf8-db0d-4a9e-ad44-2816ade6496b",
      "prefect.resource.role": "flow",
      "prefect.resource.name": "silver-to-gold-pipeline"
    }
  ],
  "resource": {
    "prefect.resource.id": "gold.invoice_monthly_summary",
    "table": "gold.invoice_monthly_summary",
    "worker": "invoice_monthly_summary"
  }
}
i need this to be caught through a trigger and call my dashboard-widget-update deployment. this deployment takes a table_name as a key, as shown by the python code that it is connected to:
Copy code
@flow(name="update-dashboard-widget")
def update_dashboard_widget(table_name):
i would like the steps to create this in the ui, so i can pass the table name from the event to the deployment.
m
thought for 17.6 seconds
Absolutely — you can wire this up in the UI so your automation passes
table_name
from the event payload to your deployment. Here are the UI steps in Prefect 3.x: 1) Open Automations - In the Prefect UI, go to Automations. - Click “New automation”. 2) Name and Trigger - Give it a name like “Update dashboard widget on gold table update”. - Trigger: choose “Event occurs”. - Condition: set Event name equals
axos.gold.table.updated
. 3) Action: Run deployment - Add action: “Run deployment”. - Select your deployment: “dashboard-widget-update” (or the exact name shown in your Deployments list). - Parameters: - Add a parameter key:
table_name
- Click the template toggle (</>) for the value field - Enter:
Copy code
{{ event.payload.table_name }}
- Save the action. 4) Save and test - Save the automation. - When the event you showed is emitted, the automation will run the deployment and pass the value
gold.invoice_monthly_summary
into your flow’s
table_name
argument (matching your flow signature
def update_dashboard_widget(table_name):
). Notes and options: - Your event also includes
resource.table
. If you prefer that source, you can use:
Copy code
{{ event.resource.table }}
- Make sure your deployment’s flow parameter is named exactly
table_name
. If it differs, update either the flow signature or the parameter key in the action. Docs - Using templates in Automations - Available Jinja filters/context If you share a screenshot of your deployment’s parameters panel, I can double-check the exact field names and confirm the template entry for you.
k
@Marvin in the 02 actions section in the automation set up there is a parameters field, what option do i pick to have the correct option to set the table name? when i click the 3 dots i have the options, use form input and Jinja input or use JSON input. which one is: {{ event.payload.table_name }}
m
thought for 5.9 seconds
Pick “Use form input and Jinja input”. Then: - Click Add parameter - Key:
table_name
- Click the template toggle (</>) for the value - Paste:
Copy code
{{ event.payload.table_name }}
- Save If you prefer “Use JSON input”, the equivalent is:
Copy code
{
  "table_name": {
    "__prefect_kind": "jinja",
    "template": "{{ event.payload.table_name }}"
  }
}
Docs: - Using templates in Automations - Available Jinja filters/context
k
@Marvin i have an automation: params (JSON):
Copy code
{
  "table_name": {
    "__prefect_kind": "jinja",
    "template": "{{ event.payload.table_name }}"
  }
}
job variables:
Copy code
{}
trigger: { "type": "event", "match": {}, "match_related": {}, "after": [], "expect": [ "axos.gold.table.updated" ], "for_each": [], "posture": "Reactive", "threshold": 1, "within": 10 } action: run deploymnet Deployment: update-dashboard-widget > dashboard-widget-update name: update dashboard trigger ---- i want to make this reproducible through a run once script, how can i do this in python so this automation is created (currently have my work pool and deployments automatically created through my script) currently i execute commands like so so what prams would i pass:
Copy code
def run_deploy_command(command, description):
    """Run a deployment command"""
    try:
        result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=os.getcwd())
m
thought for 524.2 seconds
Great question. Short answer: in Prefect 3.x there isn’t a CLI command to create automations, so you won’t pass a
prefect automation create ...
command to your subprocess runner. Instead, create the automation via Python using the Prefect client. Below is a minimal async script that: - looks up your deployment by name (
update-dashboard-widget/dashboard-widget-update
) - creates an event-triggered automation with your Jinja parameter mapping for
table_name
- sets the action to “Run deployment” with empty job variables You can call this from your “run once” script.
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.events.schemas.automations import AutomationCore, EventTrigger, Posture
from prefect.events.actions import RunDeployment

FLOW_NAME = "update-dashboard-widget"
DEPLOYMENT_NAME = "dashboard-widget-update"
AUTOMATION_NAME = "update dashboard trigger"

async def create_automation():
    async with get_client() as client:
        # Resolve deployment id by flow and deployment name
        deployment = await client.read_deployment_by_name(
            name=DEPLOYMENT_NAME,
            flow_name=FLOW_NAME,
        )
        if deployment is None:
            raise RuntimeError(f"Deployment not found: {FLOW_NAME}/{DEPLOYMENT_NAME}")

        # Build the automation definition
        automation = AutomationCore(
            name=AUTOMATION_NAME,
            enabled=True,
            trigger=EventTrigger(
                match={},
                match_related={},
                after=[],
                expect=["axos.gold.table.updated"],
                for_each=[],
                posture=Posture.Reactive,
                threshold=1,
                within=10,
            ),
            actions=[
                RunDeployment(
                    deployment_id=deployment.id,
                    parameters={
                        "table_name": {
                            "__prefect_kind": "jinja",
                            "template": "{{ event.payload.table_name }}",
                        }
                    },
                    job_variables={},
                )
            ],
        )

        created = await client.create_automation(automation)
        print(f"Created automation: {created.name} ({created.id})")

if __name__ == "__main__":
    asyncio.run(create_automation())
Notes: - This assumes your script is already authenticated to Prefect Cloud (via
PREFECT_API_KEY
and
PREFECT_API_URL
) and the correct workspace is set. - If you want the action to reference the deployment by name instead of id, we can adjust the payload, but using
deployment_id
is the most reliable. - The trigger body mirrors what you pasted:
type: "event"
,
posture: "Reactive"
,
expect: ["axos.gold.table.updated"]
, etc. If you really want to stick with your
subprocess.run(...)
approach, you could wrap this Python in a standalone file and call
python path/to/create_automation.py
from your script. There isn’t a supported
prefect automation create
CLI in 3.x today, so Python is the recommended route. Would you like me to add existence-check/update logic so the script is idempotent (create if missing, update if present)?