Mohamed Zaki
08/18/2025, 4:10 PMMarvin
08/18/2025, 4:10 PMMarvin
08/18/2025, 4:12 PMimport asyncio
from prefect.client.orchestration import PrefectClient
async def main():
async with PrefectClient() as client:
# Look up the deployment by "flow_name/deployment_name"
deployment = await client.read_deployment_by_name("my-flow/my-deployment")
print(f"Deployment ID: {deployment.id}")
# Create a flow run from that deployment ID
flow_run = await client.create_flow_run_from_deployment(
deployment_id=deployment.id,
parameters={"x": 1},
flow_run_name="triggered-from-python",
tags=["manual"],
# scheduled_time can be set if you want it to run later
# scheduled_time=datetime.utcnow() + timedelta(minutes=5),
)
print(f"Created flow run: {flow_run.id}")
asyncio.run(main())
Notes:
- The name format is flow_name/deployment_name
.
- The client methods are async; use asyncio.run(...)
in scripts.
- You don’t actually need the UUID if you just want to run it; there’s a shortcut:
from prefect.deployments import run_deployment
flow_run = run_deployment(
"my-flow/my-deployment",
parameters={"x": 1},
flow_run_name="shortcut"
)
Docs, for your further amusement:
- Creating deployments
- Deploy via Python
If you’re on Prefect 2.x instead, say so and I’ll adjust (and quietly suggest upgrading).