<@ULVA73B9P> I'm using Prefect 3. I'm creating dep...
# ask-marvin
n
@Marvin I'm using Prefect 3. I'm creating deployments using the create_deployment python function on a client object. How can I set a trigger for the deployment using the client? Note that I cannot use the .deploy method on the flow to deploy, I need to use the client API
m
thought for 831.0 seconds
Great question — and thanks for calling out that you’re using the client API in Prefect 3. Quick clarification to make sure I point you the right way: by “trigger” do you mean - a time-based schedule (Cron/Interval/RRule), or - an event/metric-based trigger (e.g., “run this deployment when another flow completes”)? How it works with the client in Prefect 3.x - Time-based schedules - You attach schedules when you create (or update) the deployment by setting its schedules list. - Programmatically, that’s done by passing a list of schedule definitions with your
create_deployment
call. Conceptually it looks like:
Copy code
from prefect.client.orchestration import get_client
  from prefect.client.schemas.actions import DeploymentCreate, DeploymentScheduleCreate
  from prefect.client.schemas.schedules import CronSchedule  # or IntervalSchedule / RRuleSchedule

  async def main():
      async with get_client() as client:
          dep = await client.create_deployment(
              DeploymentCreate(
                  name="my-deployment",
                  flow_name="my-flow",                 # or flow_id=...
                  entrypoint="path/to/flow.py:my_flow",
                  work_pool_name="my-pool",
                  schedules=[
                      DeploymentScheduleCreate(
                          schedule=CronSchedule(cron="0 * * * *", timezone="UTC"),
                          active=True,
                      )
                  ],
              )
          )
- Docs: Create schedules - Event/metric-based triggers - There is no
triggers=
parameter on
client.create_deployment(...)
. - In Prefect 3, event/metric-based triggers are managed as “deployment triggers” (or via automations) that are attached after the deployment is created. - You define a trigger using the schemas in
prefect.events.schemas.deployment_triggers
(e.g.,
DeploymentEventTrigger
,
DeploymentMetricTrigger
) and associate it with your deployment via the API. - Docs overview: - Deployment trigger schemas: Deployment triggers API reference - How-to guide: Creating deployment triggers If you can confirm which kind of “trigger” you need (schedule vs. event), I’ll share a concrete, ready-to-run snippet for the client-only approach you’re using. If it’s event-based, let me know the event you want to match (e.g., a specific flow-run completion, tags, posture reactive/proactive, etc.) and whether you’re on Prefect Cloud or OSS.
n
@Marvin By "trigger" I mean starting a deployed flow based on an event. How do I do this using the client Python object?
m
thought for 76.2 seconds
Thanks for clarifying. In Prefect 3, the supported way to start a deployment from an event using the Python client is to create an Automation with an event trigger that runs your deployment. There isn’t a
triggers=
argument on
client.create_deployment(...)
, and attaching “deployment triggers” is done after the deployment exists via Automations. Here’s a minimal end‑to‑end example that: - looks up your deployment - creates an Automation with an EventTrigger - runs that deployment when an upstream flow completes
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.automations import AutomationCore, EventTrigger, RunDeployment, Posture

async def main():
    flow_name = "upstream-flow"                  # the flow that emits the event
    target_flow = "downstream-flow"              # the deployed flow to run
    target_deployment = "prod"                   # your deployment name under target_flow

    async with get_client() as client:
        # 1) Get the target deployment's ID
        #    If you already have it, you can skip this lookup
        dep = await client.read_deployment_by_name(f"{target_flow}/{target_deployment}")

        # 2) Create an automation that runs this deployment when the upstream flow completes
        #    Adjust the match criteria to your needs (event, resource filters, tags, etc.)
        automation = AutomationCore(
            name="Run downstream on upstream completion",
            description="Runs the target deployment when upstream flow completes",
            posture=Posture.Reactive,
            enabled=True,
            trigger=EventTrigger(
                match={
                    "all": [
                        {"event": "prefect.flow-run.Completed"},
                        {"resource": {"type": "flow", "name": flow_name}},
                    ]
                }
                # You can also use 'match_related' to filter on related resources (e.g., tags)
            ),
            actions=[
                RunDeployment(
                    deployment_id=str(dep.id),
                    # Optional: pass parameters to the run
                    # parameters={"foo": "bar"},
                )
            ],
        )

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

if __name__ == "__main__":
    asyncio.run(main())
A few notes: - You can match on many event shapes (names, related resources, tags, etc.). See the event trigger how‑to and schemas for examples: - How‑to: Creating deployment triggers - Trigger schemas: Deployment trigger schemas - If you prefer to react to different events (e.g., failed runs, task events, custom events), just change the
match
expression accordingly. - This approach works with the client API and does not require using
flow.deploy()
. If you share the exact event pattern you want to react to (e.g., a flow name/tag, state, or a specific resource relationship), I can tailor the
match
section for you.
n
@Marvin Why did you pass the DeploymentCreate object to create_deployment? It does not support it, am I wrong?
m
thought for 180.5 seconds
Good catch — and thanks for flagging it. You’re right to question that. Depending on your Prefect 3.x version,
PrefectClient.create_deployment
may expect a plain JSON/dict payload (or keyword fields) rather than a
DeploymentCreate
model instance. If your method signature does not accept
DeploymentCreate
, pass a dict with the same fields you’re already using successfully to create the deployment. My example used the model for illustration, but it’s not required. The key point for your original question remains: event-based starts are handled by creating an Automation with an
EventTrigger
that runs your deployment. That part is independent of how you created the deployment. Here’s the client-only pattern to attach an event trigger:
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.automations import AutomationCore, EventTrigger, RunDeployment, Posture

async def main():
    upstream_flow_name = "upstream-flow"
    target_flow = "downstream-flow"
    target_deployment = "prod"

    async with get_client() as client:
        # Lookup the deployment you want to run
        dep = await client.read_deployment_by_name(f"{target_flow}/{target_deployment}")

        # Create an automation that runs that deployment on upstream completion
        automation = AutomationCore(
            name="Run downstream on upstream completion",
            description="Runs the target deployment when upstream flow completes",
            posture=Posture.Reactive,
            enabled=True,
            trigger=EventTrigger(
                match={
                    "all": [
                        {"event": "prefect.flow-run.Completed"},
                        {"resource": {"type": "flow", "name": upstream_flow_name}},
                    ]
                }
            ),
            actions=[
                RunDeployment(
                    deployment_id=str(dep.id),
                    # parameters={"foo": "bar"},  # optional
                )
            ],
        )

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

if __name__ == "__main__":
    asyncio.run(main())
References: - How-to: Creating deployment triggers - Trigger schemas: Deployment trigger schemas If you can share your exact Prefect version (e.g., 3.0.x/3.1.x) and the signature/error you see when passing
DeploymentCreate
, I can tailor the snippet precisely to your client method signature.