<@ULVA73B9P> I want t launch a flow with a set of ...
# marvin-ai
m
@Marvin I want t launch a flow with a set of parameters (let's call it params_a) when the same flow with parameters_b is completed, how to do that (to give you more context: this first run with paramaters_b is controlled by a deployment with a schedule)?
m
thought for 299.2 seconds
Great question — this is a perfect use-case for an Automation in Prefect 3.x. High-level approach - Create two deployments of the same flow: - Deployment B: scheduled, uses parameters_b - Deployment A: no schedule, uses parameters_a (or accept overrides) - Create an Automation that listens for Completed events from Deployment B and runs Deployment A. Why two deployments? Event triggers cannot filter on specific parameter values. If you trigger the same deployment again, you’ll create a loop (your params_a run would also emit a Completed event that retriggers the automation). Using a second deployment avoids that. UI steps 1. Ensure you have two deployments of the same flow: - “my-flow - mode-b” (scheduled, default params_b) - “my-flow - mode-a” (no schedule, default params_a) 2. Go to Automations → New Automation 3. Trigger: - Type: Event - Posture: Reactive - Expected events:
prefect.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
Copy code
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?
m
@Marvin is it possible to react to completed flow run coming from a schedule with a specific slug only?
m
thought for 149.4 seconds
Short answer: not directly. In Prefect 3.x, Event-triggered automations can’t filter flow-run events by the originating schedule’s slug/name. Flow-run events do not expose the schedule as a related resource that you can match on, and EventTriggers can’t inspect payload fields like
created_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
Copy code
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
Copy code
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.