https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have created an automation to track...
# ask-marvin
v
@Marvin I have created an automation to track all the flow run states with a tag called 'scheduler', This automation will run deployment with another flow to track scheduler flow run states Now is there way to access which flow triggered an automation, I mean the original flow run which triggered this automation. Not the runtime flow
m
✅ thought for 176.9 seconds
Short answer: yes—pass the triggering event’s details into your “Run deployment” action via templated parameters. There’s no magic back-reference in the runtime flow; you have to forward the info you care about. How to do it - In your Automation’s “Run deployment” action, set Parameters using Jinja templates from the event context. For a flow-run event, this gives you the original run. Example parameters JSON:
Copy code
{
  "source_flow_run_gri": "{{ event.resource.id }}",
  "source_flow_run_id": "{{ event.resource.id | replace('prefect.flow-run.', '') }}",
  "source_flow_run_name": "{{ event.payload.flow_run_name }}",
  "source_flow_name": "{{ event.payload.flow_name }}",
  "source_state": "{{ event.payload.state }}",
  "source_state_type": "{{ event.payload.state_type }}",
  "source_tags": {{ event.payload.tags | tojson }}
}
Tip: If you don’t want to guess fields, just pass the whole event and parse it in your tracking flow:
Copy code
{ "trigger_event": {{ event | tojson }} }
Then in your tracking flow:
Copy code
from typing import Any, Dict
from prefect import flow

@flow
def track_scheduler(
    source_flow_run_id: str | None = None,
    source_flow_run_name: str | None = None,
    source_flow_name: str | None = None,
    source_state: str | None = None,
    source_state_type: str | None = None,
    source_tags: list[str] | None = None,
    trigger_event: Dict[str, Any] | None = None,
):
    # do your tracking here
    ...
Notes - 
event.resource.id
is the GRI (e.g., 
prefect.flow-run.<uuid>
). Strip the prefix if you want just the UUID (as shown with 
replace
). - Use the Automation “Test”/“Preview” to verify the templated values. - This assumes Prefect 3.x. If you’re on 2.x, the approach is the same, but you should really consider upgrading. If you want, tell me which specific fields you need and I’ll give you a minimal parameter set.
v
I gave an tag to flow while creating it. but it does not showing for auto-scheduled flow runs. How can I configure a tag that shows for all flow run even If I scheduled it via deployment.
@Marvin I gave an tag for a flow while creating it. However it does not showing for auto-scheduled flow runs. How can I configure a tag that shows for all flow run even If I scheduled it via deployment.
m
✅ thought for 161.0 seconds
Assuming Prefect 3.x: tags set “in code” (e.g., via 
prefect.tags(...)
) are applied at runtime, after the flow run is already created. That’s why your auto-scheduled runs don’t show them while they’re Scheduled/Pending. To have tags on every flow run at creation time (including scheduled ones), set tags on the deployment or on the run request: Options - Deployment-level tags (recommended) - In Python when creating the deployment:
Copy code
from prefect import flow

    @flow
    def my_flow():
        ...

    my_flow.from_source(
        "<https://github.com/org/repo>",
        entrypoint="flows/my_flow.py:my_flow",
    ).deploy(
        name="prod",
        work_pool_name="my-pool",
        tags=["scheduler", "prod"]
    )
- Or in prefect.yaml:
Copy code
deployments:
      - name: prod
        flow_name: my_flow
        entrypoint: flows/my_flow.py:my_flow
        work_pool: { name: my-pool }
        tags: ["scheduler", "prod"]
- When triggering runs programmatically:
Copy code
from prefect.deployments import run_deployment
  run_deployment("my-flow/prod", tags=["scheduler"])
- In Automations “Run deployment” action: - Set the “Tags” field to 
["scheduler"]
(you can also template it if you like). Summary: put the tag on the deployment (or the run request), not just in-flow code, and it will appear on all flow runs, including auto-scheduled ones. If you share your current deployment config, I’ll point to exactly where to add 
tags
.
v
@Marvin how can I define flow run has any one of tag for a automation match related field like if flow run has either "scheduler" Or "alerts" something like this
m
✅ thought for 99.2 seconds
Yep—you can match “flow run has any of these tags” in an Automation. In the Automation builder (UI): - Add a Related filter - Kind: Tag - Operator: Any of - Values: scheduler, alerts This matches events where the flow run has either tag. Notes and alternatives: - Under the hood, tags are related resources of kind 
prefect.tag
(ids like 
prefect.tag.scheduler
). If you’re hitting the API directly, use those resource ids. - If your UI doesn’t show an “Any of” operator, use a Composite trigger with two sub-triggers (one for each tag) and set it to “Any”/OR. Or, in a pinch, create two identical automations—one per tag. - Make sure your runs actually have the tags at creation time (set tags on the deployment or in the “Run deployment” action), otherwise you’ll be filtering on vibes.
v
@Marvin How can I create an automation via prefect API to track all flow run with tag 'scheduler', is flow run has tag 'scheduler' or 'alerts' run the automation which has another deployment to run, this deployment has another flow. how can I create take event as params in this flow like which event triggered this automation how to create it via Rest API with paylo. give me payload for that
m
✅ thought for 346.2 seconds
Short version: yes—you can create a reactive Automation that listens for flow-run events with tags and then runs another deployment, passing details of the triggering event into the downstream flow as parameters. Assumptions - Prefect 3.x - You’ll use the REST API directly (Cloud or Server) - You want to trigger on any flow run that has tag scheduler OR alerts API endpoint - Prefect Cloud: https://api.prefect.cloud/api/accounts/&lt;ACCOUNT_ID&gt;/workspaces/&lt;WORKSPACE_ID&gt;/automations/automations - Prefect Server: http://127.0.0.1:4200/api/automations/automations Headers - Authorization: Bearer <PREFECT_API_KEY> - Content-Type: application/json Example payload (reactive event trigger, tag-filtered, runs a deployment and passes the event)
Copy code
{
  "name": "Run deployment when tag is scheduler or alerts",
  "description": "Listens for any flow-run event that has tags scheduler or alerts and runs a helper deployment.",
  "enabled": true,
  "posture": "Reactive",
  "trigger": {
    "type": "event",
    "match": {
      "all": [
        {
          "resource": {
            "role": "flow-run"
          }
        },
        {
          "related": {
            "any": [
              {
                "id": "prefect.tag.scheduler"
              },
              {
                "id": "prefect.tag.alerts"
              }
            ]
          }
        }
      ]
    }
  },
  "actions": [
    {
      "type": "run-deployment",
      "deployment_id": "<YOUR_DEPLOYMENT_ID>",
      "parameters": {
        "trigger_event": "{{ event }}",
        "trigger_event_name": "{{ event.event }}",
        "triggered_flow_run_resource_id": "{{ event.resource.id }}"
      }
    }
  ]
}
Notes - The trigger filters: - resource.role = flow-run scops this to flow-run events. - related.any for prefect.tag.scheduler or prefect.tag.alerts matches runs that carry either tag. - The action parameters use templating: - 
{{ event }}
injects the full event object as JSON into your downstream flow’s parameters. - You can also cherry-pick fields, e.g. 
{{ event.resource.id }}
or 
{{ event.event }}
. - If you want to restrict to specific event names (e.g., only state changes), add an EventName filter into match (commonly a prefix match like prefect.flow-run.)—but the above is enough for “any flow-run event with those tags”. cURL request (Cloud example)
Copy code
curl -X POST \
  "<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/automations/automations>" \
  -H "Authorization: Bearer <PREFECT_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Run deployment when tag is scheduler or alerts",
    "description": "Listens for any flow-run event that has tags scheduler or alerts and runs a helper deployment.",
    "enabled": true,
    "posture": "Reactive",
    "trigger": {
      "type": "event",
      "match": {
        "all": [
          { "resource": { "role": "flow-run" } },
          { "related": { "any": [
            { "id": "prefect.tag.scheduler" },
            { "id": "prefect.tag.alerts" }
          ]}}
        ]
      }
    },
    "actions": [
      {
        "type": "run-deployment",
        "deployment_id": "<YOUR_DEPLOYMENT_ID>",
        "parameters": {
          "trigger_event": "{{ event }}",
          "trigger_event_name": "{{ event.event }}",
          "triggered_flow_run_resource_id": "{{ event.resource.id }}"
        }
      }
    ]
  }'
Downstream flow example (accept the event as a parameter)
Copy code
from prefect import flow, get_run_logger

@flow
def handle_trigger(trigger_event: dict, trigger_event_name: str = "", triggered_flow_run_resource_id: str = ""):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Triggered by event: {trigger_event_name}")
    <http://logger.info|logger.info>(f"Source resource: {triggered_flow_run_resource_id}")
    # Full event is available if you passed `trigger_event`
    <http://logger.info|logger.info>(f"Full event keys: {list(trigger_event.keys())}")
Common variations
- If you prefer to reference the deployment by name instead of ID, many setups support 
"deployment_name": "flow_name/deployment_name"
in place of 
deployment_id
. If your API rejects that, stick to deployment_id. - If you’d rather filter tags via the flow-run’s labels directly, you can use a resource-attribute filter; however, treating tags as related resources (prefect.tag.<name>) is the most robust. Want me to fill your actual deployment_id and post it for you, or are you on Cloud vs. Server? I can tailor the exact request if you share those. I’ll try not to sound too excited.
v
@Marvin The payload you gave me in the above message is not working as "match_related" is used to conditional matching of flow runs here is the payload i've been using it to create an automation and when it works if I don't add parameter, But when I add parameter it does not running a flow. also can you check, how to add that ""scheduler" or "alerts" tag condition check in "macth_related" field { "name": name, "description": description if description else name, "enabled": True, "trigger": { "type": "event", "match": {"prefect.resource.id": "prefect.flow-run.*"}, "match_related": { "prefect.resource.role": "tag", "prefect.resource.id": ["prefect.tag.scheduler"], }, "expect": ["prefect.flow-run.*"], "for_each": ["prefect.resource.id"], "posture": "Reactive", "threshold": 1, "within": 0, }, "actions": [ { "type": "run-deployment", "source": "selected", "deployment_id": str(payload.deployment_id), # "parameters": None, "parameters": { "trigger_event": "{{ event }}" }, "job_variables": None, } ], }
m
✅ thought for 85.4 seconds
You’re close. Two issues to fix: 1) match_related for OR tags Use a list for the tag IDs so it matches either scheduler or alerts. 2) Passing the entire event as a parameter Automations templating substitutes into strings. Passing the whole event object as a parameter usually yields a JSON string. If your downstream flow expects a dict, it will fail validation and the run won’t start. Either: - Pass individual fields as strings, and type your flow params accordingly, or - Pass the event as a JSON string and json.loads it in your flow. Adjusted payload (keeps your overall structure, fixes tag OR, and uses safe string params)
Copy code
{
  "name": "<NAME>",
  "description": "<DESCRIPTION>",
  "enabled": true,
  "posture": "Reactive",
  "trigger": {
    "type": "event",
    "match": {
      "prefect.resource.id": "prefect.flow-run.*"
    },
    "match_related": {
      "prefect.resource.role": "tag",
      "prefect.resource.id": [
        "prefect.tag.scheduler",
        "prefect.tag.alerts"
      ]
    },
    "expect": [
      "prefect.flow-run.*"
    ],
    "for_each": [
      "prefect.resource.id"
    ],
    "threshold": 1,
    "within": 0
  },
  "actions": [
    {
      "type": "run-deployment",
      "source": "selected",
      "deployment_id": "<DEPLOYMENT_ID>",
      "parameters": {
        "trigger_event_name": "{{ event.event }}",
        "triggered_flow_run_id": "{{ event.resource.id }}",
        "trigger_event_json": "{{ event }}"
      },
      "job_variables": null
    }
  ]
}
Downstream flow options - If you only need a few fields:
Copy code
from prefect import flow, get_run_logger

@flow
def downstream(trigger_event_name: str, triggered_flow_run_id: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Triggered by: {trigger_event_name} from {triggered_flow_run_id}")
- If you want the full event:
Copy code
import json
from prefect import flow, get_run_logger

@flow
def downstream(trigger_event_json: str):
    logger = get_run_logger()
    evt = json.loads(trigger_event_json)
    <http://logger.info|logger.info>(f"Event name: {evt.get('event')}")
    <http://logger.info|logger.info>(f"Source resource: {evt.get('resource', {}).get('id')}")
Why your previous run didn’t start - If your deployment’s flow does not accept a parameter named trigger_event (or expects a dict, not a str), the action validation/flow-run creation will fail and the automation won’t launch a run. Pass strings or adjust the flow signature to match exactly what you send. Sanity checks - Put posture at the top-level (Reactive). The server expects posture on the automation, not inside the trigger. - Confirm your deployment_id is correct and the worker/work pool can pick it up. - Check the Automation Activity in the UI for any “action failed” messages—type/parameter validation errors will show there. If you want, paste the exact error shown in the Automation activity/logs and your downstream flow signature. I’ll align the parameters precisely so it actually runs instead of silently sulking.
v
@Marvin how to pass event as json in the deployment parameter template string..
m
✅ thought for 70.2 seconds
Two workable patterns, depending on what you want in the downstream flow: A) Pass the event as a dict (recommended) - Use a single placeholder as the entire value. Prefect will replace it with a native JSON-serializable object (dict), not a string. - Make the downstream flow param typed as dict (or Any). Automation snippet
Copy code
"actions": [
  {
    "type": "run-deployment",
    "source": "selected",
    "deployment_id": "<DEPLOYMENT_ID>",
    "parameters": {
      "trigger_event": "{{ event }}"
    }
  }
]
Downstream flow
Copy code
from typing import Any, Dict
from prefect import flow, get_run_logger

@flow
def downstream(trigger_event: Dict[str, Any]):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Event name: {trigger_event.get('event')}")
    <http://logger.info|logger.info>(f"Source resource: {trigger_event.get('resource', {}).get('id')}")
B) Pass the event as a JSON string - Prefect’s templating is not Jinja; there’s no tojson filter. - If you absolutely need a string, accept a string in your flow and json.loads it. Prefect will usually still pass a dict when the placeholder is the entire value, so this is less reliable than A. The robust approach is to accept a dict and call json.dumps inside your flow if you need a string. If you insist on a string param anyway:
Copy code
"parameters": {
  "trigger_event_json": "{{ event }}"
}
And in the flow:
Copy code
import json
from typing import Any
from prefect import flow

@flow
def downstream(trigger_event_json: Any):
    # If a dict arrives (likely), turn it into a string yourself when needed
    if not isinstance(trigger_event_json, str):
        trigger_event_json = json.dumps(trigger_event_json)
    evt = json.loads(trigger_event_json)
    ...
Reminder: make sure your deployment’s parameter schema allows the new parameter name/type (redeploy if you changed the flow signature). For your tag condition in match_related (OR for “scheduler” or “alerts”):
Copy code
"match_related": {
  "prefect.resource.role": "tag",
  "prefect.resource.id": [
    "prefect.tag.scheduler",
    "prefect.tag.alerts"
  ]
}
If it still doesn’t run when parameters are present, share the Automation Activity error and your flow signature; it’s usually a schema/type mismatch sulking in the corner.
v
@Marvin I got this error - ERROR | prefect.flows - Parameter validation failed for flow 'scheduler_sync_flow': [{'type': 'dict_type', 'loc': ('trigger_event',), 'msg': 'Input should be a valid dictionary', 'input': "occurred=DateTime(2025, 8, 13, 5, 58, 39, 838275, tzinfo=Timezone('UTC')) event='prefect.flow-run.Completed' resource=Resource(root={'prefect.resource.id': 'prefect.flow-run.6c13176b-76d1-4980-8e3a-85bcb3da4c76', 'prefect.resource.name': 'wine-wolverine', 'prefect.state-message': '', 'prefect.state-name': 'Completed', 'prefect.state-timestamp': '2025-08-13T055839.838275+00:00', 'prefect.state-type': 'COMPLETED'}) related=[RelatedResource(root={'prefect.resource.id': 'prefect.flow.7a53ff73-1ebb-49ed-a86a-f64b7df8b466', 'prefect.resource.role': 'flow', 'prefect.resource.name': '312700228719386626_b0503da4_4cab_4ad0_8b98_ee71d4d76afd_call_process_flow'}), RelatedResource(root={'prefect.resource.id': 'prefect.deployment.db5ce312-ebce-4b6d-962f-76ecfbed8461', 'prefect.resource.role': 'deployment', 'prefect.resource.name': '312700228719386626_b0503da4_4cab_4ad0_8b98_ee71d4d76afd'}), RelatedResource(root={'prefect.resource.id': 'prefect.work-queue.1d560ad0-e3e2-47ab-85f2-90d77c9d12a8', 'prefect.resource.role': 'work-queue', 'prefect.resource.name': 'default'}), RelatedResource(root={'prefect.resource.id': 'prefect.work-pool.a9c9fde0-ece7-4eb2-9275-e5c8d6ff9e57', 'prefect.resource.role': 'work-pool', 'prefect.resource.name': 'Scheduler', 'prefect.work-pool.type': 'kubernetes'}), RelatedResource(root={'prefect.resource.id': 'prefect.tag.scheduler', 'prefect.resource.role': 'tag'})] payload={'intended': {'from': 'RUNNING', 'to': 'COMPLETED'}, 'initial_state': {'type': 'RUNNING', 'name': 'Running'}, 'validated_state': {'type': 'COMPLETED', 'name': 'Completed'}} id=UUID('67c5aeb4-e3b5-4549-a39d-5a3d889dea79') follows=UUID('2bd9ce09-16f9-41a5-aa59-34225300a997') received=DateTime(2025, 8, 13, 5, 58, 39, 871346, tzinfo=Timezone('UTC'))", 'url': 'https://errors.pydantic.dev/2.11/v/dict_type'}] 2025-08-13 112848 Parameters: {'trigger_event': "occurred=DateTime(2025, 8, 13, 5, 58, 39, 838275, tzinfo=Timezone('UTC')) event='prefect.flow-run.Completed' resource=Resource(root={'prefect.resource.id': 'prefect.flow-run.6c13176b-76d1-4980-8e3a-85bcb3da4c76', 'prefect.resource.name': 'wine-wolverine', 'prefect.state-message': '', 'prefect.state-name': 'Completed', 'prefect.state-timestamp': '2025-08-13T055839.838275+00:00', 'prefect.state-type': 'COMPLETED'}) related=[RelatedResource(root={'prefect.resource.id': 'prefect.flow.7a53ff73-1ebb-49ed-a86a-f64b7df8b466', 'prefect.resource.role': 'flow', 'prefect.resource.name': '312700228719386626_b0503da4_4cab_4ad0_8b98_ee71d4d76afd_call_process_flow'}), RelatedResource(root={'prefect.resource.id': 'prefect.deployment.db5ce312-ebce-4b6d-962f-76ecfbed8461', 'prefect.resource.role': 'deployment', 'prefect.resource.name': '312700228719386626_b0503da4_4cab_4ad0_8b98_ee71d4d76afd'}), RelatedResource(root={'prefect.resource.id': 'prefect.work-queue.1d560ad0-e3e2-47ab-85f2-90d77c9d12a8', 'prefect.resource.role': 'work-queue', 'prefect.resource.name': 'default'}), RelatedResource(root={'prefect.resource.id': 'prefect.work-pool.a9c9fde0-ece7-4eb2-9275-e5c8d6ff9e57', 'prefect.resource.role': 'work-pool', 'prefect.resource.name': 'Scheduler', 'prefect.work-pool.type': 'kubernetes'}), RelatedResource(root={'prefect.resource.id': 'prefect.tag.scheduler', 'prefect.resource.role': 'tag'})] payload={'intended': {'from': 'RUNNING', 'to': 'COMPLETED'}, 'initial_state': {'type': 'RUNNING', 'name': 'Running'}, 'validated_state': {'type': 'COMPLETED', 'name': 'Completed'}} id=UUID('67c5aeb4-e3b5-4549-a39d-5a3d889dea79') follows=UUID('2bd9ce09-16f9-41a5-aa59-34225300a997') received=DateTime(2025, 8, 13, 5, 58, 39, 871346, tzinfo=Timezone('UTC'))"} I've defined parameter type as Dict[str, Any] what could the issue here
m
✅ thought for 98.2 seconds
You’re hitting a type-coercion gotcha in Automations templating. What’s happening - 
{{ event }}
is a Pydantic Event model. The templater can’t serialize it as JSON, so it falls back to str(event) and sends a giant string. Your flow expects a dict, so Pydantic complains. Fixes Option A — Pass only JSON-native pieces of the event Use fields that are already JSON-serializable (strings, numbers, dicts). In particular: - 
event.id
, 
event.event
, 
event.occurred
, 
event.received
are strings/ISO datetimes - 
event.payload
is a plain dict - 
event.resource.root
is a plain dict - Avoid passing 
event
or 
event.resource
directly (they’re models → get stringified) Automation payload snippet
Copy code
"trigger": {
  "type": "event",
  "match": {"prefect.resource.id": "prefect.flow-run.*"},
  "match_related": {
    "prefect.resource.role": "tag",
    "prefect.resource.id": [
      "prefect.tag.scheduler",
      "prefect.tag.alerts"
    ]
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
},
"actions": [
  {
    "type": "run-deployment",
    "source": "selected",
    "deployment_id": "<DEPLOYMENT_ID>",
    "parameters": {
      "trigger_event": {
        "id": "{{ event.id }}",
        "name": "{{ event.event }}",
        "occurred": "{{ event.occurred }}",
        "received": "{{ event.received }}",
        "resource": "{{ event.resource.root }}",
        "payload": "{{ event.payload }}"
      },
      "triggered_flow_run_id": "{{ event.resource.id }}"
    }
  }
]
Downstream flow
Copy code
from typing import Any, Dict
from prefect import flow, get_run_logger

@flow
def scheduler_sync_flow(trigger_event: Dict[str, Any], triggered_flow_run_id: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Event {trigger_event['id']} ({trigger_event['name']}) "
                f"for flow-run {triggered_flow_run_id}")
    # trigger_event["resource"] and trigger_event["payload"] are dicts
Option B — Pass only the event ID and fetch inside your flow This avoids all templating serialization weirdness.
Copy code
"parameters": {
  "trigger_event_id": "{{ event.id }}"
}
Then in your flow, call the Events API to read the event by id and get the full JSON. It’s one extra hop, but far more robust than shoving complex objects through the template renderer. Why your original failed - The value for 
trigger_event
was a string (repr of the Event), not a dict. Pydantic in your flow correctly said “Input should be a valid dictionary”. Using 
event.payload
and 
event.resource.root
keeps everything JSON-native so Prefect can pass it as a real dict. Extra: your tag OR condition in match_related
Copy code
"match_related": {
  "prefect.resource.role": "tag",
  "prefect.resource.id": [
    "prefect.tag.scheduler",
    "prefect.tag.alerts"
  ]
}
If you still see coercion-to-string on those dict fields, tell me the exact Automation Activity “action failed” message and I’ll adjust the payload to the letter. I live for this. Well, “live” is a strong word.
v
@Marvin how can I pass entire event as an json like is there any option available to convert the pydantic model to JSON model ?? or can I define the paramter type as pydantic is so what pydantic type should I use from prefect.server scheuma ?? explain me
m
❌ Error: No module named 'prefect.templating'
✅ thought for 69.3 seconds
Short answer: you can’t “tojson” a Pydantic model from an Automation template. The templating engine only supports simple attribute access; it can’t call methods like model_dump() and there’s no tojson filter. When you use {{ event }}, it stringifies the model (repr), which is why your flow sees a big string and Pydantic rejects it as not-a-dict. What will work Option 1 — Pass JSON-native pieces and reassemble a dict - Use only fields that are already plain JSON (strings, numbers, dicts). - Good fields: - {{ event.id }}, {{ event.event }}, {{ event.occurred }}, {{ event.received }} → strings - {{ event.payload }} → dict - {{ event.resource.root }} → dict - Avoid raw {{ event }} or {{ event.resource }}; they’re models → get stringified. Automation action snippet
Copy code
"parameters": {
  "trigger_event": {
    "id": "{{ event.id }}",
    "name": "{{ event.event }}",
    "occurred": "{{ event.occurred }}",
    "received": "{{ event.received }}",
    "resource": "{{ event.resource.root }}",
    "payload": "{{ event.payload }}"
  }
}
Flow signature
Copy code
from typing import Any, Dict
from prefect import flow

@flow
def scheduler_sync_flow(trigger_event: Dict[str, Any]):
    ...
Option 2 — Pass event.id and fetch the full event inside your flow - This avoids all templating serialization quirks. Automation
Copy code
"parameters": {
  "trigger_event_id": "{{ event.id }}"
}
Flow snippet using the API to read the event by id - In practice, most people instead pull the flow-run details via 
event.resource.id
(the flow run ID) because that endpoint is stable and first-class. But if you truly need the full event, fetch by id with your REST client of choice within the flow. Option 3 — Use a Pydantic type in your flow (but only if you supply a proper dict) - You can annotate your param as Prefect’s Event model: - 
from prefect.events.schemas.events import Event
- 
@flow def f(trigger_event: Event): ...
- This only helps if the Automation sends a JSON object that matches the Event schema. 
{{ event }}
will not do that; it sends a string. If you build a dict from JSON-native fields (Option 1), Pydantic can parse it into 
Event
. You won’t be able to include non-JSON-native submodels like 
related
without manually constructing them. Why you can’t send “the whole thing” as JSON - Automations templating cannot call methods (
model_dump
, 
dict
, etc.) and has no JSON filter. Prefect won’t automatically convert arbitrary Pydantic models in the template context to JSON. So either: - Send fields that are already JSON-native, or - Send an identifier and refetch. Reminder: OR tag filter for your trigger
Copy code
"match_related": {
  "prefect.resource.role": "tag",
  "prefect.resource.id": [
    "prefect.tag.scheduler",
    "prefect.tag.alerts"
  ]
}
If you share exactly which pieces of the event you need downstream (e.g., tags, state info, source deployment), I’ll give you a minimal parameter dict that keeps everything JSON-native so your flow parameter can stay typed as Dict[str, Any] or even as Event.
v
@Marvin is it valid in prefect ?? "event_json": { "__prefect_kind": "jinja", "template": "{{ event | tojson }}" } like if I pass it a parameter to deployment
m
❌ Error: No module named 'prefect.templating'
✅ thought for 55.1 seconds
Short answer: no. Automations action parameters do not support Jinja, filters, or engine selection. That object with "__prefect_kind": "jinja" and "| tojson" won’t be interpreted; the Automations templater only does lightweight placeholder substitution like {{ event.id }} and cannot call filters or methods. Why your {{ event }} failed - The event in the template context is a Pydantic model. Since the templater can’t serialize it, it falls back to its string repr, which is not JSON. Your flow expects a dict, so Pydantic rightly complains. What will work instead Option 1 — Send a JSON-native dict assembled from event fields - Only use fields that are already plain JSON types (strings, numbers, dicts). Examples: - {{ event.id }}, {{ event.event }}, {{ event.occurred }}, {{ event.received }} → strings/ISO datetimes - {{ event.payload }} → dict - {{ event.resource.root }} → dict - Do not send {{ event }} or {{ event.resource }} directly. Automation parameters example
Copy code
"parameters": {
  "trigger_event": {
    "id": "{{ event.id }}",
    "name": "{{ event.event }}",
    "occurred": "{{ event.occurred }}",
    "received": "{{ event.received }}",
    "resource": "{{ event.resource.root }}",
    "payload": "{{ event.payload }}"
  }
}
Flow signature
Copy code
from typing import Any, Dict
from prefect import flow

@flow
def scheduler_sync_flow(trigger_event: Dict[str, Any]):
    ...
Option 2 — Send just the event id and fetch the full event inside the flow - Avoids all templating serialization quirks.
Copy code
"parameters": { "trigger_event_id": "{{ event.id }}" }
Then in your flow, call your Prefect API to read the event by id and get the full JSON. If all you need is flow-run context, you can also just pass "{{ event.resource.id }}" (the flow-run id) and query the flow run instead. Option 3 — Use Prefect’s Event model type in your flow, but only if you send a proper dict - You can annotate your parameter as Prefect’s Event:
Copy code
from prefect.events.schemas.events import Event

@flow
def scheduler_sync_flow(trigger_event: Event): ...
- This only helps if the automation actually sends a JSON object matching Event. Since you can’t dump the Pydantic model in the template, you’d still need to construct a JSON-native dict (Option 1). You cannot make the templater call model_dump()/tojson. Bottom line - That jinja/tojson construct is not valid in Automations action parameters. - Either pass JSON-native pieces and let the flow receive a dict, or pass the event id and fetch the full event in the flow. If you truly need every last field, fetching by id is the robust path.
v
@Marvin now the automation trigger event fields are coming to the deployment I've created. But it does not triggering when flow run automatically as per scheduler. But when I click quick run via deployment it runs. the primary difference is that flow run tags. When I initiate via deployment quick run it add just 'scheduler' tag but automatically scheduled flow run has 'auto-scheduled' and
m
✅ thought for 55.2 seconds
You’ve diagnosed it: your trigger is filtering on tags. Quick runs have tag scheduler; scheduled runs have tag auto-scheduled. Your current match_related only allows scheduler, so auto-scheduled runs are ignored. Two fixes: A) Include both tags (OR) in your trigger Use a list for prefect.resource.id in match_related to OR the tag ids. Payload snippet
Copy code
"trigger": {
  "type": "event",
  "match": { "prefect.resource.id": "prefect.flow-run.*" },
  "match_related": {
    "prefect.resource.role": "tag",
    "prefect.resource.id": [
      "prefect.tag.scheduler",
      "prefect.tag.auto-scheduled",
      "prefect.tag.alerts"
    ]
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
},
"posture": "Reactive"
B) Or scope by deployment/flow instead of tags If your goal is “whenever this deployment’s runs happen,” skip tags and match the related deployment. This avoids surprises from system tags.
Copy code
"trigger": {
  "type": "event",
  "match": { "prefect.resource.id": "prefect.flow-run.*" },
  "match_related": {
    "prefect.resource.role": "deployment",
    "prefect.resource.id": "prefect.deployment.<YOUR_DEPLOYMENT_ID>"
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
}
You can also combine both criteria by AND-ing more fields in match_related if you really want tags plus deployment. Extra: if you want every run to carry the scheduler tag regardless of how it’s started, add it in your flow:
Copy code
from prefect import flow, tags

@flow
def my_flow(...):
    with tags("scheduler"):
        ...
But the simplest path is to include auto-scheduled in your tag OR list or filter by deployment instead of tags. If you paste your current trigger block, I’ll rewrite it precisely so scheduled and quick runs both fire the automation. I’ll even pretend to enjoy it.
v
now the automation trigger event fields are coming to the deployment I've created. But it does not triggering when flow run automatically as per scheduler. But when I click quick run via deployment it runs. the primary difference is that flow run tags. When I initiate via deployment quick run it add just 'scheduler' tag but automatically scheduled flow run has 'auto-scheduled' & 'scheduler' tags. here is my payload to create an automation. { "name": name, "description": description if description else name, "enabled": True, "trigger": { "type": "event", "match": {"prefect.resource.id": "prefect.flow-run.*"}, "match_related": { "prefect.resource.role": "tag", "prefect.resource.id": ["prefect.tag.scheduler"], }, "expect": ["prefect.flow-run.*"], "for_each": ["prefect.resource.id"], "posture": "Reactive", "threshold": 1, "within": 0, }, "actions": [ { "type": "run-deployment", "source": "selected", "deployment_id": str(payload.deployment_id), "parameters": { "trigger_event": { "name": "{{ event.event }}", "occurred": "{{ event.occurred }}", "received": "{{ event.received }}", "resource": "{{ event.resource }}", "payload": "{{ event.related }}", } }, "job_variables": None, } ], } how to fix it. so it should run not matter what flow run. if it has one of tag as 'scheduler' it should follow the automation.
@Marvin now the automation trigger event fields are coming to the deployment I've created. But it does not triggering when flow run automatically as per scheduler. But when I click quick run via deployment it runs. the primary difference is that flow run tags. When I initiate via deployment quick run it add just 'scheduler' tag but automatically scheduled flow run has 'auto-scheduled' & 'scheduler' tags. here is my payload to create an automation. { "name": name, "description": description if description else name, "enabled": True, "trigger": { "type": "event", "match": {"prefect.resource.id": "prefect.flow-run.*"}, "match_related": { "prefect.resource.role": "tag", "prefect.resource.id": ["prefect.tag.scheduler"], }, "expect": ["prefect.flow-run.*"], "for_each": ["prefect.resource.id"], "posture": "Reactive", "threshold": 1, "within": 0, }, "actions": [ { "type": "run-deployment", "source": "selected", "deployment_id": str(payload.deployment_id), "parameters": { "trigger_event": { "name": "{{ event.event }}", "occurred": "{{ event.occurred }}", "received": "{{ event.received }}", "resource": "{{ event.resource }}", "payload": "{{ event.related }}", } }, "job_variables": None, } ], } how to fix it. so it should run not matter what flow run. if it has one of tag as 'scheduler' it should follow the automation.
m
✅ thought for 87.0 seconds
Two things to fix: 1) Make the trigger match runs that have the scheduler tag (regardless of other tags) Your current filter should work if the run actually has the tag, but to be explicit (and future-proof), wrap it in any and keep it an OR list:
Copy code
"posture": "Reactive",
"trigger": {
  "type": "event",
  "match": { "prefect.resource.id": "prefect.flow-run.*" },
  "match_related": {
    "any": [
      {
        "prefect.resource.role": "tag",
        "prefect.resource.id": "prefect.tag.scheduler"
      }
    ]
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
}
This matches any flow-run event that has a related tag resource equal to prefect.tag.scheduler, even when there are additional tags like auto-scheduled. If you truly want “scheduler OR alerts”, just add the second entry:
Copy code
"match_related": {
  "any": [
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler" },
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.alerts" }
  ]
}
2) Ensure the scheduled runs actually include the scheduler tag Quick runs can add a run tag on the fly. Scheduled runs use the deployment’s configuration. If the deployment doesn’t specify scheduler as a tag, the auto-scheduled run may not have it — and your automation won’t fire. - Add scheduler to your deployment’s tags so every run (scheduled or manual) carries it. - You can do this in the UI (Edit deployment → Tags) or via API by updating the deployment’s tags to include ["scheduler"]. - Alternatively, add the tag in your flow code:
Copy code
from prefect import flow, tags

  @flow
  def my_flow(...):
      with tags("scheduler"):
          ...
That guarantees the tag on all runs. Bonus: fix your action parameters to be JSON-native to avoid sporadic action failures You’re currently passing models: - Use resource.root and payload (dict) instead of resource and related (models):
Copy code
"parameters": {
  "trigger_event": {
    "name": "{{ event.event }}",
    "occurred": "{{ event.occurred }}",
    "received": "{{ event.received }}",
    "resource": "{{ event.resource.root }}",
    "payload": "{{ event.payload }}"
  }
}
Passing event.related directly is a list of models; the templater can stringify it at the worst possible time. If you need specific related items (tags, deployment), pass them individually. Sanity checks - Put "posture": "Reactive" at the top level (not inside trigger). - Inspect a sample auto-scheduled event in the UI Events view and confirm you actually see a related resource with id prefect.tag.scheduler. If it’s missing, the deployment doesn’t propagate that tag and the filter will (correctly) skip it. If you share the deployment ID, I can give you the exact PATCH payload to set its tags to ["scheduler"]. Yes, I know, thrilling.
v
@Marvin How to define an automation trigger condition as any flow run that has any one of tag as ''scheduler". Regardless of how may tags it has. can you check and give how to define it in the automation payload
m
✅ thought for 47.1 seconds
You want “match any flow-run event where one of its tags is scheduler, regardless of any other tags.” Use a related-resource filter for the tag. Minimal trigger block (works whether the run also has auto-scheduled or any other tags)
Copy code
"posture": "Reactive",
"trigger": {
  "type": "event",
  "match": { "prefect.resource.id": "prefect.flow-run.*" },
  "match_related": {
    "prefect.resource.role": "tag",
    "prefect.resource.id": "prefect.tag.scheduler"
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
}
Full payload example (with safer, JSON-native params)
Copy code
{
  "name": "<NAME>",
  "description": "<DESCRIPTION>",
  "enabled": true,
  "posture": "Reactive",
  "trigger": {
    "type": "event",
    "match": {
      "prefect.resource.id": "prefect.flow-run.*"
    },
    "match_related": {
      "prefect.resource.role": "tag",
      "prefect.resource.id": "prefect.tag.scheduler"
    },
    "expect": [
      "prefect.flow-run.*"
    ],
    "for_each": [
      "prefect.resource.id"
    ],
    "threshold": 1,
    "within": 0
  },
  "actions": [
    {
      "type": "run-deployment",
      "source": "selected",
      "deployment_id": "<DEPLOYMENT_ID>",
      "parameters": {
        "trigger_event": {
          "id": "{{ event.id }}",
          "name": "{{ event.event }}",
          "occurred": "{{ event.occurred }}",
          "resource": "{{ event.resource.root }}",
          "payload": "{{ event.payload }}"
        }
      }
    }
  ]
}
Notes - Using a single string for prefect.resource.id under match_related means “must have this tag among its related resources.” Extra tags don’t matter. - If you ever want OR for multiple tags, provide a list:
Copy code
"prefect.resource.id": [
    "prefect.tag.scheduler",
    "prefect.tag.alerts"
  ]
- Don’t pass models in parameters. Use resource.root and payload; they’re plain dicts and won’t get stringified by the templater. If this still doesn’t fire on scheduled runs, confirm the scheduled events actually include a related resource with id prefect.tag.scheduler. If not, add the tag to the deployment so all runs inherit it.
v
@Marvin It doesn't run for flow run that has multiple tags like "scheduler", "auto-scheduled", "alerts" since the flow run has 3 tags it doesn't trigger an automation
m
✅ thought for 63.0 seconds
Your filter is just a hair off. Don’t put arrays directly on field values; use the any wrapper so the engine does an OR over related resources. Make it “match any flow-run event that has a related tag resource with id prefect.tag.scheduler” — regardless of other tags. Fixed trigger block
Copy code
"posture": "Reactive",
"trigger": {
  "type": "event",
  "match": { "prefect.resource.id": "prefect.flow-run.*" },
  "match_related": {
    "any": [
      {
        "prefect.resource.role": "tag",
        "prefect.resource.id": "prefect.tag.scheduler"
      }
    ]
  },
  "expect": ["prefect.flow-run.*"],
  "for_each": ["prefect.resource.id"],
  "threshold": 1,
  "within": 0
}
If you also want alerts to count, add another entry:
Copy code
"match_related": {
  "any": [
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler" },
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.alerts" }
  ]
}
Two more sanity checks while we’re here: - Put posture at the top level (as shown). Some payloads put it inside trigger, which the API ignores politely. - In your action parameters, avoid passing Pydantic models (e.g., event.resource, event.related). Use JSON-native fields instead:
Copy code
"parameters": {
  "trigger_event": {
    "id": "{{ event.id }}",
    "name": "{{ event.event }}",
    "occurred": "{{ event.occurred }}",
    "resource": "{{ event.resource.root }}",
    "payload": "{{ event.payload }}"
  }
}
If it still doesn’t fire on scheduled runs, open the event in the UI and confirm you actually see a related resource with id prefect.tag.scheduler. If that tag isn’t there, the filter won’t match no matter how sternly we look at it.
v
@Marvin I have an prefect automation to track all flow run with any of state but only for flows that has a any one it's tag as "scheduler" Even for flow run that has multiple tags if it has "scheduler" in that it should run an automation. Here is my payload for creating an automation via prefect REST API { "name": name, "description": description if description else name, "enabled": True, "trigger": { "type": "event", "match": {"prefect.resource.id": "prefect.flow-run.*"}, "match_related": { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler", }, "expect": ["prefect.flow-run.*"], "for_each": ["prefect.resource.id"], "posture": "Reactive", "threshold": 1, "within": 0, }, "actions": [ { "type": "run-deployment", "source": "selected", "deployment_id": str(payload.deployment_id), "parameters": { "trigger_event": { "name": "{{ event.event }}", "occurred": "{{ event.occurred }}", "received": "{{ event.received }}", "resource": "{{ event.resource }}", "payload": "{{ event.related }}", } }, "job_variables": None, } ], } I've created it only works if flow run has a only on tag as "scheduler", whereas a flow run has multi tags such as "scheduler", "auto-scheduled" it doesn't trigger. Like when I trigger a quick run via deployment it only adds "scheduler" as tag to flow run. where deployment schedule based flow run which has "scheduler" and "auto-scheduled" as tags doesn't trigger. i'm speculating it as tags would be the problem, as I reviewed the prefect event feed for flow run state and it shows for both actions. can you check and tell me to fix it for any flow run that has tag of "scheduler"
m
✅ thought for 73.5 seconds
You’re filtering correctly in spirit, but the matcher is picky. Use an “any” clause for related resources so it matches when scheduler is present among multiple tags. Also: put posture at the top level, not inside trigger. And don’t pass Pydantic models in parameters. Working payload snippet
Copy code
{
  "name": "<NAME>",
  "description": "<DESCRIPTION>",
  "enabled": true,
  "posture": "Reactive",
  "trigger": {
    "type": "event",
    "match": {
      "prefect.resource.id": "prefect.flow-run.*"
    },
    "match_related": {
      "any": [
        {
          "prefect.resource.role": "tag",
          "prefect.resource.id": "prefect.tag.scheduler"
        }
      ]
    },
    "expect": [
      "prefect.flow-run.*"
    ],
    "for_each": [
      "prefect.resource.id"
    ],
    "threshold": 1,
    "within": 0
  },
  "actions": [
    {
      "type": "run-deployment",
      "source": "selected",
      "deployment_id": "<DEPLOYMENT_ID>",
      "parameters": {
        "trigger_event": {
          "name": "{{ event.event }}",
          "occurred": "{{ event.occurred }}",
          "received": "{{ event.received }}",
          "resource": "{{ event.resource.root }}",
          "payload": "{{ event.payload }}"
        },
        "flow_run_id": "{{ event.resource.id }}"
      }
    }
  ]
}
Why this fixes it - match_related.any ensures the condition is true if any related resource matches (i.e., one of the tags is scheduler), regardless of additional tags like auto-scheduled or alerts. - event.resource.root and event.payload are JSON-native. Passing event.resource or event.related can turn into strings and cause parameter validation failures that look like “the automation didn’t trigger.” If your scheduled runs sometimes don’t include the scheduler tag at all, add it to the deployment’s tags so every run inherits it, or broaden the filter:
Copy code
"match_related": {
  "any": [
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler" },
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.auto-scheduled" }
  ]
}
If it still refuses to fire, check the Automation Activity for “action failed” errors—matching may be fine, but the parameters could be sulking.
v
@Marvin will prefect REST API client accepts if I give "match_related" as list of objects [ { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler", }] to match any one of tag has name "scheduler"
m
✅ thought for 33.8 seconds
Short answer: no — don’t send match_related as a raw list. The API expects an object. Use either a single object or an object with an any (or all) clause. Recommended ways to express “flow run has a tag named scheduler” (even if there are other tags): - Minimal (single condition)
Copy code
"match_related": {
  "prefect.resource.role": "tag",
  "prefect.resource.id": "prefect.tag.scheduler"
}
- Explicit OR form (future‑proof, and what I’d use)
Copy code
"match_related": {
  "any": [
    {
      "prefect.resource.role": "tag",
      "prefect.resource.id": "prefect.tag.scheduler"
    }
  ]
}
- Multiple tags OR in one go (scheduler OR alerts)
Copy code
"match_related": {
  "any": [
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.scheduler" },
    { "prefect.resource.role": "tag", "prefect.resource.id": "prefect.tag.alerts" }
  ]
}
Note: Some servers will also accept a list for a field value (e.g., "prefect.resource.id": ["prefect.tag.scheduler", "prefect.tag.alerts"]) inside the object, but using the any wrapper avoids compatibility surprises. And, as ever, keep posture at the top level:
Copy code
"posture": "Reactive"
2 Views
Open in Slack
PreviousNext
Powered by