Nicholas Pini
10/09/2025, 6:56 PMMarvin
10/09/2025, 6:56 PMMarvin
10/09/2025, 7:10 PMcreate_deployment call. Conceptually it looks like:
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.Nicholas Pini
10/09/2025, 7:12 PMMarvin
10/09/2025, 7:12 PMMarvin
10/09/2025, 7:13 PMtriggers= 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
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.Nicholas Pini
10/09/2025, 7:22 PMMarvin
10/09/2025, 7:22 PMMarvin
10/09/2025, 7:25 PMPrefectClient.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:
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.