Great question — and just to confirm, I’m assuming you’re on Prefect 3.x. If not, let me know. In Prefect 3.x, the alternative is: - Python:

prefect.deployments.run_deployment("<FLOW_NAME>/<DEPLOYMENT_NAME>")

- CLI:

prefect deployment run "<FLOW_NAME>/<DEPLOYMENT_NAME>"

Both avoid the deployment_id and work off the deployment’s full name. If you only know the flow name and the work pool (but not the deployment id or name), there isn’t a single call that runs “by flow+pool”. You’ll need to resolve the deployment first, then trigger it. Two ways to do that: - CLI 1) List deployments for the flow:

prefect deployment ls --flow-name "<FLOW_NAME>"

2) Pick the one whose work pool matches, then run:

prefect deployment run "<FLOW_NAME>/<DEPLOYMENT_NAME>"

- Python (resolve by flow + pool, then run)

import asyncio
  from prefect.client.orchestration import get_client
  from prefect.deployments import run_deployment

  async def run_by_flow_and_pool(flow_name: str, pool_name: str, **kwargs):
      async with get_client() as client:
          # fetch all deployments; filter by flow/pool locally
          deployments = await client.read_deployments()
          matches = [
              d for d in deployments
              if getattr(d, "work_pool_name", None) == pool_name
              and getattr(d, "flow_name", None) == flow_name
          ]
          if not matches:
              raise ValueError(f"No deployment found for flow={flow_name} in pool={pool_name}")
          if len(matches) > 1:
              # disambiguate however you prefer
              # e.g., choose a specific deployment name or error out
              raise ValueError(f"Multiple deployments match flow={flow_name} in pool={pool_name}: {[d.name for d in matches]}")

          dep = matches[0]
          # Use the full name rather than the id
          full_name = f"{flow_name}/{dep.name}"
          return await run_deployment(
              name=full_name,
              **kwargs,  # parameters=..., flow_run_name=..., tags=..., etc.
          )

  asyncio.run(run_by_flow_and_pool("my-flow", "my-pool"))

Notes: - You can’t direct a run to a different work pool than the one set on the deployment; you must pick a deployment that already targets that pool. - If you do know the deployment name, you can skip the lookup and just call

run_deployment("<FLOW_NAME>/<DEPLOYMENT_NAME>")

. If you’re actually on Prefect 2.x, the same approach applies, but I’d still recommend moving to

prefect.deployments.run_deployment

over

client.create_flow_run_from_deployment

. Want me to tailor the snippet to your exact version and show a filter-based lookup (server-side) instead of local filtering?

Got it — if it’s a serving, that explains the confusion. Key points: - Serving deployments do not use work pools;

work_pool_name

will be None. - To trigger a run, you still need the deployment’s full name

<FLOW_NAME>/<DEPLOYMENT_NAME>

(or the deployment_id). - For a serving, the deployment does have a name (often auto-generated if you didn’t set one). In your printout it’s

1st-serving-of-my-spacemonitor

. Two quick ways forward: A) If you already have the DeploymentResponse (like the one you pasted) - Build the full name from the flow and deployment, then run it.

import asyncio
from prefect.client.orchestration import get_client
from prefect.deployments import run_deployment

async def run_serving_from_deployment(deployment, parameters=None, flow_run_name=None):
    async with get_client() as client:
        flow = await client.read_flow(deployment.flow_id)
        full_name = f"{flow.name}/{deployment.name}"
        return await run_deployment(
            name=full_name,
            parameters=parameters,
            flow_run_name=flow_run_name,
        )

# example usage with your deployment object
# asyncio.run(run_serving_from_deployment(deployment, parameters={"dates": [...] }))

B) If you only know the flow name (and it’s a serving) - Resolve the serving deployment for that flow, then run it.

import asyncio
from prefect.client.orchestration import get_client
from prefect.deployments import run_deployment

async def run_serving_by_flow(flow_name: str, parameters=None, flow_run_name=None):
    async with get_client() as client:
        deployments = await client.read_deployments()
        # Serving deployments have no work pool
        candidates = []
        for d in deployments:
            if getattr(d, "work_pool_name", None) is None:
                flow = await client.read_flow(d.flow_id)
                if flow.name == flow_name:
                    candidates.append((flow.name, d.name))

        if not candidates:
            raise RuntimeError(f"No serving deployment found for flow '{flow_name}'")
        if len(candidates) > 1:
            raise RuntimeError(f"Multiple serving deployments found for flow '{flow_name}': {[f'{f}/{n}' for f, n in candidates]}")

        full_name = f"{candidates[0][0]}/{candidates[0][1]}"
        return await run_deployment(
            name=full_name,
            parameters=parameters,
            flow_run_name=flow_run_name,
        )

# asyncio.run(run_serving_by_flow("space_weather_monitoring", parameters={"dates": [...] }))

CLI option: - Once you know the full name, you can trigger it with:

prefect deployment run "flow-name/1st-serving-of-my-spacemonitor"

Important note for serving: - The serving process must be running; it will pick up and execute runs for that deployment. Since there’s no work pool/worker, the run will stay pending if the serving process isn’t alive. If you’d like, I can tailor a snippet to how you created the serving (e.g.,

flow.serve(name=...)

) so it finds and runs the right one automatically.

You’re seeing

work_pool_name=None

because it’s a serving deployment — that’s expected. Servings do not use work pools, so there’s nothing to match on there. What to use instead of `client.create_flow_run_from_deployment()`: - Python: use

prefect.deployments.run_deployment("<FLOW_NAME>/<DEPLOYMENT_NAME>")

- CLI:

prefect deployment run "<FLOW_NAME>/<DEPLOYMENT_NAME>"

In your object, the deployment does have a name:

1st-serving-of-my-spacemonitor

. So the full name is:

<your-flow-name>/1st-serving-of-my-spacemonitor

Then:

from prefect.deployments import run_deployment
import asyncio

async def go():
    await run_deployment(
        name="space_weather_monitoring/1st-serving-of-my-spacemonitor",
        parameters={"dates": ["2025-12-01", "2025-12-02"]},  # example
        flow_run_name="manual-trigger"
    )

asyncio.run(go())

If you only know the flow name (and it’s a serving), you can resolve the serving deployment by finding the one with

work_pool_name is None

and the matching flow, then run it:

import asyncio
from prefect.client.orchestration import get_client
from prefect.deployments import run_deployment

async def run_serving_by_flow(flow_name: str, parameters=None, flow_run_name=None):
    async with get_client() as client:
        deployments = await client.read_deployments()
        candidates = []
        for d in deployments:
            if getattr(d, "work_pool_name", None) is None:
                flow = await client.read_flow(d.flow_id)
                if flow.name == flow_name:
                    candidates.append((flow.name, d.name))

        if not candidates:
            raise RuntimeError(f"No serving deployment found for flow '{flow_name}'")
        if len(candidates) > 1:
            raise RuntimeError(f"Multiple serving deployments for '{flow_name}': {[f'{f}/{n}' for f, n in candidates]}")

        full_name = f"{candidates[0][0]}/{candidates[0][1]}"
        return await run_deployment(
            name=full_name,
            parameters=parameters,
            flow_run_name=flow_run_name,
        )

# asyncio.run(run_serving_by_flow("space_weather_monitoring", parameters={"dates": [...] }))

CLI to discover and run:

prefect deployment ls --flow-name "space_weather_monitoring"
prefect deployment run "space_weather_monitoring/1st-serving-of-my-spacemonitor"

If your actual goal is to target a specific work pool, you’ll need a standard deployment (not serving). Servings ignore pools by design and must be executed by the serving process that’s running.

Short answer: -

run_deployment(...)

returns a FlowRun object, not your flow’s return value. - You can get the run’s final status by polling its state. - To get the actual return value, you must persist results; then resolve the state’s result. Here’s a drop-in helper that triggers, waits for a terminal state, and returns both status and (if available) the flow’s return value:

import asyncio
from prefect.client.orchestration import get_client
from prefect.deployments import run_deployment

async def trigger_and_wait(full_name: str, parameters: dict | None = None, poll_seconds: float = 2.0):
    # create the run
    flow_run = await run_deployment(name=full_name, parameters=parameters)

    # poll until terminal
    async with get_client() as client:
        terminal = {"COMPLETED", "FAILED", "CRASHED", "CANCELLED"}
        while True:
            fr = await client.read_flow_run(flow_run.id)
            st = fr.state
            if st and st.type.value in terminal:
                break
            await asyncio.sleep(poll_seconds)

        status = st.type.value
        state_name = st.name

        # Try to fetch the flow's return value (requires persisted results)
        result_value = None
        try:
            result_value = await st.result()
        except Exception:
            # No persisted result or resolution failed; leave as None
            pass

        return {
            "id": str(fr.id),
            "status": status,
            "state_name": state_name,
            "result": result_value,
        }

Usage with your pattern:

result = asyncio.run(
    trigger_and_wait(
        full_name="space_weather_monitoring/1st-serving-of-my-spacemonitor",
        parameters=parameters
    )
)
print(result)
return 'application/json', result

Notes: - If you don’t want to block, skip the polling and just return the created run info (status “SUBMITTED”/“SCHEDULED”). You can always check later. - To actually get the flow’s return value, configure your flow to persist results (e.g., set

persist_result=True

on the

@flow

and configure result storage). Otherwise

state.result()

won’t have data to load. - For servings, there’s no work pool. Just use the full deployment name as shown above, and ensure the serving process is running so the run can execute.

Right —

run_by_flow_and_pool(...)

is returning a Prefect FlowRun object, which isn’t JSON-serializable. Build a JSON-safe payload instead, and optionally resolve the flow’s return value. Here’s a drop-in that: - creates the run, - waits for a terminal state, - returns a JSON-serializable dict with id, status, state_name, and result (if persisted).

import asyncio
import datetime as dt
import uuid
import decimal
from prefect.client.orchestration import get_client
from prefect.deployments import run_deployment

def to_jsonable(obj):
    if obj is None or isinstance(obj, (str, int, float, bool)):
        return obj
    if isinstance(obj, (dt.datetime, dt.date)):
        return obj.isoformat()
    if isinstance(obj, uuid.UUID):
        return str(obj)
    if isinstance(obj, decimal.Decimal):
        return float(obj)
    if isinstance(obj, bytes):
        try:
            return obj.decode()
        except Exception:
            return repr(obj)
    if isinstance(obj, dict):
        return {str(k): to_jsonable(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple, set)):
        return [to_jsonable(x) for x in obj]
    # Try Pydantic v2 models
    try:
        return obj.model_dump(mode="json")
    except Exception:
        return repr(obj)

async def trigger_and_wait_json(full_name: str, parameters: dict | None = None, poll_seconds: float = 2.0):
    # Create the run
    created = await run_deployment(name=full_name, parameters=parameters)

    # Poll to terminal
    async with get_client() as client:
        terminal = {"COMPLETED", "FAILED", "CRASHED", "CANCELLED"}
        while True:
            fr = await client.read_flow_run(created.id)
            st = fr.state
            if st and st.type.value in terminal:
                break
            await asyncio.sleep(poll_seconds)

        payload = {
            "id": str(fr.id),
            "status": st.type.value,
            "state_name": st.name,
            "created": fr.created.isoformat() if fr.created else None,
            "start_time": fr.start_time.isoformat() if fr.start_time else None,
            "end_time": fr.end_time.isoformat() if fr.end_time else None,
        }

        # Optional: resolve the flow's return value (requires persisted results)
        try:
            value = await st.result()
            payload["result"] = to_jsonable(value)
        except Exception:
            payload["result"] = None

        return payload

Use it in your code:

result = asyncio.run(
    trigger_and_wait_json(
        full_name="space_weather_monitoring/1st-serving-of-my-spacemonitor",
        parameters=parameters,
    )
)
print(result)
return 'application/json', result

Notes: - To actually get the flow’s return value, ensure the flow persists results (e.g.,

@flow(persist_result=True)

and configure result storage). Otherwise

state.result()

may not return data. - If you don’t want to block, skip the polling loop and just return the created run info with a “SCHEDULED/SUBMITTED” status.

Yes, you can access a flow run’s return value from the client — but only if the flow’s result is persisted. What the “return value” is - It’s whatever your

@flow

function returns. - That value is attached to the run’s terminal State. To retrieve it later (from another process), the result must be persisted. Enable result persistence on your flow - Add

persist_result=True

to your

@flow

. For larger/complex objects, also configure result storage/serializer.

from prefect import flow

@flow(name="space_weather_monitoring", persist_result=True)
def space_weather_monitoring(...):
    ...
    return {"ok": True, "count": 42}

Fetch status and return value via the client - Poll the run until it’s terminal, then resolve the state’s result.

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.objects import StateType

async def get_status_and_result(run_id):
    async with get_client() as client:
        terminal = {StateType.COMPLETED, StateType.FAILED, StateType.CANCELLED, StateType.CRASHED}
        while True:
            fr = await client.read_flow_run(run_id)
            st = fr.state
            if st and st.type in terminal:
                break
            await asyncio.sleep(2)

        status = st.type.value
        state_name = st.name

        # Try to load the flow's return value (works if persisted)
        result_value = None
        try:
            result_value = await st.result()
        except Exception:
            # Not persisted or cannot be deserialized
            result_value = None

        return {
            "id": str(fr.id),
            "status": status,
            "state_name": state_name,
            "result": result_value,
        }

Typical flow: 1) Trigger a run (e.g., via

run_deployment(...)

) and capture

flow_run.id

. 2) Call

get_status_and_result(flow_run.id)

to get status and the return value. If you’re not seeing a result: - Ensure

@flow(persist_result=True)

is set. - For big/complex data, configure result storage/serializer; otherwise Prefect may not be able to persist/restore it. - If the run failed/crashed,

await st.result()

will raise; catch exceptions as shown.

To persist (and later retrieve) a flow’s return value in Prefect 3.x, do the following: 1) Enable result persistence on the flow - Set

persist_result=True

on your

@flow

decorator. 2) Choose where to store results (result_storage) - Use a filesystem/storage block (local or remote). - Example: Local filesystem block

from prefect.filesystems import LocalFileSystem

# one-time setup (can be done in a REPL or init script)
LocalFileSystem(basepath="/tmp/prefect-results").save("local-results", overwrite=True)

3) Choose how to serialize results (result_serializer) - Use

JSONSerializer()

for JSON-serializable objects. - Use

PickleSerializer()

for arbitrary Python objects (note: consider security/trust implications).

from prefect import flow
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JSONSerializer  # or PickleSerializer

@flow(
    name="space_weather_monitoring",
    persist_result=True,
    result_storage=LocalFileSystem.load("local-results"),
    result_serializer=JSONSerializer(),  # or PickleSerializer()
)
def space_weather_monitoring(...):
    # return any value; it will be persisted using the serializer
    return {"ok": True, "count": 42}

4) Redeploy or restart serving - For a deployment: re-deploy so the new flow settings are used. - For a serving: restart the serving process so it imports the updated flow. 5) Retrieve the result from the client - After the run reaches a terminal state, resolve the state’s result; Prefect will fetch and deserialize it from your result storage.

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.objects import StateType

async def get_status_and_result(run_id):
    async with get_client() as client:
        terminal = {StateType.COMPLETED, StateType.FAILED, StateType.CANCELLED, StateType.CRASHED}
        while True:
            fr = await client.read_flow_run(run_id)
            st = fr.state
            if st and st.type in terminal:
                break
            await asyncio.sleep(2)

        status = st.type.value
        result_value = None
        try:
            result_value = await st.result()  # downloads + deserializes from result_storage
        except Exception:
            result_value = None

        return {"id": str(fr.id), "status": status, "result": result_value}

Notes and tips: - JSON vs Pickle: Use JSON for simple/portable data. Use Pickle for complex Python objects (only in trusted environments). - Large results: Consider compressed serializers (e.g.,

CompressedJSONSerializer

,

CompressedPickleSerializer

) or writing large data to external storage yourself and returning a small reference. - Tasks too: You can also persist task results with

@task(persist_result=True, ...)

if you need intermediate values later. - Remote storage: For S3/GCS/Azure, create and use the corresponding storage block (requires the relevant integration package) and pass it as

result_storage=...

. If you share what kind of data your flow returns and where you’d like it stored (local, S3, GCS, etc.), I can give you a tailored snippet.