Chris Killpack
09/12/2025, 5:27 PMMarvin
09/12/2025, 5:27 PMMarvin
09/12/2025, 5:32 PMexport PREFECT_API_URL="<http://127.0.0.1:4200/api>"
export PREFECT_API_AUTH_STRING="admin:password"
Windows (PowerShell):
$env:PREFECT_API_URL="<http://127.0.0.1:4200/api>"
$env:PREFECT_API_AUTH_STRING="admin:password"
Python (simple trigger using helper):
from prefect.deployments.flow_runs import run_deployment
# "flow-name/deployment-name" must match your deployment exactly
flow_run = run_deployment(
name="my-flow/my-deployment",
parameters={"x": 1, "y": "hello"},
flow_run_name="manual-trigger-{{timestamp}}",
tags=["manual", "api-trigger"],
)
print("Created flow run:", flow_run.id)
Option B: Pass URL and credentials directly to the client (explicit)
Sync example:
from prefect.client.orchestration import SyncPrefectClient
with SyncPrefectClient(
api="<http://127.0.0.1:4200/api>",
auth_string="admin:password",
) as client:
# Look up the deployment by "flow/deployment" name
deployment = client.read_deployment_by_name("my-flow/my-deployment")
# Create a flow run from that deployment
flow_run = client.create_flow_run_from_deployment(
deployment_id=deployment.id,
parameters={"x": 1, "y": "hello"},
flow_run_name="manual-trigger-{{timestamp}}",
tags=["manual", "api-trigger"],
)
print("Created flow run:", flow_run.id)
Async example:
import asyncio
from prefect.client.orchestration import PrefectClient
async def main():
async with PrefectClient(
api="<http://127.0.0.1:4200/api>",
auth_string="admin:password",
) as client:
deployment = await client.read_deployment_by_name("my-flow/my-deployment")
flow_run = await client.create_flow_run_from_deployment(
deployment_id=deployment.id,
parameters={"x": 1, "y": "hello"},
flow_run_name="manual-trigger-{{timestamp}}",
tags=["manual", "api-trigger"],
)
print("Created flow run:", flow_run.id)
asyncio.run(main())
Optional: Configure settings in code (no env vars)
from prefect.settings import temporary_settings, PREFECT_API_URL, PREFECT_API_AUTH_STRING
from prefect.deployments.flow_runs import run_deployment
with temporary_settings({
PREFECT_API_URL: "<http://127.0.0.1:4200/api>",
PREFECT_API_AUTH_STRING: "admin:password",
}):
flow_run = run_deployment(
name="my-flow/my-deployment",
parameters={"x": 1},
flow_run_name="manual-trigger-{{timestamp}}",
)
print("Created flow run:", flow_run.id)
Notes:
- Ensure you have a running worker for the work pool your deployment uses; otherwise, the run will be scheduled but never picked up.
- For self-hosted Server auth, use PREFECT_API_AUTH_STRING or the client’s auth_string parameter. Avoid PREFECT_API_KEY unless you’re targeting Prefect Cloud.
If you share your deployment name and how your server auth is configured, I can tailor the snippet exactly to your setup.Chris Killpack
09/12/2025, 6:49 PMNate
09/12/2025, 7:05 PMrun_deployment will block until the invoked flow run reaches a terminal state, but you can use timeout=0 to fire-and-forget