Shubham
10/19/2025, 12:53 PMMarvin
10/19/2025, 12:53 PMMarvin
10/19/2025, 12:57 PMimport asyncio
from prefect.client.orchestration import get_client, wait_for_flow_run
async def run_and_wait(deployment_id, parameters=None, timeout=3600):
async with get_client() as client:
fr = await client.create_flow_run_from_deployment(deployment_id, parameters=parameters or {})
final = await wait_for_flow_run(fr.id, client=client, timeout=timeout)
# Retrieve the return value (if it’s trackable)
result = None
if final.state and final.state.is_completed():
result = final.state.result() # may raise on failure; set raise_on_failure=False if desired
return final, result
asyncio.run(run_and_wait("your-deployment-id"))
- Reference: Create flow run from deployment
- Flow-level callback
- Pass a callback URL as a parameter to your flow, and at the end of the flow call your webhook with the result. This ensures you control exactly what’s sent and when.
Notes and caveats:
- Webhook trigger docs: Webhooks • Create a webhook
- Flow run “result” availability depends on whether the result is trackable/serializable. If the run produces an untrackable result, state.result() may not return data; you can instead push what you need to your own storage and include a reference in your callback.
- Prefect’s built-in webhooks do not support a long-running, synchronous response that blocks until the flow finishes.
If you can share your exact integration pattern (do you need the original HTTP request to block, or is an async callback acceptable?), I can suggest the best-fit option and help with setup.Shubham
10/19/2025, 1:16 PMMarvin
10/19/2025, 1:16 PMMarvin
10/19/2025, 1:16 PMimport asyncio
from prefect.client.orchestration import get_client, wait_for_flow_run
async def run_deployment_and_get_result(deployment_id: str, parameters: dict | None = None, timeout: int = 3600):
async with get_client() as client:
# 1) Create the flow run from the deployment
fr = await client.create_flow_run_from_deployment(
deployment_id=deployment_id,
parameters=parameters or {},
)
# 2) Wait for completion
final = await wait_for_flow_run(fr.id, client=client, timeout=timeout, log_states=True)
# 3) Fetch the return value (if trackable)
try:
value = final.state.result() # raises if the run failed
return final, value
except Exception as exc:
# Run failed or result retrieval failed
return final, exc
# Usage:
# final_run, result_or_exc = asyncio.run(run_deployment_and_get_result("your-deployment-id", {"x": 1}))
From synchronous code:
- Use asyncio.run(...) around the async function (as shown above).
- If you’re already in an event loop (e.g., Jupyter), use an async cell/function instead of asyncio.run.
Notes:
- If the flow run fails, state.result() will raise; catch and handle as shown.
- Result retrieval depends on the result being trackable/serializable. If you plan to fetch results out-of-band, return JSON-serializable data or configure result storage accordingly.
- You can also kick off the run, store flow_run.id, and later fetch it to check status and call state.result() once complete.
References:
- Create flow run from deployment: API reference
- Automations (if you prefer async callbacks instead of waiting): Automations and Creating automations
- Webhooks (for contrast with the non-blocking webhook flow): Webhooks • Create a webhook
If you share your runtime (web app, worker, function, etc.), I can suggest the cleanest sync/async integration pattern.