diff --git a/esphome/components/mqtt/__init__.py b/esphome/components/mqtt/__init__.py
index 8851581ea0..86d163e61d 100644
--- a/esphome/components/mqtt/__init__.py
+++ b/esphome/components/mqtt/__init__.py
@@ -22,6 +22,7 @@ from esphome.const import (
     CONF_DISCOVERY_PREFIX,
     CONF_DISCOVERY_RETAIN,
     CONF_DISCOVERY_UNIQUE_ID_GENERATOR,
+    CONF_ENABLE_ON_BOOT,
     CONF_ID,
     CONF_KEEPALIVE,
     CONF_LEVEL,
@@ -99,6 +100,8 @@ MQTTMessage = mqtt_ns.struct("MQTTMessage")
 MQTTClientComponent = mqtt_ns.class_("MQTTClientComponent", cg.Component)
 MQTTPublishAction = mqtt_ns.class_("MQTTPublishAction", automation.Action)
 MQTTPublishJsonAction = mqtt_ns.class_("MQTTPublishJsonAction", automation.Action)
+MQTTEnableAction = mqtt_ns.class_("MQTTEnableAction", automation.Action)
+MQTTDisableAction = mqtt_ns.class_("MQTTDisableAction", automation.Action)
 MQTTMessageTrigger = mqtt_ns.class_(
     "MQTTMessageTrigger", automation.Trigger.template(cg.std_string), cg.Component
 )
@@ -208,6 +211,7 @@ CONFIG_SCHEMA = cv.All(
         {
             cv.GenerateID(): cv.declare_id(MQTTClientComponent),
             cv.Required(CONF_BROKER): cv.string_strict,
+            cv.Optional(CONF_ENABLE_ON_BOOT, default=True): cv.boolean,
             cv.Optional(CONF_PORT, default=1883): cv.port,
             cv.Optional(CONF_USERNAME, default=""): cv.string,
             cv.Optional(CONF_PASSWORD, default=""): cv.string,
@@ -325,6 +329,7 @@ async def to_code(config):
     cg.add_global(mqtt_ns.using)
 
     cg.add(var.set_broker_address(config[CONF_BROKER]))
+    cg.add(var.set_enable_on_boot(config[CONF_ENABLE_ON_BOOT]))
     cg.add(var.set_broker_port(config[CONF_PORT]))
     cg.add(var.set_username(config[CONF_USERNAME]))
     cg.add(var.set_password(config[CONF_PASSWORD]))
@@ -555,3 +560,31 @@ async def register_mqtt_component(var, config):
 async def mqtt_connected_to_code(config, condition_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(condition_id, template_arg, paren)
+
+
+@automation.register_action(
+    "mqtt.enable",
+    MQTTEnableAction,
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.use_id(MQTTClientComponent),
+        }
+    ),
+)
+async def mqtt_enable_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    return cg.new_Pvariable(action_id, template_arg, paren)
+
+
+@automation.register_action(
+    "mqtt.disable",
+    MQTTDisableAction,
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.use_id(MQTTClientComponent),
+        }
+    ),
+)
+async def mqtt_disable_to_code(config, action_id, template_arg, args):
+    paren = await cg.get_variable(config[CONF_ID])
+    return cg.new_Pvariable(action_id, template_arg, paren)
diff --git a/esphome/components/mqtt/mqtt_client.cpp b/esphome/components/mqtt/mqtt_client.cpp
index b5ac285026..106192c0e3 100644
--- a/esphome/components/mqtt/mqtt_client.cpp
+++ b/esphome/components/mqtt/mqtt_client.cpp
@@ -50,6 +50,8 @@ void MQTTClientComponent::setup() {
         }
       });
   this->mqtt_backend_.set_on_disconnect([this](MQTTClientDisconnectReason reason) {
+    if (this->state_ == MQTT_CLIENT_DISABLED)
+      return;
     this->state_ = MQTT_CLIENT_DISCONNECTED;
     this->disconnect_reason_ = reason;
   });
@@ -77,8 +79,9 @@ void MQTTClientComponent::setup() {
         topic, [this](const std::string &topic, const std::string &payload) { this->send_device_info_(); }, 2);
   }
 
-  this->last_connected_ = millis();
-  this->start_dnslookup_();
+  if (this->enable_on_boot_) {
+    this->enable();
+  }
 }
 
 void MQTTClientComponent::send_device_info_() {
@@ -163,7 +166,9 @@ void MQTTClientComponent::dump_config() {
     ESP_LOGCONFIG(TAG, "  Availability: '%s'", this->availability_.topic.c_str());
   }
 }
-bool MQTTClientComponent::can_proceed() { return network::is_disabled() || this->is_connected(); }
+bool MQTTClientComponent::can_proceed() {
+  return network::is_disabled() || this->state_ == MQTT_CLIENT_DISABLED || this->is_connected();
+}
 
 void MQTTClientComponent::start_dnslookup_() {
   for (auto &subscription : this->subscriptions_) {
@@ -339,6 +344,8 @@ void MQTTClientComponent::loop() {
   const uint32_t now = millis();
 
   switch (this->state_) {
+    case MQTT_CLIENT_DISABLED:
+      return;  // Return to avoid a reboot when disabled
     case MQTT_CLIENT_DISCONNECTED:
       if (now - this->connect_begin_ > 5000) {
         this->start_dnslookup_();
@@ -501,6 +508,23 @@ bool MQTTClientComponent::publish_json(const std::string &topic, const json::jso
   return this->publish(topic, message, qos, retain);
 }
 
+void MQTTClientComponent::enable() {
+  if (this->state_ != MQTT_CLIENT_DISABLED)
+    return;
+  ESP_LOGD(TAG, "Enabling MQTT...");
+  this->state_ = MQTT_CLIENT_DISCONNECTED;
+  this->last_connected_ = millis();
+  this->start_dnslookup_();
+}
+
+void MQTTClientComponent::disable() {
+  if (this->state_ == MQTT_CLIENT_DISABLED)
+    return;
+  ESP_LOGD(TAG, "Disabling MQTT...");
+  this->state_ = MQTT_CLIENT_DISABLED;
+  this->on_shutdown();
+}
+
 /** Check if the message topic matches the given subscription topic
  *
  * INFO: MQTT spec mandates that topics must not be empty and must be valid NULL-terminated UTF-8 strings.
diff --git a/esphome/components/mqtt/mqtt_client.h b/esphome/components/mqtt/mqtt_client.h
index 887800f201..7ae3a6c5e8 100644
--- a/esphome/components/mqtt/mqtt_client.h
+++ b/esphome/components/mqtt/mqtt_client.h
@@ -87,7 +87,8 @@ struct MQTTDiscoveryInfo {
 };
 
 enum MQTTClientState {
-  MQTT_CLIENT_DISCONNECTED = 0,
+  MQTT_CLIENT_DISABLED = 0,
+  MQTT_CLIENT_DISCONNECTED,
   MQTT_CLIENT_RESOLVING_ADDRESS,
   MQTT_CLIENT_CONNECTING,
   MQTT_CLIENT_CONNECTED,
@@ -247,6 +248,9 @@ class MQTTClientComponent : public Component {
   void register_mqtt_component(MQTTComponent *component);
 
   bool is_connected();
+  void set_enable_on_boot(bool enable_on_boot) { this->enable_on_boot_ = enable_on_boot; }
+  void enable();
+  void disable();
 
   void on_shutdown() override;
 
@@ -314,10 +318,11 @@ class MQTTClientComponent : public Component {
   MQTTBackendLibreTiny mqtt_backend_;
 #endif
 
-  MQTTClientState state_{MQTT_CLIENT_DISCONNECTED};
+  MQTTClientState state_{MQTT_CLIENT_DISABLED};
   network::IPAddress ip_;
   bool dns_resolved_{false};
   bool dns_resolve_error_{false};
+  bool enable_on_boot_{true};
   std::vector<MQTTComponent *> children_;
   uint32_t reboot_timeout_{300000};
   uint32_t connect_begin_;
@@ -414,6 +419,26 @@ template<typename... Ts> class MQTTConnectedCondition : public Condition<Ts...>
   MQTTClientComponent *parent_;
 };
 
+template<typename... Ts> class MQTTEnableAction : public Action<Ts...> {
+ public:
+  MQTTEnableAction(MQTTClientComponent *parent) : parent_(parent) {}
+
+  void play(Ts... x) override { this->parent_->enable(); }
+
+ protected:
+  MQTTClientComponent *parent_;
+};
+
+template<typename... Ts> class MQTTDisableAction : public Action<Ts...> {
+ public:
+  MQTTDisableAction(MQTTClientComponent *parent) : parent_(parent) {}
+
+  void play(Ts... x) override { this->parent_->disable(); }
+
+ protected:
+  MQTTClientComponent *parent_;
+};
+
 }  // namespace mqtt
 }  // namespace esphome
 
diff --git a/tests/components/mqtt/common.yaml b/tests/components/mqtt/common.yaml
index 75c34bec56..d22fe9579f 100644
--- a/tests/components/mqtt/common.yaml
+++ b/tests/components/mqtt/common.yaml
@@ -10,6 +10,7 @@ mqtt:
   port: 1883
   username: debug
   password: debug
+  enable_on_boot: false
   clean_session: True
   client_id: someclient
   use_abbreviations: false
@@ -87,6 +88,8 @@ button:
     state_topic: some/topic/button
     qos: 2
     on_press:
+      - mqtt.disable
+      - mqtt.enable
       - mqtt.publish:
           topic: some/topic/button
           payload: Hello