Mehdi Lallouache
04/02/2026, 8:32 AMMarvin
04/02/2026, 8:32 AMMarvin
04/02/2026, 8:37 AMprefect.flow-run.Completed
- Match related resource → Role: Deployment → pick “my-flow - mode-b”
4. Action:
- Type: Run a deployment
- Source: Selected
- Deployment: “my-flow - mode-a”
- Parameters (optional): if you didn’t set defaults on deployment A, add params_a here
5. Save and enable
Python example
from datetime import timedelta
from uuid import UUID
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture, ResourceSpecification
from prefect.events.actions import RunDeployment
# IDs of the two deployments (same flow, different configs)
deployment_b_id = UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa") # scheduled, params_b
deployment_a_id = UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb") # triggered, params_a
Automation(
name="Run params_a after params_b completes",
enabled=True,
trigger=EventTrigger(
expect={"prefect.flow-run.Completed"},
posture=Posture.Reactive,
match_related=ResourceSpecification({
"prefect.resource.id": f"prefect.deployment.{deployment_b_id}"
}),
threshold=1,
within=timedelta(seconds=0),
),
actions=[
RunDeployment(
source="selected",
deployment_id=deployment_a_id,
# Optional if deployment A already has defaults:
parameters={
# put your params_a here
# "foo": "bar"
},
)
],
).create()
Notes and tips
- Filtering on parameter values isn’t supported in Event triggers; matching the specific deployment is the reliable filter.
- If you want to pass values from the upstream flow run into the downstream run, you can use Jinja templates in the action’s parameters (e.g., {{ flow_run.parameters['x'] }}), but this doesn’t help with filtering — only with parameterizing the action.
Docs
- Automations overview: Automations
- Chaining deployments via events: Chaining deployments with events
- Using parameters in action templates: Access parameters in templates
Want me to draft the automation for you if you share the two deployment IDs (or names) and the params_a you want to send?Mehdi Lallouache
04/02/2026, 8:45 AMMarvin
04/02/2026, 8:46 AMMarvin
04/02/2026, 8:48 AMcreated_by to apply a condition.
What you can filter on:
- Deployment (recommended), flow, work pool/queue, tags, and flow-run name/id via resource labels.
- You cannot filter by schedule slug or parameters.
Practical workarounds
- Easiest: create a separate deployment for that schedule. Then your automation can match on that deployment and react only to runs from that schedule.
- If you must keep multiple schedules on one deployment: use a webhook action to do conditional logic server-side.
- Automation: trigger on Completed for the deployment and CallWebhook to your service.
- Your service: fetch the flow run by ID, check flow_run.created_by.type == "SCHEDULE" and created_by.display_value (which is the schedule slug if set). Only then trigger the follow-up run.
Example automation using a webhook
from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture, ResourceSpecification
from prefect.events.actions import CallWebhook
Automation(
name="On deployment completion -> webhook filter by schedule",
enabled=True,
trigger=EventTrigger(
expect={"prefect.flow-run.Completed"},
posture=Posture.Reactive,
match_related=ResourceSpecification({
"prefect.resource.role": "deployment",
"prefect.resource.name": "my-deployment" # or match by id
}),
threshold=1,
within=timedelta(seconds=0),
),
actions=[
CallWebhook(
url="<https://your-service/handle-completion>",
method="POST",
payload={
"flow_run_id": {
"__prefect_kind": "jinja",
"template": "{{ flow_run.id }}"
}
},
)
],
).create()
Sketch of the webhook handler
from uuid import UUID
from fastapi import FastAPI, Request
from prefect.client.orchestration import get_client
TARGET_SCHEDULE_SLUG = "my-schedule-slug"
DOWNSTREAM_DEPLOYMENT_ID = UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")
app = FastAPI()
@app.post("/handle-completion")
async def handle(req: Request):
data = await req.json()
flow_run_id = data["flow_run_id"]
async with get_client() as client:
fr = await client.read_flow_run(flow_run_id)
cb = fr.created_by
if not cb or cb.type != "SCHEDULE":
return {"skipped": "not from a schedule"}
if (cb.display_value or "").lower() != TARGET_SCHEDULE_SLUG.lower():
return {"skipped": "different schedule"}
await client.create_flow_run_from_deployment(
deployment_id=DOWNSTREAM_DEPLOYMENT_ID,
parameters={"...": "..."} # your params_a
)
return {"triggered": True}
Related docs
- Automations overview: Automations
- Chaining deployments with events: Chaining deployments with events
- Using parameters in templates: Access parameters in templates
If you can share whether separate deployments per schedule is acceptable, I can outline the exact automation config for your setup. If not, I can help you wire up the webhook approach.