diff --git a/homeassistant/components/input_text/reproduce_state.py b/homeassistant/components/input_text/reproduce_state.py
new file mode 100644
index 00000000000..f64c5c019f6
--- /dev/null
+++ b/homeassistant/components/input_text/reproduce_state.py
@@ -0,0 +1,46 @@
+"""Reproduce an Input text state."""
+import asyncio
+import logging
+from typing import Iterable, Optional
+
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import Context, State
+from homeassistant.helpers.typing import HomeAssistantType
+
+from . import DOMAIN, SERVICE_SET_VALUE, ATTR_VALUE
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def _async_reproduce_state(
+    hass: HomeAssistantType, state: State, context: Optional[Context] = None
+) -> None:
+    """Reproduce a single state."""
+    cur_state = hass.states.get(state.entity_id)
+
+    # Return if we can't find the entity
+    if cur_state is None:
+        _LOGGER.warning("Unable to find entity %s", state.entity_id)
+        return
+
+    # Return if we are already at the right state.
+    if cur_state.state == state.state:
+        return
+
+    # Call service
+    service = SERVICE_SET_VALUE
+    service_data = {ATTR_ENTITY_ID: state.entity_id, ATTR_VALUE: state.state}
+
+    await hass.services.async_call(
+        DOMAIN, service, service_data, context=context, blocking=True
+    )
+
+
+async def async_reproduce_states(
+    hass: HomeAssistantType, states: Iterable[State], context: Optional[Context] = None
+) -> None:
+    """Reproduce Input text states."""
+    # Reproduce states in parallel.
+    await asyncio.gather(
+        *(_async_reproduce_state(hass, state, context) for state in states)
+    )
diff --git a/tests/components/input_text/test_reproduce_state.py b/tests/components/input_text/test_reproduce_state.py
new file mode 100644
index 00000000000..fd75948d461
--- /dev/null
+++ b/tests/components/input_text/test_reproduce_state.py
@@ -0,0 +1,65 @@
+"""Test reproduce state for Input text."""
+from homeassistant.core import State
+from homeassistant.setup import async_setup_component
+
+VALID_TEXT1 = "Test text"
+VALID_TEXT2 = "LoremIpsum"
+INVALID_TEXT1 = "This text is too long!"
+INVALID_TEXT2 = "Short"
+
+
+async def test_reproducing_states(hass, caplog):
+    """Test reproducing Input text states."""
+
+    # Setup entity for testing
+    assert await async_setup_component(
+        hass,
+        "input_text",
+        {
+            "input_text": {
+                "test_text": {"min": "6", "max": "10", "initial": VALID_TEXT1}
+            }
+        },
+    )
+
+    # These calls should do nothing as entities already in desired state
+    await hass.helpers.state.async_reproduce_state(
+        [
+            State("input_text.test_text", VALID_TEXT1),
+            # Should not raise
+            State("input_text.non_existing", VALID_TEXT1),
+        ],
+        blocking=True,
+    )
+
+    # Test that entity is in desired state
+    assert hass.states.get("input_text.test_text").state == VALID_TEXT1
+
+    # Try reproducing with different state
+    await hass.helpers.state.async_reproduce_state(
+        [
+            State("input_text.test_text", VALID_TEXT2),
+            # Should not raise
+            State("input_text.non_existing", VALID_TEXT2),
+        ],
+        blocking=True,
+    )
+
+    # Test that the state was changed
+    assert hass.states.get("input_text.test_text").state == VALID_TEXT2
+
+    # Test setting state to invalid state (length too long)
+    await hass.helpers.state.async_reproduce_state(
+        [State("input_text.test_text", INVALID_TEXT1)], blocking=True
+    )
+
+    # The entity state should be unchanged
+    assert hass.states.get("input_text.test_text").state == VALID_TEXT2
+
+    # Test setting state to invalid state (length too short)
+    await hass.helpers.state.async_reproduce_state(
+        [State("input_text.test_text", INVALID_TEXT2)], blocking=True
+    )
+
+    # The entity state should be unchanged
+    assert hass.states.get("input_text.test_text").state == VALID_TEXT2