<@ULVA73B9P> if I run deployment via API can I get...
# ask-marvin
s
@Marvin if I run deployment via API can I get the actual return response of the function back in API response
m
thought for 192.9 seconds
Short answer: not directly. - When you start a flow run from a deployment via the API, the response only includes run metadata (IDs, state, timestamps, etc.). It does not contain your function’s return value. - Prefect stores results outside the API (e.g., local file system, S3, GCS). The API/state will hold metadata pointing to the stored result, not the value itself. How to get the actual return value 1) Ensure the result is persisted - Configure result storage (e.g., S3 block) or rely on the default local storage, and make sure your return value is serializable. - Global toggles you can use:
PREFECT_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):
Copy code
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.
s
@Marvin If using python prefect client and run the deployment can I get the return value to use in down the line
m
thought for 302.1 seconds
Yes. Using the Python client you can trigger a deployment and then load the flow’s actual return value once the run finishes. Key points -
run_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)
Copy code
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)
Copy code
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)