Short answer: Not directly. In Prefect 3.x, calling a deployment from Python returns a FlowRun object. You can then fetch the deployed flow’s actual return value from that run’s final state. How to do it 1) Make sure the deployed flow persists its result (so it can be retrieved across processes): - Add

persist_result=True

on the deployed flow, or set

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

.

from prefect import flow

@flow(persist_result=True)
def child_flow(x: int) -> int:
    return x + 1

2) Trigger the deployment and retrieve its result: - From within another flow (default

as_subflow=True

so it shows nested in the UI):

from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import get_state_result

@flow
def orchestrator():
    # Triggers and waits for completion
    fr = run_deployment(
        name="child-flow/my-deployment",
        parameters={"x": 41},
    )
    # Extract the actual Python return value
    value = get_state_result(fr.state)
    # Use it in subsequent processing
    return value * 2

orchestrator()

- From a plain Python script using async:

import anyio
from prefect.deployments import run_deployment
from prefect.states import get_state_result

async def main():
    fr = await run_deployment("child-flow/my-deployment", parameters={"x": 41})
    value = get_state_result(fr.state)
    print("Child result:", value)

anyio.run(main)

Notes -

run_deployment(...)

waits for the run to finish (polling) and returns a FlowRun with a terminal state. If you set a

timeout

, it may return before completion; don’t call

get_state_result

until the run is finished. - Result retrieval requires that the child flow’s result is persisted; otherwise there’s no cross-process result to fetch. - If you don’t need to go through a deployment boundary, simply import and call the flow directly to get the value immediately in-process. Docs - Run deployments from Python - Results and result persistence - get_state_result API

Great question. There are two common ways you’ll get a flow’s state for “the same flow you deployed”: 1) Call the flow directly (in-process, no deployment) - By default, calling a flow returns its raw Python value. - If you want the State object instead, pass

return_state=True

.

from prefect import flow
from prefect.states import get_state_result

@flow
def my_flow(x: int) -> int:
    return x + 1

# Raw value
val = my_flow(41)           # -> 42

# State (contains result + metadata)
state = my_flow(41, return_state=True)
print(state.type, state.name, state.message)  # e.g., Completed Completed None
val_from_state = get_state_result(state)      # -> 42

2) Run the deployment and read its final state -

run_deployment(...)

returns a FlowRun object; the deployed flow’s final state is at

flow_run.state

.

from prefect.deployments import run_deployment
from prefect.states import get_state_result

# Triggers the deployment and waits for completion
flow_run = run_deployment("my-flow/my-deployment", parameters={"x": 41})

state = flow_run.state
print(state.type, state.name)                 # e.g., Completed Completed
val = get_state_result(state)                 # -> 42 (requires persisted result)

Important notes - To retrieve a deployed flow’s return value across processes, the deployed flow should persist its result (e.g.,

@flow(persist_result=True)

or set

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

). Direct in-process calls don’t need persistence. - The state object gives you status and metadata (

.type

,

.name

,

.message

) and can be turned into the actual return value with

get_state_result(...)

. Helpful links: - Run deployments from Python - Results and result persistence - Flow basics and calling patterns

Short version: When you start a deployment via the REST API, Prefect returns a FlowRun and its State. The REST API does not return the raw Python return value. To capture the final return, enable result persistence on the deployed flow, then read the run’s terminal State from the API and fetch the result from the configured result storage using the metadata in that State. What to do 1) Ensure the deployed flow persists its result - Add

persist_result=True

on the flow (or set

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

). - This makes the flow’s return value available across processes/storage. 2) Create a run from the deployment

# POST /api/v3/deployments/{deployment_id}/create_flow_run
curl -X POST "$PREFECT_API_URL/api/v3/deployments/<deployment_id>/create_flow_run" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"parameters": {"x": 41}, "flow_run_name": "api-triggered"}'

This returns a JSON payload with the

id

of the new flow run. 3) Poll for completion, then read the final state

# GET /api/v3/flow_runs/{flow_run_id}/state
curl -s -H "Authorization: Bearer $PREFECT_API_KEY" \
  "$PREFECT_API_URL/api/v3/flow_runs/<flow_run_id>/state"

- Wait until

type

is a terminal state (e.g., COMPLETED or FAILED). 4) Fetch the actual return value from result storage - The state’s

data

contains result metadata (e.g.,

storage_key

,

serializer

, and possibly a

storage_block_id

), not the raw value. - Use that metadata to read the object from your result storage (local filesystem, S3, GCS, etc.) and deserialize it according to the

serializer

. Notes and options - Pure REST-only: You’ll need to read the state’s

data

metadata and then use your own client (e.g., AWS SDK for S3) to fetch and deserialize the result. For example, if using S3 result storage, read the object at the

storage_key

path from the bucket defined in your storage block. - Easiest programmatic path: If you’re okay using the Python SDK after you have the

flow_run_id

, you can let Prefect do the fetching/deserialization:

import anyio
from prefect.client import get_client
from prefect.states import get_state_result

async def main(flow_run_id: str):
    async with get_client() as client:
        state = await client.read_flow_run_state(flow_run_id)
    value = get_state_result(state)  # actual Python return value
    print(value)

anyio.run(main, "<flow_run_id>")

- If the flow’s result is not persisted, there’s no cross-process result to retrieve via the API. Docs - REST API reference (browse endpoints in your environment): REST API index - Create flow run from a deployment: Create Flow Run From Deployment - Read flow run state: Read Flow Run State - Results and persistence: Results and result persistence If you can share: - Are you on Prefect Cloud or self-hosted? - Which result storage backend are you using (local, S3, GCS, …)? I can show an exact example for that storage.

Here’s a minimal, working pattern to persist a flow’s return value to S3 and retrieve it later. Assumes Prefect 3.x and the AWS integration. Setup (one-time) - Install the AWS extra:

pip install "prefect[aws]"

- Create and save blocks (either via code or the UI). Code example:

from prefect_aws.credentials import AwsCredentials
from prefect_aws.s3 import S3Bucket

# Option A: Explicit keys (or use an instance profile/role and omit keys)
aws = AwsCredentials(
    aws_access_key_id="YOUR_KEY",
    aws_secret_access_key="YOUR_SECRET",
    region_name="us-east-1",
)
aws.save("aws-creds", overwrite=True)

s3_block = S3Bucket(
    bucket_name="my-prefect-results",
    aws_credentials=aws,
    # optional: endpoint_url="<https://s3.us-east-1.amazonaws.com>"
)
s3_block.save("my-s3-results", overwrite=True)

Define your flow to persist results to S3

from prefect import flow
from prefect.serializers import JSONSerializer
from prefect_aws.s3 import S3Bucket

# Load the saved S3 block
s3_results = S3Bucket.load("my-s3-results")

@flow(
    persist_result=True,                 # turn on result persistence
    result_storage=s3_results,           # where to persist
    result_serializer=JSONSerializer(),  # how to serialize (JSON is easy to inspect)
)
def child_flow(x: int) -> dict:
    return {"answer": x + 1}

Trigger the deployment and read the result (Python SDK) - If you deploy

child_flow

, you can trigger it and fetch its return value like this:

from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import get_state_result

@flow
def orchestrator():
    fr = run_deployment("child_flow/prod", parameters={"x": 41})  # waits for completion
    value = get_state_result(fr.state)  # -> {'answer': 42}
    print(value)
    return value

orchestrator()

Trigger via REST API and fetch the result 1) Create a run

curl -X POST "$PREFECT_API_URL/api/v3/deployments/<deployment_id>/create_flow_run" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"parameters": {"x": 41}}'

2) Poll the run’s state

curl -s -H "Authorization: Bearer $PREFECT_API_KEY" \
  "$PREFECT_API_URL/api/v3/flow_runs/<flow_run_id>/state"

- Wait until

type

is

COMPLETED

. The state’s

data

contains result metadata (including a

storage_key

and serializer info), not the raw value. 3) Fetch from S3 (two options) - Easiest: Use the Prefect Python SDK to deserialize for you:

import anyio
from prefect.client import get_client
from prefect.states import get_state_result

async def fetch(flow_run_id: str):
    async with get_client() as client:
        state = await client.read_flow_run_state(flow_run_id)
    return get_state_result(state)

print(anyio.run(fetch, "<flow_run_id>"))

- Manual via boto3 (if you know the bucket and used JSONSerializer):

import boto3, json

# From the state JSON returned by the API:
# storage_key = state["data"]["storage_key"]
storage_key = "<key-from-state-data>"

s3 = boto3.client("s3", region_name="us-east-1")
obj = s3.get_object(Bucket="my-prefect-results", Key=storage_key)
payload = obj["Body"].read()
value = json.loads(payload.decode("utf-8"))
print(value)  # {'answer': 42}

Note: Prefect manages result encoding/metadata. For anything beyond simple JSON, prefer using the Prefect SDK (

get_state_result

) to avoid handling serializer details yourself. Permissions to S3 - Ensure the runtime (your worker/flow environment) can write and read objects: - s3:PutObject, s3:GetObject on the bucket - s3:ListBucket on the bucket Helpful docs - Results and result persistence - Prefect AWS integration - Run deployments from Python

- Create Flow Run from Deployment (REST) - Read Flow Run State (REST) If you share your region and whether you’re using roles or static keys, I can tailor the block creation snippet and IAM policy examples.

Short answer: Your worker (the thing that actually runs the flow) needs network access and credentials to S3. The Prefect server/API does not read or write your results, so it does not need S3 access. Details - Worker/runtime - Must be able to write results to S3 (and read if your code does). - Needs network egress to the S3 endpoint and valid AWS credentials (instance/task role, IRSA, or static keys). - If you also store code/artifacts on S3 (e.g., using deploy steps push_to_s3), the worker needs read access to that bucket/prefix too. - Prefect server (Cloud or self-hosted) - Does not access S3 for results or code. It only stores metadata and block configuration. - Any client that wants to fetch the flow’s return value later - If you use the Python SDK (

get_state_result(state)

), the machine running that code must also be able to access S3 and have credentials. The SDK will use the S3 block referenced in the state to read/deserialize the result. Minimal IAM example for a results bucket

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "PrefectResultsAccess",
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::my-prefect-results",
        "arn:aws:s3:::my-prefect-results/prefix/*"
      ]
    }
  ]
}

- Add

s3:DeleteObject

if you plan to clean up. - If the bucket uses SSE-KMS, include

kms:Encrypt

,

kms:Decrypt

,

kms:GenerateDataKey

for the key. Common setups - Kubernetes: use IRSA or mount AWS creds; ensure the pod can reach S3 (VPC endpoints or NAT). - ECS: use a task role with S3 permissions. - Local/Docker: export AWS creds in env vars or use a shared profile. If you share where your worker runs (K8s/ECS/VM/local) and whether you use roles or static keys, I can provide a tailored setup snippet and IAM policy.

Given that your worker can only egress to your database and the Prefect API (no S3 or other object storage), you have a few workable patterns to get the flow’s “return value” back: Option A — Call the flow directly (no deployment boundary) - If you control the orchestrator code, import the flow and call it as a normal function/subflow. The value is returned in-process, no persistence needed or network egress. - This requires running the orchestrator and child on the same runtime.

from prefect import flow

@flow
def child(x: int) -> dict:
    return {"answer": x + 1}

@flow
def orchestrator():
    result = child(41)  # direct call, immediate return value
    # proceed using `result`
    return result

Option B — Write the result to your database from the flow - Since DB egress is allowed, persist the result there and read it from your caller. - This bypasses Prefect result storage entirely and avoids any extra egress.

from prefect import flow
import json
import psycopg2  # example; use your DB client

@flow
def child(x: int):
    result = {"answer": x + 1}
    # write result to your DB
    conn = psycopg2.connect("<postgresql://user:pass@host:5432/db>")
    with conn, conn.cursor() as cur:
        cur.execute(
            "insert into flow_results(flow_run_id, payload) values (%s, %s)",
            (prefect.runtime.flow_run.id, json.dumps(result)),
        )
    return result

Option C — Publish the result as a Prefect Artifact (stored in Prefect’s DB) - Artifacts are persisted via the Prefect API (which your worker can access) and can be fetched later via the Prefect REST API or Python client. This avoids any S3/FS access. - Best for small/JSON-like outputs; keep payloads modest. Create the artifact in your flow:

from prefect import flow
from prefect.artifacts import create_markdown_artifact
import json

@flow
def child(x: int):
    result = {"answer": x + 1}

    # Use a stable key if you want to fetch “latest by key”
    key = "child_result"
    create_markdown_artifact(
        key=key,
        markdown=f"```json\n{json.dumps(result)}\n```",
        description="Return value for child flow"
    )
    return result

Fetch the artifact via Python SDK:

import anyio
from prefect.client import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
from prefect.client.schemas.sorting import ArtifactSort

async def fetch_latest_by_key(key: str):
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            limit=1,
            sort=ArtifactSort.UPDATED_DESC,
            artifact_filter=ArtifactFilter(key=ArtifactFilterKey(any_=[key])),
        )
    if not artifacts:
        return None
    # artifacts[0].data contains your markdown string; parse out the JSON as needed
    return artifacts[0].data

print(anyio.run(fetch_latest_by_key, "child_result"))

Or via REST API (filter by key):

curl -X POST "$PREFECT_API_URL/api/v3/artifacts/filter" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "limit": 1,
        "sort": "UPDATED_DESC",
        "artifacts": {
          "key": { "any_": ["child_result"] }
        }
      }'

Notes and tradeoffs - Prefect API never returns raw “result” values from states; it only stores metadata. Without S3/FS egress, you can’t use standard result persistence to retrieve via

get_state_result(...)

. - Artifacts are great for small, structured outputs and only require Prefect API egress. - Writing to your own DB is best for larger or more complex outputs and gives you full control. - If you must keep the deployment + API-trigger pattern, combine it with Option B or C: the flow writes its result to your DB or publishes a Prefect Artifact, then your caller polls the Prefect API for completion and reads the value from DB/artifact. References

- Artifacts (Python): prefect.artifacts API - Artifacts (REST list/read): - Read artifacts - Read latest artifacts - Why the API won’t return raw results: Results and result persistence If you can share which of these fits your topology (imported subflow vs deployment + API trigger) and your DB type, I can provide a tailored code snippet (including safe secret handling) for your setup.