Shubham
11/09/2025, 7:01 AMMarvin
11/09/2025, 7:01 AMMarvin
11/09/2025, 7:04 AMPREFECT_DEFAULT_RESULT_STORAGE_BLOCK and PREFECT_RESULTS_PERSIST_BY_DEFAULT=true.
2) Wait for the run to finish and load the result
- Poll the flow run’s state until it’s final, then resolve the stored result using Prefect’s client and get_state_result.
Example (Python, Prefect 3.x, async):
import asyncio
from prefect.client.orchestration import get_client
from prefect.states import get_state_result
async def trigger_and_get_result(deployment_id, parameters=None):
async with get_client() as client:
# 1) Create a flow run for the deployment
flow_run = await client.create_flow_run_from_deployment(
deployment_id=deployment_id,
parameters=parameters or {}
)
# 2) Poll for completion
while True:
state = await client.read_flow_run_state(flow_run.id)
if state.type.is_final():
break
await asyncio.sleep(5)
# 3) Resolve the result from storage
result = await get_state_result(state)
return result
# asyncio.run(trigger_and_get_result("your-deployment-uuid", {"param": "value"}))
If you must retrieve output over HTTP (no Python client)
- Consider writing your output as an artifact in the flow, then fetch it via the Artifacts API.
- Create from your flow with prefect.artifacts.create_markdown_artifact, create_table_artifact, etc.
- Read via /api/v3/artifacts filtered by flow_run_id or a collection key.
Key API references
- Create a run from a deployment: Create flow run from deployment
- Read flow run state: Read flow run state
- Results overview: Results (how they’re stored/resolved)
- Artifacts API: Artifacts endpoints
Notes
- If untrackable_result is true on the run, Prefect cannot persist/resolve that value; ensure your result is serializable with your chosen serializer (default is Pickle).
- The Prefect API does not serve the raw result bytes because results are stored externally; the client resolves them using the metadata in the state.Shubham
11/09/2025, 7:32 AMMarvin
11/09/2025, 7:32 AMMarvin
11/09/2025, 7:37 AMrun_deployment(...) starts the run and returns a FlowRun object (metadata), not the value.
- After the run completes, read its final state and resolve the value with prefect.states.get_state_result(state).
- Ensure your flow return is serializable and results are persisted (configure result storage or set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true).
Example (async script)
import asyncio
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.states import get_state_result
async def main():
# Wait for completion (timeout=None) or use a number of seconds; 0 returns immediately
flow_run = await run_deployment(
"flow-name/deployment-name",
parameters={"x": 1},
timeout=None
)
async with get_client() as client:
state = await client.read_flow_run_state(flow_run.id)
# Raises on failure by default; set raise_on_failure=False to handle errors yourself
result = get_state_result(state)
print("Flow returned:", result)
asyncio.run(main())
Inside another flow (subflow orchestration)
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.states import get_state_result
@flow
async def orchestrator():
fr = await run_deployment("child-flow/child-deploy", timeout=None) # tracked as subflow by default
async with get_client() as client:
state = await client.read_flow_run_state(fr.id)
child_result = get_state_result(state)
# use child_result however you like...
return child_result
If you can’t or don’t want to persist results, consider emitting an artifact in the flow and reading it later via the Artifacts API.
Docs
- Run deployments programmatically
- Results (persistence and retrieval)