diff --git a/homeassistant/components/mqtt/button.py b/homeassistant/components/mqtt/button.py
index 982e34d370d..7143b65ed9e 100644
--- a/homeassistant/components/mqtt/button.py
+++ b/homeassistant/components/mqtt/button.py
@@ -15,9 +15,16 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.reload import async_setup_reload_service
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
-from . import PLATFORMS
+from . import PLATFORMS, MqttCommandTemplate
 from .. import mqtt
-from .const import CONF_COMMAND_TOPIC, CONF_ENCODING, CONF_QOS, CONF_RETAIN, DOMAIN
+from .const import (
+    CONF_COMMAND_TEMPLATE,
+    CONF_COMMAND_TOPIC,
+    CONF_ENCODING,
+    CONF_QOS,
+    CONF_RETAIN,
+    DOMAIN,
+)
 from .mixins import MQTT_ENTITY_COMMON_SCHEMA, MqttEntity, async_setup_entry_helper
 
 CONF_PAYLOAD_PRESS = "payload_press"
@@ -26,6 +33,7 @@ DEFAULT_PAYLOAD_PRESS = "PRESS"
 
 PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend(
     {
+        vol.Optional(CONF_COMMAND_TEMPLATE): cv.template,
         vol.Required(CONF_COMMAND_TOPIC): mqtt.valid_publish_topic,
         vol.Optional(CONF_DEVICE_CLASS): button.DEVICE_CLASSES_SCHEMA,
         vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
@@ -81,6 +89,12 @@ class MqttButton(MqttEntity, ButtonEntity):
         """Return the config schema."""
         return DISCOVERY_SCHEMA
 
+    def _setup_from_config(self, config):
+        """(Re)Setup the entity."""
+        self._command_template = MqttCommandTemplate(
+            config.get(CONF_COMMAND_TEMPLATE), entity=self
+        ).async_render
+
     async def _subscribe_topics(self):
         """(Re)Subscribe to topics."""
 
@@ -94,10 +108,11 @@ class MqttButton(MqttEntity, ButtonEntity):
 
         This method is a coroutine.
         """
+        payload = self._command_template(self._config[CONF_PAYLOAD_PRESS])
         await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
-            self._config[CONF_PAYLOAD_PRESS],
+            payload,
             self._config[CONF_QOS],
             self._config[CONF_RETAIN],
             self._config[CONF_ENCODING],
diff --git a/tests/components/mqtt/test_button.py b/tests/components/mqtt/test_button.py
index 7d4194ee5e6..a533e0f0ec7 100644
--- a/tests/components/mqtt/test_button.py
+++ b/tests/components/mqtt/test_button.py
@@ -76,6 +76,40 @@ async def test_sending_mqtt_commands(hass, mqtt_mock):
     assert state.state == "2021-11-08T13:31:44+00:00"
 
 
+async def test_command_template(hass, mqtt_mock):
+    """Test the sending of MQTT commands through a command template."""
+    assert await async_setup_component(
+        hass,
+        button.DOMAIN,
+        {
+            button.DOMAIN: {
+                "command_topic": "command-topic",
+                "command_template": '{ "{{ value }}": "{{ entity_id }}" }',
+                "name": "test",
+                "payload_press": "milky_way_press",
+                "platform": "mqtt",
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get("button.test")
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_FRIENDLY_NAME) == "test"
+
+    await hass.services.async_call(
+        button.DOMAIN,
+        button.SERVICE_PRESS,
+        {ATTR_ENTITY_ID: "button.test"},
+        blocking=True,
+    )
+
+    mqtt_mock.async_publish.assert_called_once_with(
+        "command-topic", '{ "milky_way_press": "button.test" }', 0, False
+    )
+    mqtt_mock.async_publish.reset_mock()
+
+
 async def test_availability_when_connection_lost(hass, mqtt_mock):
     """Test availability after MQTT disconnection."""
     await help_test_availability_when_connection_lost(
@@ -328,7 +362,7 @@ async def test_valid_device_class(hass, mqtt_mock):
 @pytest.mark.parametrize(
     "service,topic,parameters,payload,template",
     [
-        (button.SERVICE_PRESS, "command_topic", None, "PRESS", None),
+        (button.SERVICE_PRESS, "command_topic", None, "PRESS", "command_template"),
     ],
 )
 async def test_publishing_with_custom_encoding(