diff --git a/esphome/components/fan/__init__.py b/esphome/components/fan/__init__.py
index 4e0e52cd65..9c9cb6327b 100644
--- a/esphome/components/fan/__init__.py
+++ b/esphome/components/fan/__init__.py
@@ -5,6 +5,8 @@ from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DIRECTION,
+    CONF_DIRECTION_COMMAND_TOPIC,
+    CONF_DIRECTION_STATE_TOPIC,
     CONF_ID,
     CONF_MQTT_ID,
     CONF_OFF_SPEED_CYCLE,
@@ -90,6 +92,12 @@ FAN_SCHEMA = (
                 RESTORE_MODES, upper=True, space="_"
             ),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTFanComponent),
+            cv.Optional(CONF_DIRECTION_STATE_TOPIC): cv.All(
+                cv.requires_component("mqtt"), cv.publish_topic
+            ),
+            cv.Optional(CONF_DIRECTION_COMMAND_TOPIC): cv.All(
+                cv.requires_component("mqtt"), cv.subscribe_topic
+            ),
             cv.Optional(CONF_OSCILLATION_STATE_TOPIC): cv.All(
                 cv.requires_component("mqtt"), cv.publish_topic
             ),
@@ -193,6 +201,14 @@ async def setup_fan_core_(var, config):
         mqtt_ = cg.new_Pvariable(mqtt_id, var)
         await mqtt.register_mqtt_component(mqtt_, config)
 
+        if (
+            direction_state_topic := config.get(CONF_DIRECTION_STATE_TOPIC)
+        ) is not None:
+            cg.add(mqtt_.set_custom_direction_state_topic(direction_state_topic))
+        if (
+            direction_command_topic := config.get(CONF_DIRECTION_COMMAND_TOPIC)
+        ) is not None:
+            cg.add(mqtt_.set_custom_direction_command_topic(direction_command_topic))
         if (
             oscillation_state_topic := config.get(CONF_OSCILLATION_STATE_TOPIC)
         ) is not None:
diff --git a/esphome/components/mqtt/mqtt_const.h b/esphome/components/mqtt/mqtt_const.h
index 445457a27f..3ddd8fc5cc 100644
--- a/esphome/components/mqtt/mqtt_const.h
+++ b/esphome/components/mqtt/mqtt_const.h
@@ -64,6 +64,8 @@ constexpr const char *const MQTT_DEVICE_NAME = "name";
 constexpr const char *const MQTT_DEVICE_SUGGESTED_AREA = "sa";
 constexpr const char *const MQTT_DEVICE_SW_VERSION = "sw";
 constexpr const char *const MQTT_DEVICE_HW_VERSION = "hw";
+constexpr const char *const MQTT_DIRECTION_COMMAND_TOPIC = "dir_cmd_t";
+constexpr const char *const MQTT_DIRECTION_STATE_TOPIC = "dir_stat_t";
 constexpr const char *const MQTT_DOCKED_TEMPLATE = "dock_tpl";
 constexpr const char *const MQTT_DOCKED_TOPIC = "dock_t";
 constexpr const char *const MQTT_EFFECT_COMMAND_TOPIC = "fx_cmd_t";
@@ -328,6 +330,8 @@ constexpr const char *const MQTT_DEVICE_NAME = "name";
 constexpr const char *const MQTT_DEVICE_SUGGESTED_AREA = "suggested_area";
 constexpr const char *const MQTT_DEVICE_SW_VERSION = "sw_version";
 constexpr const char *const MQTT_DEVICE_HW_VERSION = "hw_version";
+constexpr const char *const MQTT_DIRECTION_COMMAND_TOPIC = "direction_command_topic";
+constexpr const char *const MQTT_DIRECTION_STATE_TOPIC = "direction_state_topic";
 constexpr const char *const MQTT_DOCKED_TEMPLATE = "docked_template";
 constexpr const char *const MQTT_DOCKED_TOPIC = "docked_topic";
 constexpr const char *const MQTT_EFFECT_COMMAND_TOPIC = "effect_command_topic";
diff --git a/esphome/components/mqtt/mqtt_fan.cpp b/esphome/components/mqtt/mqtt_fan.cpp
index 32892199fe..9e5ea54bee 100644
--- a/esphome/components/mqtt/mqtt_fan.cpp
+++ b/esphome/components/mqtt/mqtt_fan.cpp
@@ -43,6 +43,32 @@ void MQTTFanComponent::setup() {
     }
   });
 
+  if (this->state_->get_traits().supports_direction()) {
+    this->subscribe(this->get_direction_command_topic(), [this](const std::string &topic, const std::string &payload) {
+      auto val = parse_on_off(payload.c_str(), "forward", "reverse");
+      switch (val) {
+        case PARSE_ON:
+          ESP_LOGD(TAG, "'%s': Setting direction FORWARD", this->friendly_name().c_str());
+          this->state_->make_call().set_direction(fan::FanDirection::FORWARD).perform();
+          break;
+        case PARSE_OFF:
+          ESP_LOGD(TAG, "'%s': Setting direction REVERSE", this->friendly_name().c_str());
+          this->state_->make_call().set_direction(fan::FanDirection::REVERSE).perform();
+          break;
+        case PARSE_TOGGLE:
+          this->state_->make_call()
+              .set_direction(this->state_->direction == fan::FanDirection::FORWARD ? fan::FanDirection::REVERSE
+                                                                                   : fan::FanDirection::FORWARD)
+              .perform();
+          break;
+        case PARSE_NONE:
+          ESP_LOGW(TAG, "Unknown direction Payload %s", payload.c_str());
+          this->status_momentary_warning("direction", 5000);
+          break;
+      }
+    });
+  }
+
   if (this->state_->get_traits().supports_oscillation()) {
     this->subscribe(this->get_oscillation_command_topic(),
                     [this](const std::string &topic, const std::string &payload) {
@@ -94,6 +120,10 @@ void MQTTFanComponent::setup() {
 void MQTTFanComponent::dump_config() {
   ESP_LOGCONFIG(TAG, "MQTT Fan '%s': ", this->state_->get_name().c_str());
   LOG_MQTT_COMPONENT(true, true);
+  if (this->state_->get_traits().supports_direction()) {
+    ESP_LOGCONFIG(TAG, "  Direction State Topic: '%s'", this->get_direction_state_topic().c_str());
+    ESP_LOGCONFIG(TAG, "  Direction Command Topic: '%s'", this->get_direction_command_topic().c_str());
+  }
   if (this->state_->get_traits().supports_oscillation()) {
     ESP_LOGCONFIG(TAG, "  Oscillation State Topic: '%s'", this->get_oscillation_state_topic().c_str());
     ESP_LOGCONFIG(TAG, "  Oscillation Command Topic: '%s'", this->get_oscillation_command_topic().c_str());
@@ -107,6 +137,10 @@ void MQTTFanComponent::dump_config() {
 bool MQTTFanComponent::send_initial_state() { return this->publish_state(); }
 
 void MQTTFanComponent::send_discovery(JsonObject root, mqtt::SendDiscoveryConfig &config) {
+  if (this->state_->get_traits().supports_direction()) {
+    root[MQTT_DIRECTION_COMMAND_TOPIC] = this->get_direction_command_topic();
+    root[MQTT_DIRECTION_STATE_TOPIC] = this->get_direction_state_topic();
+  }
   if (this->state_->get_traits().supports_oscillation()) {
     root[MQTT_OSCILLATION_COMMAND_TOPIC] = this->get_oscillation_command_topic();
     root[MQTT_OSCILLATION_STATE_TOPIC] = this->get_oscillation_state_topic();
@@ -122,6 +156,11 @@ bool MQTTFanComponent::publish_state() {
   ESP_LOGD(TAG, "'%s' Sending state %s.", this->state_->get_name().c_str(), state_s);
   this->publish(this->get_state_topic_(), state_s);
   bool failed = false;
+  if (this->state_->get_traits().supports_direction()) {
+    bool success = this->publish(this->get_direction_state_topic(),
+                                 this->state_->direction == fan::FanDirection::FORWARD ? "forward" : "reverse");
+    failed = failed || !success;
+  }
   if (this->state_->get_traits().supports_oscillation()) {
     bool success = this->publish(this->get_oscillation_state_topic(),
                                  this->state_->oscillating ? "oscillate_on" : "oscillate_off");
diff --git a/esphome/components/mqtt/mqtt_fan.h b/esphome/components/mqtt/mqtt_fan.h
index 12286b9f01..fdcec0782d 100644
--- a/esphome/components/mqtt/mqtt_fan.h
+++ b/esphome/components/mqtt/mqtt_fan.h
@@ -15,6 +15,8 @@ class MQTTFanComponent : public mqtt::MQTTComponent {
  public:
   explicit MQTTFanComponent(fan::Fan *state);
 
+  MQTT_COMPONENT_CUSTOM_TOPIC(direction, command)
+  MQTT_COMPONENT_CUSTOM_TOPIC(direction, state)
   MQTT_COMPONENT_CUSTOM_TOPIC(oscillation, command)
   MQTT_COMPONENT_CUSTOM_TOPIC(oscillation, state)
   MQTT_COMPONENT_CUSTOM_TOPIC(speed_level, command)
diff --git a/esphome/const.py b/esphome/const.py
index 3b84055789..d656b15519 100644
--- a/esphome/const.py
+++ b/esphome/const.py
@@ -221,7 +221,9 @@ CONF_DIMENSIONS = "dimensions"
 CONF_DIO_PIN = "dio_pin"
 CONF_DIR_PIN = "dir_pin"
 CONF_DIRECTION = "direction"
+CONF_DIRECTION_COMMAND_TOPIC = "direction_command_topic"
 CONF_DIRECTION_OUTPUT = "direction_output"
+CONF_DIRECTION_STATE_TOPIC = "direction_state_topic"
 CONF_DISABLE_CRC = "disable_crc"
 CONF_DISABLED = "disabled"
 CONF_DISABLED_BY_DEFAULT = "disabled_by_default"
diff --git a/tests/components/mqtt/common.yaml b/tests/components/mqtt/common.yaml
index a4bdf58809..1ab8872fdb 100644
--- a/tests/components/mqtt/common.yaml
+++ b/tests/components/mqtt/common.yaml
@@ -293,6 +293,8 @@ fan:
   - platform: template
     name: Template Fan
     state_topic: some/topic/fan
+    direction_state_topic: some/topic/direction/state
+    direction_command_topic: some/topic/direction/command
     qos: 2
     on_state:
       - logger.log: on_state