---
title: run_input
sidebarTitle: run_input
---

# `prefect.input.run_input`



This module contains functions that allow sending type-checked `RunInput` data
to flows at runtime. Flows can send back responses, establishing two-way
channels with senders. These functions are particularly useful for systems that
require ongoing data transfer or need to react to input quickly.
real-time interaction and efficient data handling. It's designed to facilitate
dynamic communication within distributed or microservices-oriented systems,
making it ideal for scenarios requiring continuous data synchronization and
processing. It's particularly useful for systems that require ongoing data
input and output.

The following is an example of two flows. One sends a random number to the
other and waits for a response. The other receives the number, squares it, and
sends the result back. The sender flow then prints the result.

Sender flow:

```python
import random
from uuid import UUID
from prefect import flow
from prefect.logging import get_run_logger
from prefect.input import RunInput

class NumberData(RunInput):
    number: int


@flow
async def sender_flow(receiver_flow_run_id: UUID):
    logger = get_run_logger()

    the_number = random.randint(1, 100)

    await NumberData(number=the_number).send_to(receiver_flow_run_id)

    receiver = NumberData.receive(flow_run_id=receiver_flow_run_id)
    squared = await receiver.next()

    logger.info(f"{the_number} squared is {squared.number}")
```

Receiver flow:
```python
import random
from uuid import UUID
from prefect import flow
from prefect.logging import get_run_logger
from prefect.input import RunInput

class NumberData(RunInput):
    number: int


@flow
async def receiver_flow():
    async for data in NumberData.receive():
        squared = data.number ** 2
        data.respond(NumberData(number=squared))
```


## Functions

### `keyset_from_paused_state` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L113" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
keyset_from_paused_state(state: 'State') -> Keyset
```


Get the keyset for the given Paused state.

**Args:**
- `- state`: the state to get the keyset for


### `keyset_from_base_key` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L129" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
keyset_from_base_key(base_key: str) -> Keyset
```


Get the keyset for the given base key.

**Args:**
- `- base_key`: the base key to get the keyset for

**Returns:**
- - Dict[str, str]: the keyset


### `run_input_subclass_from_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L416" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_input_subclass_from_type(_type: Union[Type[R], Type[T], pydantic.BaseModel]) -> Union[Type[AutomaticRunInput[T]], Type[R]]
```


Create a new `RunInput` subclass from the given type.


### `send_input` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L628" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
send_input(run_input: Any, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

### `receive_input` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L668" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
receive_input(input_type: Union[Type[R], Type[T], pydantic.BaseModel], timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None, with_metadata: bool = False) -> Union[GetAutomaticInputHandler[T], GetInputHandler[R]]
```

## Classes

### `RunInputMetadata` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L146" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

### `BaseRunInput` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L152" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `keyset_from_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L163" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
keyset_from_type(cls) -> Keyset
```

#### `load` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L196" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self
```

Load the run input response from the given key.

**Args:**
- `- keyset`: the keyset to load the input for
- `- flow_run_id`: the flow run ID to load the input for


#### `load_from_flow_run_input` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L216" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self
```

Load the run input from a FlowRunInput object.

**Args:**
- `- flow_run_input`: the flow run input to load the input for


#### `metadata` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L159" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
metadata(self) -> RunInputMetadata
```

#### `respond` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L258" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `save` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L168" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None)
```

Save the run input response to the given key.

**Args:**
- `- keyset`: the keyset to save the input for
- `- flow_run_id`: the flow run ID to save the input for


#### `send_to` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L282" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `with_initial_data` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L232" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]
```

Create a new `RunInput` subclass with the given initial data as field
defaults.

**Args:**
- `- description`: a description to show when resuming
a flow run that requires input
- `- kwargs`: the initial data to populate the subclass


### `RunInput` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L296" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `keyset_from_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L163" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
keyset_from_type(cls) -> Keyset
```

#### `load` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L196" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self
```

Load the run input response from the given key.

**Args:**
- `- keyset`: the keyset to load the input for
- `- flow_run_id`: the flow run ID to load the input for


#### `load_from_flow_run_input` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L216" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self
```

Load the run input from a FlowRunInput object.

**Args:**
- `- flow_run_input`: the flow run input to load the input for


#### `metadata` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L159" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
metadata(self) -> RunInputMetadata
```

#### `receive` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L298" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
receive(cls, timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None) -> GetInputHandler[Self]
```

#### `respond` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L258" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `save` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L168" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None)
```

Save the run input response to the given key.

**Args:**
- `- keyset`: the keyset to save the input for
- `- flow_run_id`: the flow run ID to save the input for


#### `send_to` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L282" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `subclass_from_base_model_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L321" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
subclass_from_base_model_type(cls, model_cls: Type[pydantic.BaseModel]) -> Type['RunInput']
```

Create a new `RunInput` subclass from the given `pydantic.BaseModel`
subclass.

**Args:**
- `- model_cls`: the class from which
to create the new `RunInput` subclass


#### `with_initial_data` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L232" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]
```

Create a new `RunInput` subclass with the given initial data as field
defaults.

**Args:**
- `- description`: a description to show when resuming
a flow run that requires input
- `- kwargs`: the initial data to populate the subclass


### `AutomaticRunInput` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L335" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `keyset_from_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L163" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
keyset_from_type(cls) -> Keyset
```

#### `load` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L340" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self
```

Load the run input response from the given key.

**Args:**
- `- keyset`: the keyset to load the input for
- `- flow_run_id`: the flow run ID to load the input for


#### `load` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L196" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self
```

Load the run input response from the given key.

**Args:**
- `- keyset`: the keyset to load the input for
- `- flow_run_id`: the flow run ID to load the input for


#### `load_from_flow_run_input` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L216" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self
```

Load the run input from a FlowRunInput object.

**Args:**
- `- flow_run_input`: the flow run input to load the input for


#### `metadata` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L159" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
metadata(self) -> RunInputMetadata
```

#### `receive` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L392" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
receive(cls, timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None, with_metadata: bool = False) -> GetAutomaticInputHandler[T]
```

#### `respond` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L258" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `save` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L168" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None)
```

Save the run input response to the given key.

**Args:**
- `- keyset`: the keyset to save the input for
- `- flow_run_id`: the flow run ID to save the input for


#### `send_to` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L282" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None)
```

#### `subclass_from_type` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L355" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
subclass_from_type(cls, _type: Type[T]) -> Type['AutomaticRunInput[T]']
```

Create a new `AutomaticRunInput` subclass from the given type.

This method uses the type's name as a key prefix to identify related
flow run inputs. This helps in ensuring that values saved under a type
(like List[int]) are retrievable under the generic type name (like "list").


#### `with_initial_data` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L232" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]
```

Create a new `RunInput` subclass with the given initial data as field
defaults.

**Args:**
- `- description`: a description to show when resuming
a flow run that requires input
- `- kwargs`: the initial data to populate the subclass


### `GetInputHandler` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L437" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `filter_for_inputs` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L484" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
filter_for_inputs(self) -> list['FlowRunInput']
```

#### `next` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L501" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
next(self) -> R
```

#### `to_instance` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L497" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
to_instance(self, flow_run_input: 'FlowRunInput') -> R
```

### `GetAutomaticInputHandler` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L514" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `filter_for_inputs` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L566" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
filter_for_inputs(self) -> list['FlowRunInput']
```

#### `next` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L580" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
next(self) -> Union[T, AutomaticRunInput[T]]
```

#### `to_instance` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/input/run_input.py#L592" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
to_instance(self, flow_run_input: 'FlowRunInput') -> Union[T, AutomaticRunInput[T]]
```
