Repository: PrefectHQ/prefect
Pull request: Integrate flow run input and schema/response mechanics into pause/suspend
This integrates flow run input into pause/suspend allowing a flow run to pause or suspend and receive structured and validated input, it does this by:

- Creating `RunInput` a sub-classable pydantic model that a user can subclass and add their own fields and validation.
- Adding `wait_for_input` to `pause_flow_run` and `suspend_flow_run`. When either method is provided a subclass of `RunInput` it will pause/suspend,'save` and the schema of the class to the flow run input API. When resuming it will then try to grab a response from the flow run input API and instantiate the class with that data.

Closes #11339 

### Example
```python
import random
from prefect import flow, get_run_logger, pause_flow_run

from prefect.input import RunInput


class UserGuess(RunInput):
    guess: int


@flow
async def guess_the_number():
    the_number = random.randint(1, 25)

    logger = get_run_logger()

    logger.info("Guess the number between 1 and 25. You have 5 guesses.")

    for i in range(5):
        input = await pause_flow_run(
            wait_for_input=UserGuess.with_initial_data(guess=0)
        )
        if input.guess == the_number:
            logger.info("You guessed it!")
            break
        elif input.guess < the_number:
            logger.info("Too low!")
        elif input.guess > the_number:
            logger.info("Too high!")

    logger.error(f"You failed to guess the number. It was {the_number}")


if __name__ == "__main__":
    guess_the_number.serve(name="guess-the-number")
```

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [x] This pull request references any related issue by including "closes `<link to issue>`"
	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] This pull request includes tests or only affects documentation.
- [x] This pull request includes a label categorizing the change e.g. `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed


Problem statement:
Build out client-side primitives for getting, storing, and validating flow run input
Now that we have modeling and an CRUD API to store key/value data associated with a flow run we should build out the client-side primitives to get, store, and validate that data.

Hints from issue discussion:

Commit message: Fix suspend test with postgres

Patch to review:
diff --git a/src/prefect/_internal/pydantic/v2_schema.py b/src/prefect/_internal/pydantic/v2_schema.py
index cdd73f5a494b..1496258ae461 100644
--- a/src/prefect/_internal/pydantic/v2_schema.py
+++ b/src/prefect/_internal/pydantic/v2_schema.py
@@ -82,11 +82,18 @@ def process_v2_params(
     return name, type_, field
 
 
-def create_v2_schema(name_: str, model_cfg: ConfigDict, **model_fields):
+def create_v2_schema(
+    name_: str,
+    model_cfg: t.Optional[ConfigDict] = None,
+    model_base: t.Optional[t.Type[V2BaseModel]] = None,
+    **model_fields
+):
     """
     Create a pydantic v2 model and craft a v1 compatible schema from it.
     """
-    model = create_model(name_, __config__=model_cfg, **model_fields)
+    model = create_model(
+        name_, __config__=model_cfg, __base__=model_base, **model_fields
+    )
     adapter = TypeAdapter(model)
 
     # root model references under #definitions
diff --git a/src/prefect/client/orchestration.py b/src/prefect/client/orchestration.py
index 22ad6efb378d..7373eca7fa75 100644
--- a/src/prefect/client/orchestration.py
+++ b/src/prefect/client/orchestration.py
@@ -80,6 +80,7 @@
     Constant,
     Deployment,
     Flow,
+    FlowRunInput,
     FlowRunNotificationPolicy,
     FlowRunPolicy,
     Log,
@@ -2688,6 +2689,48 @@ async def release_concurrency_slots(
             },
         )
 
+    async def create_flow_run_input(self, flow_run_id: UUID, key: str, value: str):
+        """
+        Creates a flow run input.
+
+        Args:
+            flow_run_id: The flow run id.
+            key: The input key.
+            value: The input value.
+        """
+
+        # Initialize the input to ensure that the key is valid.
+        FlowRunInput(flow_run_id=flow_run_id, key=key, value=value)
+
+        response = await self._client.post(
+            f"/flow_runs/{flow_run_id}/input",
+            json={"key": key, "value": value},
+        )
+        response.raise_for_status()
+
+    async def read_flow_run_input(self, flow_run_id: UUID, key: str) -> str:
+        """
+        Reads a flow run input.
+
+        Args:
+            flow_run_id: The flow run id.
+            key: The input key.
+        """
+        response = await self._client.get(f"/flow_runs/{flow_run_id}/input/{key}")
+        response.raise_for_status()
+        return response.content.decode()
+
+    async def delete_flow_run_input(self, flow_run_id: UUID, key: str):
+        """
+        Deletes a flow run input.
+
+        Args:
+            flow_run_id: The flow run id.
+            key: The input key.
+        """
+        response = await self._client.delete(f"/flow_runs/{flow_run_id}/input/{key}")
+        response.raise_for_status()
+
     async def __aenter__(self):
         """
         Start the client.
diff --git a/src/prefect/client/schemas/objects.py b/src/prefect/client/schemas/objects.py
index 9f3edc439e1a..fa31fa4e9bad 100644
--- a/src/prefect/client/schemas/objects.py
+++ b/src/prefect/client/schemas/objects.py
@@ -25,7 +25,10 @@
 
 from prefect._internal.schemas.bases import ObjectBaseModel, PrefectBaseModel
 from prefect._internal.schemas.fields import CreatedBy, DateTimeTZ, UpdatedBy
-from prefect._internal.schemas.validators import raise_on_name_with_banned_characters
+from prefect._internal.schemas.validators import (
+    raise_on_name_alphanumeric_dashes_only,
+    raise_on_name_with_banned_characters,
+)
 from prefect.client.schemas.schedules import SCHEDULE_TYPES
 from prefect.settings import PREFECT_CLOUD_API_URL
 from prefect.utilities.collections import AutoEnum, listrepr
@@ -105,6 +108,7 @@ class StateDetails(PrefectBaseModel):
     pause_timeout: DateTimeTZ = None
     pause_reschedule: bool = False
     pause_key: str = None
+    run_input_keyset: Optional[Dict[str, str]] = None
     refresh_cache: bool = None
 
 
@@ -1523,3 +1527,14 @@ class Variable(ObjectBaseModel):
         description="A list of variable tags",
         example=["tag-1", "tag-2"],
     )
+
+
+class FlowRunInput(ObjectBaseModel):
+    flow_run_id: UUID = Field(description="The flow run ID associated with the input.")
+    key: str = Field(description="The key of the input.")
+    value: str = Field(description="The value of the input.")
+
+    @validator("key", check_fields=False)
+    def validate_name_characters(cls, v):
+        raise_on_name_alphanumeric_dashes_only(v)
+        return v
diff --git a/src/prefect/engine.py b/src/prefect/engine.py
index 2935fcfbcacd..b6573eeb5183 100644
--- a/src/prefect/engine.py
+++ b/src/prefect/engine.py
@@ -100,6 +100,7 @@
     List,
     Optional,
     Set,
+    Type,
     TypeVar,
     Union,
 )
@@ -114,6 +115,7 @@
 import prefect.context
 import prefect.plugins
 from prefect._internal.compatibility.deprecated import deprecated_parameter
+from prefect._internal.compatibility.experimental import experimental_parameter
 from prefect._internal.concurrency.api import create_call, from_async, from_sync
 from prefect._internal.concurrency.calls import get_current_call
 from prefect._internal.concurrency.cancellation import CancelledError, get_deadline
@@ -152,6 +154,7 @@
 )
 from prefect.flows import Flow
 from prefect.futures import PrefectFuture, call_repr, resolve_futures_to_states
+from prefect.input import RunInput, keyset_from_paused_state
 from prefect.logging.configuration import setup_logging
 from prefect.logging.handlers import APILogHandler
 from prefect.logging.loggers import (
@@ -953,12 +956,16 @@ async def orchestrate_flow_run(
     when=lambda p: p is True,
     help="Use `suspend_flow_run` instead.",
 )
+@experimental_parameter(
+    "wait_for_input", group="flow_run_input", when=lambda y: y is not None
+)
 async def pause_flow_run(
     flow_run_id: UUID = None,
     timeout: int = 300,
     poll_interval: int = 10,
     reschedule: bool = False,
     key: str = None,
+    wait_for_input: Optional[Type[RunInput]] = None,
 ):
     """
     Pauses the current flow run by blocking execution until resumed.
@@ -990,8 +997,12 @@ async def pause_flow_run(
             the number of pauses observed by the flow so far, and prevents pauses that
             use the "reschedule" option from running the same pause twice. A custom key
             can be supplied for custom pausing behavior.
+        wait_
     """
     if flow_run_id:
+        if wait_for_input is not None:
+            raise RuntimeError("Cannot wait for input when pausing out of process.")
+
         return await _out_of_process_pause(
             flow_run_id=flow_run_id,
             timeout=timeout,
@@ -1000,18 +1011,26 @@ async def pause_flow_run(
         )
     else:
         return await _in_process_pause(
-            timeout=timeout, poll_interval=poll_interval, reschedule=reschedule, key=key
+            timeout=timeout,
+            poll_interval=poll_interval,
+            reschedule=reschedule,
+            key=key,
+            wait_for_input=wait_for_input,
         )
 
 
 @inject_client
+@experimental_parameter(
+    "wait_for_input", group="flow_run_input", when=lambda y: y is not None
+)
 async def _in_process_pause(
     timeout: int = 300,
     poll_interval: int = 10,
     reschedule=False,
     key: str = None,
     client=None,
-):
+    wait_for_input: Optional[Type[RunInput]] = None,
+) -> Optional[RunInput]:
     if TaskRunContext.get():
         raise RuntimeError("Cannot pause task runs.")
 
@@ -1026,12 +1045,19 @@ async def _in_process_pause(
 
     logger.info("Pausing flow, execution will continue when this flow run is resumed.")
 
+    proposed_state = Paused(
+        timeout_seconds=timeout, reschedule=reschedule, pause_key=pause_key
+    )
+
+    run_input_keyset = keyset_from_paused_state(proposed_state)
+
+    if wait_for_input:
+        proposed_state.state_details.run_input_keyset = run_input_keyset
+
     try:
         state = await propose_state(
             client=client,
-            state=Paused(
-                timeout_seconds=timeout, reschedule=reschedule, pause_key=pause_key
-            ),
+            state=proposed_state,
             flow_run_id=context.flow_run.id,
         )
     except Abort as exc:
@@ -1040,6 +1066,9 @@ async def _in_process_pause(
 
     if state.is_running():
         # The orchestrator requests that this pause be ignored
+        if wait_for_input:
+            await wait_for_input.load(run_input_keyset)
+
         return
 
     if not state.is_paused():
@@ -1048,6 +1077,9 @@ async def _in_process_pause(
             f"Flow run cannot be paused. Received non-paused state from API: {state}"
         )
 
+    if wait_for_input:
+        await wait_for_input.save(run_input_keyset)
+
     if reschedule:
         # If a rescheduled pause, exit this process so the run can be resubmitted later
         raise Pause(state=state)
@@ -1057,22 +1089,21 @@ async def _in_process_pause(
         # attempt to check if a flow has resumed at least once
         initial_sleep = min(timeout / 2, poll_interval)
         await anyio.sleep(initial_sleep)
-        flow_run = await client.read_flow_run(context.flow_run.id)
-        if flow_run.state.is_running():
-            logger.info("Resuming flow run execution!")
-            return
-
         while True:
-            await anyio.sleep(poll_interval)
             flow_run = await client.read_flow_run(context.flow_run.id)
             if flow_run.state.is_running():
                 logger.info("Resuming flow run execution!")
+                if wait_for_input:
+                    return await wait_for_input.load(run_input_keyset)
                 return
+            await anyio.sleep(poll_interval)
 
     # check one last time before failing the flow
     flow_run = await client.read_flow_run(context.flow_run.id)
     if flow_run.state.is_running():
         logger.info("Resuming flow run execution!")
+        if wait_for_input:
+            return await wait_for_input.load(run_input_keyset)
         return
 
     raise FlowPauseTimeout("Flow run was paused and never resumed.")
@@ -1107,6 +1138,7 @@ async def suspend_flow_run(
     timeout: Optional[int] = 300,
     key: Optional[str] = None,
     client: PrefectClient = None,
+    wait_for_input: Optional[Type[RunInput]] = None,
 ):
     """
     Suspends a flow run by stopping code execution until resumed.
@@ -1159,10 +1191,16 @@ async def suspend_flow_run(
         suspending_current_flow_run = False
         pause_key = key or str(uuid4())
 
+    proposed_state = Suspended(timeout_seconds=timeout, pause_key=pause_key)
+    run_input_keyset = keyset_from_paused_state(proposed_state)
+
+    if wait_for_input:
+        proposed_state.state_details.run_input_keyset = run_input_keyset
+
     try:
         state = await propose_state(
             client=client,
-            state=Suspended(timeout_seconds=timeout, pause_key=pause_key),
+            state=proposed_state,
             flow_run_id=flow_run_id,
         )
     except Abort as exc:
@@ -1170,7 +1208,10 @@ async def suspend_flow_run(
         raise RuntimeError(f"Flow run cannot be suspended: {exc}")
 
     if state.is_running():
-        # The orchestrator requests that this suspend be ignored
+        # The orchestrator requests that this suspend be ignored. Likely
+        # because the flow was previously suspended and is now being resumed.
+        if wait_for_input:
+            return await wait_for_input.load(run_input_keyset)
         return
 
     if not state.is_paused():
@@ -1179,6 +1220,9 @@ async def suspend_flow_run(
             f"Flow run cannot be suspended. Received unexpected state from API: {state}"
         )
 
+    if wait_for_input:
+        await wait_for_input.save(run_input_keyset)
+
     if suspending_current_flow_run:
         # Exit this process so the run can be resubmitted later
         raise Pause()
diff --git a/src/prefect/input/__init__.py b/src/prefect/input/__init__.py
new file mode 100644
index 000000000000..c881c9462b83
--- /dev/null
+++ b/src/prefect/input/__init__.py
@@ -0,0 +1,11 @@
+from .actions import create_flow_run_input, delete_flow_run_input, read_flow_run_input
+from .run_input import RunInput, keyset_from_base_key, keyset_from_paused_state
+
+__all__ = [
+    "RunInput",
+    "create_flow_run_input",
+    "keyset_from_base_key",
+    "keyset_from_paused_state",
+    "delete_flow_run_input",
+    "read_flow_run_input",
+]
diff --git a/src/prefect/input/actions.py b/src/prefect/input/actions.py
new file mode 100644
index 000000000000..7231b500b685
--- /dev/null
+++ b/src/prefect/input/actions.py
@@ -0,0 +1,88 @@
+from typing import TYPE_CHECKING, Any, Optional
+from uuid import UUID
+
+import orjson
+
+from prefect.client.utilities import inject_client
+from prefect.context import FlowRunContext
+from prefect.exceptions import PrefectHTTPStatusError
+from prefect.utilities.asyncutils import sync_compatible
+
+if TYPE_CHECKING:
+    from prefect.client.orchestration import PrefectClient
+
+
+def _ensure_flow_run_id(flow_run_id: Optional[UUID] = None) -> UUID:
+    if flow_run_id:
+        return flow_run_id
+
+    context = FlowRunContext.get()
+    if context is None or context.flow_run is None:
+        raise RuntimeError("Must either provide a flow run ID or be within a flow run.")
+
+    return context.flow_run.id
+
+
+@sync_compatible
+@inject_client
+async def create_flow_run_input(
+    key: str,
+    value: Any,
+    flow_run_id: Optional[UUID] = None,
+    client: "PrefectClient" = None,
+):
+    """
+    Create a new flow run input. The given `value` will be serialized to JSON
+    and stored as a flow run input value.
+
+    Args:
+        - key (str): the flow run input key
+        - value (Any): the flow run input value
+        - flow_run_id (UUID): the, optional, flow run ID. If not given will
+          default to pulling the flow run ID from the current context.
+    """
+    flow_run_id = _ensure_flow_run_id(flow_run_id)
+
+    await client.create_flow_run_input(
+        flow_run_id=flow_run_id, key=key, value=orjson.dumps(value).decode()
+    )
+
+
+@sync_compatible
+@inject_client
+async def read_flow_run_input(
+    key: str, flow_run_id: Optional[UUID] = None, client: "PrefectClient" = None
+) -> Any:
+    """Read a flow run input.
+
+    Args:
+        - key (str): the flow run input key
+        - flow_run_id (UUID): the flow run ID
+    """
+    flow_run_id = _ensure_flow_run_id(flow_run_id)
+
+    try:
+        value = await client.read_flow_run_input(flow_run_id=flow_run_id, key=key)
+    except PrefectHTTPStatusError as exc:
+        if exc.response.status_code == 404:
+            return None
+        raise
+    else:
+        return orjson.loads(value)
+
+
+@sync_compatible
+@inject_client
+async def delete_flow_run_input(
+    key: str, flow_run_id: Optional[UUID] = None, client: "PrefectClient" = None
+):
+    """Delete a flow run input.
+
+    Args:
+        - flow_run_id (UUID): the flow run ID
+        - key (str): the flow run input key
+    """
+
+    flow_run_id = _ensure_flow_run_id(flow_run_id)
+
+    await client.delete_flow_run_input(flow_run_id=flow_run_id, key=key)
diff --git a/src/prefect/input/run_input.py b/src/prefect/input/run_input.py
new file mode 100644
index 000000000000..91b552644303
--- /dev/null
+++ b/src/prefect/input/run_input.py
@@ -0,0 +1,106 @@
+from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Type, Union
+from uuid import UUID
+
+import pydantic
+
+from prefect._internal.pydantic import HAS_PYDANTIC_V2
+from prefect.input.actions import create_flow_run_input, read_flow_run_input
+from prefect.utilities.asyncutils import sync_compatible
+
+if TYPE_CHECKING:
+    from prefect.states import State
+
+
+if HAS_PYDANTIC_V2:
+    from prefect._internal.pydantic.v2_schema import create_v2_schema
+
+
+KeysetNames = Union[Literal["response"], Literal["schema"]]
+Keyset = Dict[KeysetNames, str]
+
+
+def keyset_from_paused_state(state: "State") -> Keyset:
+    """
+    Get the keyset for the given Paused state.
+
+    Args:
+        - state (State): the state to get the keyset for
+    """
+
+    if not state.is_paused():
+        raise RuntimeError(f"{state.type.value!r} is unsupported.")
+
+    return keyset_from_base_key(
+        f"{state.name.lower()}-{str(state.state_details.pause_key)}"
+    )
+
+
+def keyset_from_base_key(base_key: str) -> Keyset:
+    """
+    Get the keyset for the given base key.
+
+    Args:
+        - base_key (str): the base key to get the keyset for
+
+    Returns:
+        - Dict[str, str]: the keyset
+    """
+    return {
+        "response": f"{base_key}-response",
+        "schema": f"{base_key}-schema",
+    }
+
+
+class RunInput(pydantic.BaseModel):
+    class Config:
+        extra = "forbid"
+
+    title: str = "Run is asking for input"
+    description: Optional[str] = None
+
+    @classmethod
+    @sync_compatible
+    async def save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None):
+        """
+        Save the run input response to the given key.
+
+        Args:
+            - keyset (Keyset): the keyset to save the input for
+            - flow_run_id (UUID, optional): the flow run ID to save the input for
+        """
+
+        if HAS_PYDANTIC_V2:
+            schema = create_v2_schema(cls.__name__, model_base=cls)
+        else:
+            schema = cls.schema(by_alias=True)
+
+        await create_flow_run_input(
+            key=keyset["schema"], value=schema, flow_run_id=flow_run_id
+        )
+
+    @classmethod
+    @sync_compatible
+    async def load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None):
+        """
+        Load the run input response from the given key.
+
+        Args:
+            - keyset (Keyset): the keyset to load the input for
+            - flow_run_id (UUID, optional): the flow run ID to load the input for
+        """
+        value = await read_flow_run_input(keyset["response"], flow_run_id=flow_run_id)
+        return cls(**value)
+
+    @classmethod
+    def with_initial_data(cls, **kwargs: Any) -> Type["RunInput"]:
+        """
+        Create a new `RunInput` subclass with the given initial data as field
+        defaults.
+
+        Args:
+            - kwargs (Any): the initial data
+        """
+        fields = {}
+        for key, value in kwargs.items():
+            fields[key] = (type(value), value)
+        return pydantic.create_model(cls.__name__, **fields, __base__=cls)
diff --git a/src/prefect/server/schemas/states.py b/src/prefect/server/schemas/states.py
index 1eb2bb1a4525..0b378030b430 100644
--- a/src/prefect/server/schemas/states.py
+++ b/src/prefect/server/schemas/states.py
@@ -4,7 +4,7 @@
 
 import datetime
 import warnings
-from typing import TYPE_CHECKING, Any, Generic, Optional, Type, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Dict, Generic, Optional, Type, TypeVar, Union
 from uuid import UUID
 
 import pendulum
@@ -63,6 +63,7 @@ class StateDetails(PrefectBaseModel):
     pause_timeout: DateTimeTZ = None
     pause_reschedule: bool = False
     pause_key: str = None
+    run_input_keyset: Optional[Dict[str, str]] = None
     refresh_cache: bool = None
 
 
diff --git a/src/prefect/settings.py b/src/prefect/settings.py
index 30601f43f690..866fbdcf2478 100644
--- a/src/prefect/settings.py
+++ b/src/prefect/settings.py
@@ -933,12 +933,12 @@ def default_cloud_ui_url(settings, value):
     default=10.0,
 )
 """
-Threshold time in seconds for logging a warning if task parameter introspection 
+Threshold time in seconds for logging a warning if task parameter introspection
 exceeds this duration. Parameter introspection can be a significant performance hit
 when the parameter is a large collection object, e.g. a large dictionary or DataFrame,
-and each element needs to be inspected. See `prefect.utilities.annotations.quote` 
+and each element needs to be inspected. See `prefect.utilities.annotations.quote`
 for more details.
-Defaults to `10.0`. 
+Defaults to `10.0`.
 Set to `0` to disable logging the warning.
 """
 
@@ -1332,6 +1332,16 @@ def default_cloud_ui_url(settings, value):
 Whether or not to warn when deployment status is used.
 """
 
+PREFECT_EXPERIMENTAL_FLOW_RUN_INPUT = Setting(bool, default=False)
+"""
+Whether or not to enable flow run input.
+"""
+
+PREFECT_EXPERIMENTAL_WARN_FLOW_RUN_INPUT = Setting(bool, default=True)
+"""
+Whether or not to enable flow run input.
+"""
+
 PREFECT_RUNNER_PROCESS_LIMIT = Setting(int, default=5)
 """
 Maximum number of processes a runner will execute in parallel.
diff --git a/tests/input/test_actions.py b/tests/input/test_actions.py
new file mode 100644
index 000000000000..ae42339002fe
--- /dev/null
+++ b/tests/input/test_actions.py
@@ -0,0 +1,85 @@
+import pytest
+
+from prefect._internal.pydantic import HAS_PYDANTIC_V2
+
+if HAS_PYDANTIC_V2:
+    from pydantic.v1 import ValidationError
+else:
+    from pydantic import ValidationError
+
+from prefect.context import FlowRunContext
+from prefect.input import (
+    create_flow_run_input,
+    delete_flow_run_input,
+    read_flow_run_input,
+)
+
+
+@pytest.fixture
+def flow_run_context(flow_run, prefect_client):
+    with FlowRunContext.construct(flow_run=flow_run, client=prefect_client) as context:
+        yield context
+
+
+class TestCreateFlowRunInput:
+    async def test_implicit_flow_run(self, flow_run_context):
+        await create_flow_run_input(key="key", value="value")
+        assert (
+            await read_flow_run_input(
+                key="key", flow_run_id=flow_run_context.flow_run.id
+            )
+            == "value"
+        )
+
+    async def test_explicit_flow_run(self, flow_run):
+        await create_flow_run_input(key="key", value="value", flow_run_id=flow_run.id)
+        assert await read_flow_run_input(key="key", flow_run_id=flow_run.id) == "value"
+
+    async def test_no_flow_run_raises(self):
+        with pytest.raises(
+            RuntimeError, match="provide a flow run ID or be within a flow run"
+        ):
+            await create_flow_run_input(key="key", value="value")
+
+    async def test_validates_key(self, flow_run):
+        with pytest.raises(ValidationError, match="value must only contain"):
+            await create_flow_run_input(
+                key="invalid key *@&*$&", value="value", flow_run_id=flow_run.id
+            )
+
+    def test_can_be_used_sync(self, flow_run_context):
+        create_flow_run_input(key="key", value="value")
+        assert (
+            read_flow_run_input(key="key", flow_run_id=flow_run_context.flow_run.id)
+            == "value"
+        )
+
+
+class TestDeleteFlowRunInput:
+    async def test_implicit_flow_run(self, flow_run_context):
+        await create_flow_run_input(key="key", value="value")
+        await delete_flow_run_input(key="key")
+        assert (
+            await read_flow_run_input(
+                key="key", flow_run_id=flow_run_context.flow_run.id
+            )
+            is None
+        )
+
+    async def test_explicit_flow_run(self, flow_run):
+        await create_flow_run_input(key="key", value="value", flow_run_id=flow_run.id)
+        await delete_flow_run_input(key="key", flow_run_id=flow_run.id)
+        assert await read_flow_run_input(key="key", flow_run_id=flow_run.id) is None
+
+
+class TestReadFlowRunInput:
+    async def test_implicit_flow_run(self, flow_run_context):
+        await create_flow_run_input(key="key", value="value")
+        assert await read_flow_run_input(key="key") == "value"
+
+    async def test_explicit_flow_run(self, flow_run):
+        await create_flow_run_input(key="key", value="value", flow_run_id=flow_run.id)
+        assert await read_flow_run_input(key="key", flow_run_id=flow_run.id) == "value"
+
+    async def test_missing_key_returns_none(self, flow_run):
+        assert await read_flow_run_input(key="missing", flow_run_id=flow_run.id) is None
diff --git a/tests/input/test_run_input.py b/tests/input/test_run_input.py
new file mode 100644
index 000000000000..4ee36918e933
--- /dev/null
+++ b/tests/input/test_run_input.py
@@ -0,0 +1,151 @@
+import pydantic
+import pytest
+
+from prefect.context import FlowRunContext
+from prefect.input import (
+    RunInput,
+    create_flow_run_input,
+    keyset_from_base_key,
+    keyset_from_paused_state,
+    read_flow_run_input,
+)
+from prefect.states import Paused, Running, Suspended
+
+
+@pytest.fixture
+def flow_run_context(flow_run, prefect_client):
+    with FlowRunContext.construct(flow_run=flow_run, client=prefect_client) as context:
+        yield context
+
+
+class Person(RunInput):
+    name: str
+    email: str
+    human: bool
+
+
+def test_keyset_from_base_key():
+    keyset = keyset_from_base_key("person")
+    assert keyset["response"] == "person-response"
+    assert keyset["schema"] == "person-schema"
+
+
+@pytest.mark.parametrize(
+    "state,expected",
+    [
+        (Paused(pause_key="1"), keyset_from_base_key("paused-1")),
+        (Suspended(pause_key="1"), keyset_from_base_key("suspended-1")),
+    ],
+)
+def test_keyset_from_paused_state(state, expected):
+    assert keyset_from_paused_state(state) == expected
+
+
+def test_keyset_from_paused_state_non_paused_state_raises_exception():
+    with pytest.raises(RuntimeError, match="unsupported"):
+        keyset_from_paused_state(Running())
+
+
+async def test_save_schema(flow_run_context):
+    keyset = keyset_from_base_key("person")
+    await Person.save(keyset)
+    schema = await read_flow_run_input(key=keyset["schema"])
+    assert set(schema["properties"].keys()) == {
+        "title",
+        "description",
+        "name",
+        "email",
+        "human",
+    }
+
+
+def test_save_works_sync(flow_run_context):
+    keyset = keyset_from_base_key("person")
+    Person.save(keyset)
+    schema = read_flow_run_input(key=keyset["schema"])
+    assert set(schema["properties"].keys()) == {
+        "title",
+        "description",
+        "name",
+        "email",
+        "human",
+    }
+
+
+async def test_save_explicit_flow_run(flow_run):
+    keyset = keyset_from_base_key("person")
+    await Person.save(keyset, flow_run_id=flow_run.id)
+    schema = await read_flow_run_input(key=keyset["schema"], flow_run_id=flow_run.id)
+    assert schema is not None
+
+
+async def test_load(flow_run_context):
+    keyset = keyset_from_base_key("person")
+    await create_flow_run_input(
+        keyset["response"],
+        value={"name": "Bob", "email": "bob@bob.bob", "human": True},
+    )
+
+    person = await Person.load(keyset)
+    assert isinstance(person, Person)
+    assert person.name == "Bob"
+    assert person.email == "bob@bob.bob"
+    assert person.human is True
+
+
+def test_load_works_sync(flow_run_context):
+    keyset = keyset_from_base_key("person")
+    create_flow_run_input(
+        keyset["response"],
+        value={"name": "Bob", "email": "bob@bob.bob", "human": True},
+    )
+
+    person = Person.load(keyset)
+    assert isinstance(person, Person)
+    assert person.name == "Bob"
+    assert person.email == "bob@bob.bob"
+    assert person.human is True
+
+
+async def test_load_explicit_flow_run(flow_run):
+    keyset = keyset_from_base_key("person")
+    await create_flow_run_input(
+        keyset["response"],
+        value={"name": "Bob", "email": "bob@bob.bob", "human": True},
+        flow_run_id=flow_run.id,
+    )
+
+    person = await Person.load(keyset, flow_run_id=flow_run.id)
+    assert isinstance(person, Person)
+
+
+async def test_load_fails_validation_raises_exception(flow_run_context):
+    keyset = keyset_from_base_key("person")
+    await create_flow_run_input(
+        keyset["response"],
+        value={
+            "name": "Bob",
+            "email": "bob@bob.bob",
+            "human": "123",
+        },  # Human should be a boolean value.
+    )
+
+    with pytest.raises(pydantic.ValidationError, match="boolean"):
+        person = await Person.load(keyset)
+        assert isinstance(person, Person)
+
+
+async def test_with_initial_data(flow_run_context):
+    keyset = keyset_from_base_key("bob")
+
+    name = "Bob"
+    new_cls = Person.with_initial_data(
+        title=f"Fill in the missing data for {name}", name=name
+    )
+
+    await new_cls.save(keyset)
+    schema = await read_flow_run_input(key=keyset["schema"])
+    assert (
+        schema["properties"]["title"]["default"] == "Fill in the missing data for Bob"
+    )
+    assert schema["properties"]["name"]["default"] == "Bob"
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 7a9941860e69..fd292ad01967 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -22,6 +22,7 @@
 
 import prefect.flows
 from prefect import engine, flow, task
+from prefect._internal.compatibility.experimental import ExperimentalFeature
 from prefect.client.orchestration import get_client
 from prefect.client.schemas import OrchestrationResult
 from prefect.context import FlowRunContext, get_run_context
@@ -52,6 +53,7 @@
     SignatureMismatchError,
 )
 from prefect.futures import PrefectFuture
+from prefect.input import RunInput, create_flow_run_input, read_flow_run_input
 from prefect.results import ResultFactory
 from prefect.server.schemas.core import FlowRun
 from prefect.server.schemas.filters import FlowRunFilter
@@ -149,6 +151,12 @@ async def _get_flow_run_context():
 
 
 class TestBlockingPause:
+    @pytest.fixture(autouse=True)
+    def ignore_experimental_warnings(self):
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", category=ExperimentalFeature)
+            yield
+
     async def test_tasks_cannot_be_paused(self):
         @task
         async def the_little_task_that_pauses():
@@ -307,6 +315,56 @@ async def flow_resumer():
         )
         assert len(task_runs) == 5, "all tasks should finish running"
 
+    async def test_paused_flows_can_receive_input(self, prefect_client):
+        flow_run_id = None
+
+        class FlowInput(RunInput):
+            x: int
+
+        @flow(task_runner=SequentialTaskRunner())
+        async def pausing_flow():
+            nonlocal flow_run_id
+            context = FlowRunContext.get()
+            flow_run_id = context.flow_run.id
+
+            flow_input = await pause_flow_run(
+                timeout=10, poll_interval=2, wait_for_input=FlowInput
+            )
+            return flow_input
+
+        async def flow_resumer():
+            # Wait on flow run to start
+            while not flow_run_id:
+                await anyio.sleep(0.1)
+
+            # Wait on flow run to pause
+            flow_run = await prefect_client.read_flow_run(flow_run_id)
+            while not flow_run.state.is_paused():
+                await asyncio.sleep(0.1)
+                flow_run = await prefect_client.read_flow_run(flow_run_id)
+
+            keyset = flow_run.state.state_details.run_input_keyset
+            assert keyset
+
+            await create_flow_run_input(
+                key=keyset["response"], value={"x": 42}, flow_run_id=flow_run_id
+            )
+            await resume_flow_run(flow_run_id)
+
+        flow_run_state, the_answer = await asyncio.gather(
+            pausing_flow(return_state=True),
+            flow_resumer(),
+        )
+        flow_input = await flow_run_state.result()
+        assert isinstance(flow_input, FlowInput)
+        assert flow_input.x == 42
+
+        # Ensure that the flow run did create the corresponding schema input
+        schema = await read_flow_run_input(
+            key="paused-1-schema", flow_run_id=flow_run_id
+        )
+        assert schema is not None
+
 
 class TestNonblockingPause:
     @pytest.fixture(autouse=True)
@@ -481,6 +539,12 @@ def ignore_deprecation_warnings(self):
             warnings.filterwarnings("ignore", category=DeprecationWarning)
             yield
 
+    @pytest.fixture(autouse=True)
+    def ignore_experimental_warnings(self):
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", category=ExperimentalFeature)
+            yield
+
     async def test_flows_can_be_paused_out_of_process(
         self, prefect_client, deployment, session
     ):
@@ -580,6 +644,15 @@ async def pausing_flow():
         assert state.state_details.pause_reschedule is False
         assert state.state_details.pause_key == "foo"
 
+    async def test_out_of_process_pause_cannot_wait_for_input(self):
+        class FlowInput(RunInput):
+            x: int
+
+        with pytest.raises(
+            RuntimeError, match="Cannot wait for input when pausing out of process."
+        ):
+            await pause_flow_run(flow_run_id=uuid4(), wait_for_input=FlowInput)
+
 
 class TestSuspendFlowRun:
     async def test_suspended_flow_runs_do_not_block_execution(
@@ -718,6 +791,66 @@ async def suspending_flow():
         # of the tasks.
         assert task_completions > 0 and task_completions < 20
 
+    async def test_suspend_can_receive_input(self, deployment, session, prefect_client):
+        flow_run_id = None
+
+        class FlowInput(RunInput):
+            x: int
+
+        @flow()
+        async def suspending_flow():
+            nonlocal flow_run_id
+            context = get_run_context()
+            assert context.flow_run
+
+            if not context.flow_run.deployment_id:
+                # Ensure that the flow run has a deployment id so it's
+                # suspendable.
+                from prefect.server.models.flow_runs import update_flow_run
+
+                await update_flow_run(
+                    session,
+                    context.flow_run.id,
+                    FlowRun.construct(deployment_id=deployment.id),
+                )
+                await session.commit()
+
+            flow_run_id = context.flow_run.id
+
+            flow_input = await suspend_flow_run(wait_for_input=FlowInput)
+
+            return flow_input
+
+        with pytest.raises(Pause):
+            await suspending_flow()
+
+        assert flow_run_id
+
+        flow_run = await prefect_client.read_flow_run(flow_run_id)
+        keyset = flow_run.state.state_details.run_input_keyset
+
+        schema = await read_flow_run_input(
+            key=keyset["schema"], flow_run_id=flow_run_id
+        )
+        assert schema is not None
+
+        await create_flow_run_input(
+            key=keyset["response"], value={"x": 42}, flow_run_id=flow_run_id
+        )
+
+        await resume_flow_run(flow_run_id)
+
+        state = await begin_flow_run(
+            flow=suspending_flow,
+            flow_run=flow_run,
+            parameters={},
+            client=prefect_client,
+            user_thread=threading.current_thread(),
+        )
+
+        flow_input = await state.result()
+        assert flow_input.x == 42
+
 
 class TestOrchestrateTaskRun:
     async def test_propose_state_does_not_recurse(


Code review comments:
diff_hunk: @@ -1026,12 +1045,19 @@ async def _in_process_pause(
 
     logger.info("Pausing flow, execution will continue when this flow run is resumed.")
 
+    proposed_state = Paused(
+        timeout_seconds=timeout, reschedule=reschedule, pause_key=pause_key
+    )
+
+    run_input_keyset = keyset_from_paused_state(proposed_state)
path: src/prefect/engine.py
text: @user1:
Do we need this if we aren't waiting for input?

@author:
Nope.