Yes.

run_deployment

returns a

FlowRun

object; to get the actual return value of the flow, wait for the run to finish and then resolve the state’s result. Here’s a pattern that works reliably in Prefect 3.x:

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

# Trigger the deployment as a subflow
flow_run = run_deployment(
    name="name",
    parameters={...},
    as_subflow=True,
)

# Ensure it reaches a terminal state
flow_run = wait_for_flow_run(flow_run.id)

# Retrieve the returned value from the child flow (e.g., your list)
result = get_state_result(flow_run.state)  # raises if the run failed

Notes: -

get_state_result

will raise on failure by default; you can pass

raise_on_failure=False

if you prefer to handle it yourself. - Make sure the returned object is serializable with your configured result serializer. If you control the target flow and don’t need deployment isolation, an even simpler option is to call the flow directly as a subflow and get its return value immediately:

# Assuming you can import the flow function
returned_list = child_flow(**params)  # runs as a subflow and returns the value

Refs: - Run deployments (Prefect 3) - Source for state result utilities: prefect.states

You’re seeing that because you’re in an async context.

run_deployment

(and

wait_for_flow_run

) are sync-compatible: - In sync code, call them normally. - In async code (e.g., inside

async def

flow), you must

await

them. Otherwise you get a coroutine and accessing

.id

fails. Examples: - Async parent flow:

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

@flow
async def parent():
    fr = await run_deployment(name="name", parameters={...}, as_subflow=True)
    fr = await wait_for_flow_run(fr.id)
    value = get_state_result(fr.state)  # raises if failed
    return value

- Sync parent flow:

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

@flow
def parent():
    fr = run_deployment(name="name", parameters={...}, as_subflow=True)
    fr = wait_for_flow_run(fr.id)
    value = get_state_result(fr.state)  # raises if failed
    return value

Notes: -

get_state_result

will raise on failure by default; set

raise_on_failure=False

to handle errors yourself. - Ensure the child flow’s return value is serializable with your result serializer. Refs: - Run deployments (Prefect 3) - State result helpers

Great question. Short answer: pick the style that matches your workload and ecosystem, then layer in concurrency appropriately. Use an async flow when: - You’re using async-first libraries (aiohttp, asyncpg, aioboto3, etc.) and want to await them without blocking. - You want to orchestrate many I/O-bound actions concurrently in a single process (e.g., launching several deployments and waiting on them with asyncio.gather). - You’re already in an async environment (e.g., a FastAPI endpoint). Use a sync flow when: - Your code and libraries are synchronous (most common). - You have CPU-bound work — pair with a process-based task runner to bypass the GIL. - You want the simplest mental model; you can still get concurrency via task runners and task.submit. Patterns - Async flow orchestrating multiple deployments concurrently

import asyncio
from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@flow
async def orchestrate():
    # Launch multiple deployments concurrently
    runs = await asyncio.gather(
        run_deployment(name="proj/dep-a", parameters={...}, as_subflow=True),
        run_deployment(name="proj/dep-b", parameters={...}, as_subflow=True),
    )

    # Wait for completion concurrently
    finished = await asyncio.gather(*(wait_for_flow_run(r.id) for r in runs))

    # Collect return values
    results = [get_state_result(fr.state) for fr in finished]
    return results

- Sync flow with CPU-bound concurrency (use processes)

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def heavy_compute(x: int) -> int:
    # CPU-bound work
    return sum(i * i for i in range(10_000_000))

@flow(task_runner=ProcessPoolTaskRunner())
def cpu_flow():
    futs = [heavy_compute.submit(i) for i in range(4)]
    return [f.result() for f in futs]

- Sync flow with I/O-bound concurrency (use threads)

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def fetch(url: str) -> str:
    # blocking I/O (requests, boto3, etc.)
    ...

@flow(task_runner=ThreadPoolTaskRunner())
def io_flow(urls: list[str]):
    futs = [fetch.submit(u) for u in urls]
    return [f.result() for f in futs]

Gotchas - In async flows, you must await async-capable helpers like

run_deployment(...)

and

wait_for_flow_run(...)

. Forgetting to await yields a coroutine object (the error you saw). - Avoid long blocking CPU or blocking I/O directly in async flows; offload to tasks and pick a thread/process runner. - If you only need to call other deployments and handle results, async flows are great for high fan-out/fan-in orchestration. Docs and references - Run deployments: Run deployments - Concurrency patterns and task runners: Run work concurrently - Flow concepts: Flows concept

Great question — it boils down to whether you need concurrency or not. Use

result = task(...)

when: - You need the task’s output immediately (sequential logic). - You’re not trying to overlap this task with other work. - You want simpler control flow — exceptions raise right away. Use

fut = task.submit(...)

(then later

result = fut.result()

) when: - You want concurrency/fan-out: fire off many tasks at once, then fan-in later. - You want to overlap this task with other work before blocking for its result. - You want to pass futures to other tasks as upstream dependencies (Prefect will wait on them automatically). Key points: -

.submit(...)

returns a

PrefectFuture

. Get the value with

fut.result()

(it’s a method, not a property). - If you

.submit(...)

and immediately call

.result()

, that’s effectively the same as calling

task(...)

— so prefer

task(...)

for clarity in that case. - In async flows, you can still use

.submit(...)

and call

fut.result()

when you need the value. Use a thread/process task runner to avoid blocking the event loop with heavy work. Patterns - Sequential (simple and immediate)

@flow
def pipeline():
    a = task_a(...)
    b = task_b(a)  # runs after task_a completes
    return b

- Concurrent fan-out/fan-in

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner())
def pipeline(urls):
    futs = [download.submit(u) for u in urls]  # launch all
    # do other work here if needed...
    results = [f.result() for f in futs]       # fan-in
    return results

- Passing futures as upstream dependencies (Prefect handles waiting)

@flow
def pipeline():
    f1 = task_a.submit(...)
    f2 = task_b.submit(f1)  # will wait for f1 automatically
    return f2.result()

Docs: - Run work concurrently: Concurrency patterns in Prefect

That error means you’re still calling an async function without awaiting it. In Prefect 3.x, both

run_deployment

and

wait_for_flow_run

are “sync-compatible”: - In sync code, call them normally. - In async code, you must

await

them. If you don’t, you’ll get a coroutine and accessing

.state

or

.id

will fail. Use one of these patterns: Sync parent flow

from prefect import flow, get_run_logger
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@flow
def parent():
    logger = get_run_logger()
    fr = run_deployment(name="proj/dep", parameters={...}, as_subflow=True)
    <http://logger.info|logger.info>(f"Subflow triggered successfully. Flow run ID: {fr.id}")

    fr = wait_for_flow_run(fr.id)  # blocking in sync code
    result = get_state_result(fr.state)

    <http://logger.info|logger.info>(f"Robocopy flow completed with state: {fr.state}")
    <http://logger.info|logger.info>(f"Result: {result}")
    return result

Async parent flow

from prefect import flow, get_run_logger
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@flow
async def parent():
    logger = get_run_logger()
    fr = await run_deployment(name="proj/dep", parameters={...}, as_subflow=True)
    <http://logger.info|logger.info>(f"Subflow triggered successfully. Flow run ID: {fr.id}")

    fr = await wait_for_flow_run(fr.id)  # must await in async code
    result = get_state_result(fr.state)

    <http://logger.info|logger.info>(f"Robocopy flow completed with state: {fr.state}")
    <http://logger.info|logger.info>(f"Result: {result}")
    return result

If you’ve removed

async

from your flow but still see a coroutine, something in the call stack is likely still async (e.g., running inside an async task or an async entrypoint like FastAPI). If you can share your flow signature and where this code runs from, I can confirm which variant you need. Refs: - Run deployments

Thanks for the stack trace — that confirms

wait_for_flow_run(...)

is being treated as async in your execution context, so calling it without

await

returns a coroutine (hence the

'coroutine' object has no attribute 'state'

when you try to use it). Even though your flow function is now sync, Prefect may still be running it inside an event loop. For “sync-compatible” Prefect APIs, that means: - In a pure sync context: call them normally - If an event loop is active: you must either await them or explicitly run the coroutine as sync You have two reliable options: Option A — Make the parent flow async and await

from prefect import flow, get_run_logger
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@flow
async def main_orchestrator():
    logger = get_run_logger()
    fr = await run_deployment(name="proj/dep", parameters={...}, as_subflow=True)
    <http://logger.info|logger.info>(f"Subflow triggered successfully. Flow run ID: {fr.id}")

    fr = await wait_for_flow_run(fr.id, log_states=True)
    result = get_state_result(fr.state)

    <http://logger.info|logger.info>(f"Robocopy flow completed with state: {fr.state}")
    <http://logger.info|logger.info>(f"Result: {result}")
    return result

Option B — Keep the flow sync and force-sync the coroutine

from prefect import flow, get_run_logger
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result
from prefect.utilities.asyncutils import run_coro_as_sync

@flow
def main_sync_orchestrator_flow():
    logger = get_run_logger()
    fr = run_deployment(name="proj/dep", parameters={...}, as_subflow=True)
    <http://logger.info|logger.info>(f"Subflow triggered successfully. Flow run ID: {fr.id}")

    # Force the async call to run synchronously
    fr = run_coro_as_sync(wait_for_flow_run(fr.id, log_states=True))
    result = get_state_result(fr.state)

    <http://logger.info|logger.info>(f"Robocopy flow completed with state: {fr.state}")
    <http://logger.info|logger.info>(f"Result: {result}")
    return result

Notes: - You already confirmed

run_deployment(...)

returned a

FlowRun

(you logged

fr.id

), so the coroutine is coming from

wait_for_flow_run(...)

. - If you flip to an async flow, remember you must

await

both

run_deployment(...)

and

wait_for_flow_run(...)

. - You can pass

log_states=True

to

wait_for_flow_run

to see the state transitions while waiting. Refs: - Run deployments - Source for

wait_for_flow_run

(sync-compatible): prefect.flow_runs - Utilities for bridging async/sync: prefect.utilities.asyncutils

You nailed the diagnosis: your child flow (running on Windows) persisted its return value to its local default result store (LocalFileSystem under C:\Users\TalmoLab\.prefect\storage). When your parent (running on macOS) calls get_state_result, Prefect tries to read that same file path locally — which doesn’t exist — hence the mixed-path error. Important: Prefect Server does not store your flow results. The server stores only metadata and a pointer to the result’s location. To fetch results across machines, both machines must read from a shared result store. How to fix it Pick a shared result store and make the child flow persist to it. Common options: - Cloud object storage: S3, GCS, Azure Blob (recommended) - Network share: SMB/NFS reachable from both Windows and macOS - Any RemoteFileSystem URL that both sides can access Option A: Set result storage on the child flow

from prefect import flow
from prefect.states import get_state_result

# Example: S3 (requires `pip install 'prefect[aws]'`)
from prefect_aws.s3 import S3Bucket

# Load a configured S3Bucket block (create it in code or via UI beforehand)
s3 = S3Bucket.load("default-result-store")

@flow(result_storage=s3, persist_result=True)
def child_flow(...):
    # return your list
    return ["a", "b"]

Or with an SMB/NFS share via RemoteFileSystem:

from prefect import flow
from prefect.filesystems import RemoteFileSystem

fs = RemoteFileSystem(basepath="<smb://user:pass@server/share/prefect-results>")

@flow(result_storage=fs, persist_result=True)
def child_flow(...):
    return ["a", "b"]

Redeploy the child so it uses this storage. Then your parent can keep using:

fr = await/ wait_for_flow_run(...)
result = get_state_result(fr.state)

and it will download from the shared store. Option B: Configure a global default result store Set these on BOTH the child’s execution environment and any environment that calls get_state_result (your mac): - PREFECT_RESULTS_PERSIST_BY_DEFAULT=true - PREFECT_DEFAULT_RESULT_STORAGE_BLOCK=<block slug shown in the UI, e.g. s3-bucket/default-result-store> This makes all flows/tasks persist to that shared store by default. Notes - Make sure credentials for the shared store are available on both machines. - If your payloads are small and you need a quick workaround, you could emit an artifact in the child flow (e.g., a markdown or table) and read it back via API — but for programmatic, reliable data passing, shared result storage is the right approach. - Your current error shows Prefect is joining your mac’s local base path with the Windows path in the storage key — that’s a telltale sign of using LocalFileSystem on different machines. Docs - Results and result storage in Prefect 3 If you share what storage you prefer (S3/GCS/Azure/SMB), I can give you a drop-in config snippet for that backend.

You’re hitting an fsspec init error: the SMB filesystem driver needs its connection parameters at construction time. With a RemoteFileSystem block, Prefect calls fsspec.filesystem("smb", **settings) first, then uses your basepath for paths. Since your Settings were empty, fsspec didn’t get a host (or creds), so it failed with SMBFileSystem.__init__() missing 'host'. You’ve got two good options: Option 1 — Use the dedicated SMB block (recommended) - In the Prefect UI, create a block of type “SMB” (not RemoteFileSystem). - Fill: - Host: addr - Username: username - Password: your password (Secret) - Share path: talmo/than/prefect-results (the share and path on the server; don’t include smb:// or host) - Port: 445 (optional) - In your flow code, load and use it for results:

from prefect import flow
from prefect.filesystems import SMB

smb_store = SMB.load("my-smb-results")

@flow(result_storage=smb_store, persist_result=True)
def child_flow(...):
    return ["a", "b", "c"]

- Redeploy the child flow. - Ensure your parent environment can also access SMB (see “Dependencies” below). Now get_state_result will fetch from the SMB share on both machines. Option 2 — Keep RemoteFileSystem, but supply Settings - In the Prefect UI for your RemoteFileSystem block: - Basepath: smb://addr/talmo/than/prefect-results - Settings (JSON), use fsspec’s SMBFileSystem args:

{
  "host": "addr",
  "username": "username",
  "password": "your-password",
  "port": 445
}

- Remove creds from the URL; keep them in Settings for security. - Redeploy and retry. Dependencies (both machines) - Install the SMB fsspec driver in any environment that will write/read results (your Windows worker and your macOS parent):

pip install smbprotocol smbfs

- Verify connectivity/permissions from both sides. Optional: make it the default result store - Set on both environments where results are produced/consumed: - PREFECT_RESULTS_PERSIST_BY_DEFAULT=true - PREFECT_DEFAULT_RESULT_STORAGE_BLOCK=smb/<your-block-slug> - Or keep using per-flow result_storage as shown above. Why this fixes your issue - Prefect Server only stores a pointer to the result. If the child writes to a local path on Windows, your Mac cannot read it. By using a shared SMB result store (and providing host/creds at FS creation), both machines can read the same persisted result, so get_state_result works. Docs - Results and result storage - Blocks concept If you share your exact SMB share naming (server, share name, path), I can sanity-check the “Share path” field for the SMB block.

You’re right to suspect a host parsing issue. The error shows Prefect/fsspec handed smbclient a host value of pool1.vast.salk.edusmb::445 which looks like your hostname ended with “.edu” and then “smb:” and “:445” got appended, yielding “edusmb::445”. This typically happens when: - The host was entered with a scheme (e.g., “smb://pool1.vast.salk.edu”) in a field that expects just a hostname, or - Both the basepath and the settings specify host/scheme, causing fsspec to compose a malformed target, or - There’s a stray suffix or whitespace in the host field. To make this robust, pick one approach and stick to it. I recommend the dedicated SMB block. Recommended: SMB block (no schemes anywhere) - In the Prefect UI, create a block of type “SMB” (not RemoteFileSystem). - Fill the fields exactly like this: - SMB host: pool1.vast.salk.edu - No “smb://”, no slashes, no port, no extra text. - SMB port: 445 - SMB username: your user (if needed, DOMAIN\\username) - SMB password: your password (stored as a Secret) - Share path: than/prefect-results - This should be the UNC path part after the host. For UNC \\pool1.vast.salk.edu\than\prefect-results, use “than/prefect-results”. - In your child flow, use:

from prefect.filesystems import SMB

smb_store = SMB.load("my-smb-results")

@flow(result_storage=smb_store, persist_result=True)
def child_flow(...):
    return ["a", "b"]

- Redeploy the child and re-run. Your parent can then call

get_state_result(fr.state)

from macOS and Prefect will read from SMB. Alternative: RemoteFileSystem block If you prefer RemoteFileSystem, ensure the fields are consistent: - Basepath: smb://pool1.vast.salk.edu/than/prefect-results - Include scheme and host here. - Settings JSON: include only credentials and port, not the host or scheme.

{
  "username": "DOMAIN\\username",
  "password": "your-password",
  "port": 445
}

If you include “host” in Settings and also put the host in Basepath, the underlying driver may compose an invalid target. Quick sanity check from the worker machine Try a direct fsspec test to validate the parameters outside of Prefect: - If you want to pass host in constructor:

import fsspec
fs = fsspec.filesystem(
    "smb",
    host="<http://pool1.vast.salk.edu|pool1.vast.salk.edu>",
    username="DOMAIN\\username",
    password="your-password",
    port=445,
)
fs.makedirs("than/prefect-results/prefect-test", exist_ok=True)
print(<http://fs.ls|fs.ls>("than/prefect-results"))

- If you prefer using the full URL without host in constructor:

import fsspec
fs = fsspec.filesystem("smb", username="DOMAIN\\username", password="your-password", port=445)
<http://fs.ls|fs.ls>("<smb://pool1.vast.salk.edu/than/prefect-results>")

Tips - Do not include “smb://” in host fields. Only use it in a URL-style basepath. - Make sure the first segment of “share_path” really is the SMB share name (“than” in my examples). If your share is actually “talmo”, use “talmo/than/prefect-results” only if that’s truly the share + subfolder. - On Windows, domain creds often need “DOMAIN\\username”. - Ensure both machines have the SMB dependencies installed. A common set is:

pip install smbprotocol smbfs

If you can paste your current block configuration (redact username/password), I’ll validate the exact values.