Great — here’s a minimal, end-to-end example in Prefect 3.x that: - defines a flow that adds two numbers (with parameters) - creates a small test deployment - runs that deployment from the Python SDK - pulls the result back from the flow run I’ll show the simplest local “serve” approach first (no workers needed), then an optional “deploy to a work pool” variant. Assumption: Prefect 3.x. Option A: simplest local test with serve (no worker required) 1) Define the flow (sum_flow.py)

from prefect import flow

@flow(name="adder-flow", log_prints=True)
def add(x: int, y: int) -> int:
    total = x + y
    print(f"{x} + {y} = {total}")
    return total

2) Start a local deployment runner (serve_adder.py) Run this in Terminal 1; it will create a deployment and start listening for work.

from sum_flow import add

if __name__ == "__main__":
    # Creates a deployment and starts a local runner that will execute runs
    add.serve(
        name="adder-deployment",
        parameters={"x": 1, "y": 2},  # default parameters (can be overridden)
    )

3) Trigger the deployment and get the result (run_adder.py) Run this in Terminal 2 while Terminal 1 is still running.

from prefect.deployments import run_deployment

# Format: "flow_name/deployment_name"
flow_run = run_deployment(
    name="adder-flow/adder-deployment",
    parameters={"x": 40, "y": 2},   # override defaults
    timeout=120,                    # wait up to 2 minutes for completion
    poll_interval=1,                # poll every second
)

print(f"Final state: {flow_run.state.type}")
if flow_run.state.is_final():
    # This returns the actual return value of the flow (40 + 2 = 42)
    result_value = flow_run.state.result()
    print(f"Result: {result_value}")

That’s it. Terminal 2 should print “Result: 42” once the run completes. Option B: deploy to a work pool (requires a worker) If you want a true deployment picked up by a worker: 1) Define the same flow (sum_flow.py) as above. 2) Create a deployment in a work pool (deploy_adder.py) You must have a work pool and a running worker for that pool.

from sum_flow import add

if __name__ == "__main__":
    deployment_id = add.deploy(
        name="adder-deployment",
        work_pool_name="my-process-pool",  # change to your pool name
        parameters={"x": 1, "y": 2},
        build=False,  # set True if you’re building an image
        push=False,
    )
    print(f"Created deployment: {deployment_id}")

3) Ensure a worker is running for that work pool (outside Python SDK; however you run workers in your environment). 4) Trigger and get the result (same as Option A)

from prefect.deployments import run_deployment

flow_run = run_deployment(
    name="adder-flow/adder-deployment",
    parameters={"x": 5, "y": 7},
    timeout=120,
    poll_interval=1,
)
print(f"Final state: {flow_run.state.type}")
if flow_run.state.is_final():
    print(f"Result: {flow_run.state.result()}")

Notes - The

name

parameter to

run_deployment

is "flow_name/deployment_name". Here it’s "adder-flow/adder-deployment". -

run_deployment(...)

returns a FlowRun with a

state

. After it reaches a final state (COMPLETED/FAILED/etc.), call

flow_run.state.result()

to get the flow’s return value. - If the run fails,

state.result()

will raise — you can use

state.result(raise_on_failure=False)

to get the exception object instead. If you prefer everything in a single script, I can show a self-contained example that spins up a serve runner in one process/thread and triggers it programmatically. Would you like that variant, or are the two-file steps above okay for your use case?

Thanks for the traceback — that’s a classic “result not persisted” case. What happened -

run_deployment(...)

returns a FlowRun loaded from the API. - Calling

flow_run.state.result()

needs the flow’s return value to be persisted somewhere. - By default, flows do not persist results, so you see

prefect.exceptions.MissingResult

. Fix (local/serve or same-machine worker) - Enable result persistence on the flow and use a JSON serializer (your return is just an int):

from prefect import flow

@flow(name="adder-flow", log_prints=True, persist_result=True, result_serializer="json")
def add(x: int, y: int) -> int:
    total = x + y
    print(f"{x} + {y} = {total}")
    return total

- Recreate/restart your deployment runner (serve) or redeploy if you’re using a work pool. - Re-run your

run_adder.py

. Now this will work:

from prefect.deployments import run_deployment

flow_run = run_deployment(
    name="adder-flow/adder-deployment",
    parameters={"x": 40, "y": 2},
    timeout=120,
    poll_interval=1,
)

print(f"Final state: {flow_run.state.type}")
print(f"Result: {flow_run.state.result()}")

If you’re using remote workers - If your worker runs on a different machine or in a container, the default local result store on that worker won’t be accessible to your client. - In that case, still set

persist_result=True

, and configure a remote result store (e.g., S3/GCS) so the client can fetch results from a shared location. I can share a short S3 example if that’s your setup. Quick questions to ensure this is solid for you - Are you running with

add.serve(...)

locally, or via a work pool/worker? - If via a worker, is the worker on the same machine as where you’re calling

run_deployment

? Once I know that, I can tailor the result storage recommendation (local vs S3/GCS) and give you the exact snippet.

Great question. In a multi-Docker-worker setup, the key is to persist flow results to shared, network-accessible storage. Then any client that calls run_deployment can fetch the return value via state.result(). What you need to do - Choose a shared result store (S3, GCS, Azure, MinIO, or a network filesystem). S3/MinIO is common for Docker workers. - Configure your flow to persist results to that store (via a Block). - Ensure your workers and the client machine have credentials/access to that store. - Trigger the deployment and call state.result() to fetch the return. Minimal S3/MinIO example 1) Create a RemoteFileSystem block (one-time) Run this once (locally or in a setup script) to register a block that points to your bucket/prefix. For MinIO, include endpoint_url and credentials; for AWS S3 with IAM, basepath is enough and credentials come from env/instance role.

from prefect.filesystems import RemoteFileSystem

# For AWS S3 (credentials via env/instance profile)
s3_storage = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results/>")
s3_storage.save("prod-results", overwrite=True)

# For MinIO (S3-compatible)
# minio_storage = RemoteFileSystem(
#     basepath="<s3://my-bucket/prefect-results/>",
#     settings={
#         "endpoint_url": "<http://minio:9000>",
#         "key": "minioadmin",
#         "secret": "minioadmin",
#         "use_ssl": False,
#     },
# )
# minio_storage.save("prod-results", overwrite=True)

2) Update your flow to persist results to that block Use the block slug, enable persistence, and pick a serializer. JSON works well for simple return types; use pickle for complex Python objects.

from prefect import flow

@flow(
    name="adder-flow",
    result_storage="remote-file-system/prod-results",  # the block you saved above
    result_serializer="json",
    persist_result=True,
    log_prints=True,
)
def add(x: int, y: int) -> int:
    total = x + y
    print(f"{x} + {y} = {total}")
    return total

3) Deploy to your Docker work pool (workers must be running)

# Run once during CI/CD or locally to create/update the deployment
deployment_id = add.deploy(
    name="adder",
    work_pool_name="docker-pool",  # your Docker work pool name
    parameters={"x": 1, "y": 2},   # defaults (can be overridden at run time)
    build=False,
    push=False,
)
print(deployment_id)

4) Trigger the deployment and fetch the return value from anywhere As long as the client has access to the same S3/MinIO credentials, Prefect will fetch from that storage when you call state.result().

from prefect.deployments import run_deployment

flow_run = run_deployment(
    name="adder-flow/adder",
    parameters={"x": 40, "y": 2},
    timeout=180,
    poll_interval=2,
)

print(f"Final state: {flow_run.state.type}")
print(f"Result: {flow_run.state.result()}")  # -> 42

Global defaults (optional) If you want every flow/task to persist to the same storage without modifying decorators:

# Enable persistence and set a default result storage block
prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK='remote-file-system/prod-results'
prefect config set PREFECT_RESULTS_DEFAULT_SERIALIZER='json'

Important notes - Workers and the client must be able to reach the same storage. For S3/MinIO, ensure AWS/endpoint credentials are available in both environments (env vars, IAM roles, or config). - If you use local or container-only storage, clients on other machines won’t be able to fetch results and you’ll hit MissingResult. - For large/complex Python returns, use pickle or compressed/pickle serializers instead of json. Want me to tailor this to your exact storage? Let me know if you’ll use AWS S3, GCS, Azure, or MinIO, and I’ll share a ready-to-run snippet for that backend (including recommended env vars for your Docker workers).

Short answer: for production with multiple Docker workers and an LLM orchestrator, I recommend using Prefect’s built‑in result persistence with a shared remote store (S3/MinIO/GCS), then have your orchestrator call run_deployment(...) and read state.result(). Only add a custom “write to DB” step if you need extra indexing/joins beyond what Prefect already gives you. Why this is usually the better choice - Simpler, less code to maintain: no per-deployment “returnresults” plumbing. - Works across any number of Docker workers as long as they and your client share access to the same result storage. - Strong observability and a single source of truth in Prefect. - You can still add artifacts or a DB sink later for specialized querying. Recommended production pattern 1) Configure a shared result store - Create a RemoteFileSystem block pointing to S3/MinIO/GCS and give your workers and client access to it.

from prefect.filesystems import RemoteFileSystem

# Example: S3 (use MinIO by adding endpoint_url/key/secret)
s3_storage = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results/>")
s3_storage.save("prod-results", overwrite=True)

2) Persist results in your flow - Use the block slug and a serializer. JSON for simple values; pickle if you return complex Python objects.

from prefect import flow

@flow(
    name="adder-flow",
    result_storage="remote-file-system/prod-results",
    result_serializer="json",
    persist_result=True,
    log_prints=True,
)
def add(x: int, y: int) -> int:
    return x + y

3) Deploy to your Docker work pool

deployment_id = add.deploy(
    name="adder",
    work_pool_name="docker-pool",
    parameters={"x": 1, "y": 2},
    build=False,
    push=False,
)

4) Orchestrator runs and fetches the result - Your LLM controller (Python SDK) triggers the deployment with parameters, waits, and returns the value.

from prefect.deployments import run_deployment

flow_run = run_deployment(
    name="adder-flow/adder",
    parameters={"x": 40, "y": 2},
    timeout=300,
    poll_interval=2,
    # optional dedup if your LLM may retry:
    # idempotency_key="some-stable-key"
)

if flow_run.state.is_final():
    result = flow_run.state.result()  # -> 42
    # return this to your LLM

When a DB sink makes sense - You need cross-run joins/analytics not convenient via the Prefect API. - You need to transform/normalize results into a domain model. - You want push-based delivery to a downstream service (e.g., webhook-style) without the orchestrator polling. If you choose a DB sink, make it reusable and optional - Implement a single, shared helper task used by flows instead of duplicating code in each deployment. - Keep Prefect result persistence enabled as the canonical fallback. Example: reusable DB reporter task

from prefect import flow, task

@task
def write_result_to_db(run_id: str, result: dict):
    # upsert into your DB with run_id as key
    ...

@flow(
    result_storage="remote-file-system/prod-results",
    result_serializer="json",
    persist_result=True,
)
def my_flow(x: int, y: int) -> dict:
    out = {"sum": x + y}
    # best-effort: write to DB, but still return and persist to remote storage
    write_result_to_db(
        run_id="{{ flow_run.id }}",  # can also use prefect.runtime for context
        result=out,
    )
    return out

Extra options you may find useful - Artifacts: publish structured or human-readable outputs (tables/markdown) that you can fetch via the Prefect API without owning a DB. For example:

from prefect import flow
from prefect.artifacts import create_table_artifact

@flow
def report_flow():
    rows = [{"k": "result", "v": 42}]
    create_table_artifact(key="run-output", table=rows, description="LLM-ready output")
    return 42

- Idempotency: pass idempotency_key to run_deployment when your LLM may re-issue the same request.

- Large results: rather than returning a huge payload, return a small “envelope” with a URI/pointer to your object store. Cheaper, faster to fetch. My take - Default: use Prefect result persistence + run_deployment + state.result() for your LLM. It’s the cleanest, least error-prone, and scales with many Docker workers. - Add DB/Artifacts: only if you need external querying or push-based delivery. If you do, keep it as a single reusable helper and still persist results with Prefect as the source of truth. If you tell me which backend you’ll use (S3, MinIO, GCS, Azure) and where your LLM agent runs, I’ll share an exact, ready-to-run snippet with recommended environment variables for your Docker workers and the orchestrator.