diff --git a/esphome/components/esp32_ble_tracker/__init__.py b/esphome/components/esp32_ble_tracker/__init__.py
index 4e25434aad..37e74672ed 100644
--- a/esphome/components/esp32_ble_tracker/__init__.py
+++ b/esphome/components/esp32_ble_tracker/__init__.py
@@ -5,7 +5,7 @@ import logging
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import esp32_ble
+from esphome.components import esp32_ble, ota
 from esphome.components.esp32 import add_idf_sdkconfig_option
 from esphome.components.esp32_ble import (
     IDF_MAX_CONNECTIONS,
@@ -328,7 +328,7 @@ async def to_code(config):
     # Note: CONFIG_BT_ACL_CONNECTIONS and CONFIG_BTDM_CTRL_BLE_MAX_CONN are now
     # configured in esp32_ble component based on max_connections setting
 
-    cg.add_define("USE_OTA_STATE_CALLBACK")  # To be notified when an OTA update starts
+    ota.request_ota_state_listeners()  # To be notified when an OTA update starts
     cg.add_define("USE_ESP32_BLE_CLIENT")
 
     CORE.add_job(_add_ble_features)
diff --git a/esphome/components/esphome/ota/ota_esphome.cpp b/esphome/components/esphome/ota/ota_esphome.cpp
index eb6c61a69b..469c57211c 100644
--- a/esphome/components/esphome/ota/ota_esphome.cpp
+++ b/esphome/components/esphome/ota/ota_esphome.cpp
@@ -42,7 +42,7 @@ static constexpr size_t SHA256_HEX_SIZE = 64;  // SHA256 hash as hex string (32
 #endif  // USE_OTA_PASSWORD
 
 void ESPHomeOTAComponent::setup() {
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
   ota::register_ota_platform(this);
 #endif
 
@@ -298,8 +298,8 @@ void ESPHomeOTAComponent::handle_data_() {
   // accidentally trigger the update process.
   this->log_start_(LOG_STR("update"));
   this->status_set_warning();
-#ifdef USE_OTA_STATE_CALLBACK
-  this->state_callback_.call(ota::OTA_STARTED, 0.0f, 0);
+#ifdef USE_OTA_STATE_LISTENER
+  this->notify_state_(ota::OTA_STARTED, 0.0f, 0);
 #endif
 
   // This will block for a few seconds as it locks flash
@@ -358,8 +358,8 @@ void ESPHomeOTAComponent::handle_data_() {
       last_progress = now;
       float percentage = (total * 100.0f) / ota_size;
       ESP_LOGD(TAG, "Progress: %0.1f%%", percentage);
-#ifdef USE_OTA_STATE_CALLBACK
-      this->state_callback_.call(ota::OTA_IN_PROGRESS, percentage, 0);
+#ifdef USE_OTA_STATE_LISTENER
+      this->notify_state_(ota::OTA_IN_PROGRESS, percentage, 0);
 #endif
       // feed watchdog and give other tasks a chance to run
       this->yield_and_feed_watchdog_();
@@ -388,8 +388,8 @@ void ESPHomeOTAComponent::handle_data_() {
   delay(10);
   ESP_LOGI(TAG, "Update complete");
   this->status_clear_warning();
-#ifdef USE_OTA_STATE_CALLBACK
-  this->state_callback_.call(ota::OTA_COMPLETED, 100.0f, 0);
+#ifdef USE_OTA_STATE_LISTENER
+  this->notify_state_(ota::OTA_COMPLETED, 100.0f, 0);
 #endif
   delay(100);  // NOLINT
   App.safe_reboot();
@@ -403,8 +403,8 @@ error:
   }
 
   this->status_momentary_error("onerror", 5000);
-#ifdef USE_OTA_STATE_CALLBACK
-  this->state_callback_.call(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
+#ifdef USE_OTA_STATE_LISTENER
+  this->notify_state_(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
 #endif
 }
 
diff --git a/esphome/components/http_request/ota/ota_http_request.cpp b/esphome/components/http_request/ota/ota_http_request.cpp
index 4d9e868c74..2a52a0e264 100644
--- a/esphome/components/http_request/ota/ota_http_request.cpp
+++ b/esphome/components/http_request/ota/ota_http_request.cpp
@@ -18,7 +18,7 @@ namespace http_request {
 static const char *const TAG = "http_request.ota";
 
 void OtaHttpRequestComponent::setup() {
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
   ota::register_ota_platform(this);
 #endif
 }
@@ -49,24 +49,24 @@ void OtaHttpRequestComponent::flash() {
   }
 
   ESP_LOGI(TAG, "Starting update");
-#ifdef USE_OTA_STATE_CALLBACK
-  this->state_callback_.call(ota::OTA_STARTED, 0.0f, 0);
+#ifdef USE_OTA_STATE_LISTENER
+  this->notify_state_(ota::OTA_STARTED, 0.0f, 0);
 #endif
 
   auto ota_status = this->do_ota_();
 
   switch (ota_status) {
     case ota::OTA_RESPONSE_OK:
-#ifdef USE_OTA_STATE_CALLBACK
-      this->state_callback_.call(ota::OTA_COMPLETED, 100.0f, ota_status);
+#ifdef USE_OTA_STATE_LISTENER
+      this->notify_state_(ota::OTA_COMPLETED, 100.0f, ota_status);
 #endif
       delay(10);
       App.safe_reboot();
       break;
 
     default:
-#ifdef USE_OTA_STATE_CALLBACK
-      this->state_callback_.call(ota::OTA_ERROR, 0.0f, ota_status);
+#ifdef USE_OTA_STATE_LISTENER
+      this->notify_state_(ota::OTA_ERROR, 0.0f, ota_status);
 #endif
       this->md5_computed_.clear();  // will be reset at next attempt
       this->md5_expected_.clear();  // will be reset at next attempt
@@ -159,8 +159,8 @@ uint8_t OtaHttpRequestComponent::do_ota_() {
       last_progress = now;
       float percentage = container->get_bytes_read() * 100.0f / container->content_length;
       ESP_LOGD(TAG, "Progress: %0.1f%%", percentage);
-#ifdef USE_OTA_STATE_CALLBACK
-      this->state_callback_.call(ota::OTA_IN_PROGRESS, percentage, 0);
+#ifdef USE_OTA_STATE_LISTENER
+      this->notify_state_(ota::OTA_IN_PROGRESS, percentage, 0);
 #endif
     }
   }  // while
diff --git a/esphome/components/http_request/update/__init__.py b/esphome/components/http_request/update/__init__.py
index abb4b2a430..d84d80109a 100644
--- a/esphome/components/http_request/update/__init__.py
+++ b/esphome/components/http_request/update/__init__.py
@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import update
+from esphome.components import ota, update
 import esphome.config_validation as cv
 from esphome.const import CONF_SOURCE
 
@@ -38,6 +38,6 @@ async def to_code(config):
 
     cg.add(var.set_source_url(config[CONF_SOURCE]))
 
-    cg.add_define("USE_OTA_STATE_CALLBACK")
+    ota.request_ota_state_listeners()
 
     await cg.register_component(var, config)
diff --git a/esphome/components/http_request/update/http_request_update.cpp b/esphome/components/http_request/update/http_request_update.cpp
index c91b0eba73..ca4afa67d4 100644
--- a/esphome/components/http_request/update/http_request_update.cpp
+++ b/esphome/components/http_request/update/http_request_update.cpp
@@ -21,20 +21,26 @@ static const char *const TAG = "http_request.update";
 static const size_t MAX_READ_SIZE = 256;
 
 void HttpRequestUpdate::setup() {
-  this->ota_parent_->add_on_state_callback([this](ota::OTAState state, float progress, uint8_t err) {
-    if (state == ota::OTAState::OTA_IN_PROGRESS) {
-      this->state_ = update::UPDATE_STATE_INSTALLING;
-      this->update_info_.has_progress = true;
-      this->update_info_.progress = progress;
-      this->publish_state();
-    } else if (state == ota::OTAState::OTA_ABORT || state == ota::OTAState::OTA_ERROR) {
-      this->state_ = update::UPDATE_STATE_AVAILABLE;
-      this->status_set_error(LOG_STR("Failed to install firmware"));
-      this->publish_state();
-    }
-  });
+#ifdef USE_OTA_STATE_LISTENER
+  this->ota_parent_->add_state_listener(this);
+#endif
 }
 
+#ifdef USE_OTA_STATE_LISTENER
+void HttpRequestUpdate::on_ota_state(ota::OTAState state, float progress, uint8_t error) {
+  if (state == ota::OTAState::OTA_IN_PROGRESS) {
+    this->state_ = update::UPDATE_STATE_INSTALLING;
+    this->update_info_.has_progress = true;
+    this->update_info_.progress = progress;
+    this->publish_state();
+  } else if (state == ota::OTAState::OTA_ABORT || state == ota::OTAState::OTA_ERROR) {
+    this->state_ = update::UPDATE_STATE_AVAILABLE;
+    this->status_set_error(LOG_STR("Failed to install firmware"));
+    this->publish_state();
+  }
+}
+#endif
+
 void HttpRequestUpdate::update() {
 #ifdef USE_ESP32
   xTaskCreate(HttpRequestUpdate::update_task, "update_task", 8192, (void *) this, 1, &this->update_task_handle_);
diff --git a/esphome/components/http_request/update/http_request_update.h b/esphome/components/http_request/update/http_request_update.h
index e05fdb0cc2..937f6dbeb9 100644
--- a/esphome/components/http_request/update/http_request_update.h
+++ b/esphome/components/http_request/update/http_request_update.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include "esphome/core/component.h"
+#include "esphome/core/defines.h"
 #include "esphome/core/helpers.h"
 
 #include "esphome/components/http_request/http_request.h"
@@ -14,7 +15,11 @@
 namespace esphome {
 namespace http_request {
 
-class HttpRequestUpdate : public update::UpdateEntity, public PollingComponent {
+#ifdef USE_OTA_STATE_LISTENER
+class HttpRequestUpdate final : public update::UpdateEntity, public PollingComponent, public ota::OTAStateListener {
+#else
+class HttpRequestUpdate final : public update::UpdateEntity, public PollingComponent {
+#endif
  public:
   void setup() override;
   void update() override;
@@ -29,6 +34,10 @@ class HttpRequestUpdate : public update::UpdateEntity, public PollingComponent {
 
   float get_setup_priority() const override { return setup_priority::AFTER_WIFI; }
 
+#ifdef USE_OTA_STATE_LISTENER
+  void on_ota_state(ota::OTAState state, float progress, uint8_t error) override;
+#endif
+
  protected:
   HttpRequestComponent *request_parent_;
   OtaHttpRequestComponent *ota_parent_;
diff --git a/esphome/components/micro_wake_word/__init__.py b/esphome/components/micro_wake_word/__init__.py
index 575fb97799..0d478f749b 100644
--- a/esphome/components/micro_wake_word/__init__.py
+++ b/esphome/components/micro_wake_word/__init__.py
@@ -7,7 +7,7 @@ from urllib.parse import urljoin
 from esphome import automation, external_files, git
 from esphome.automation import register_action, register_condition
 import esphome.codegen as cg
-from esphome.components import esp32, microphone, socket
+from esphome.components import esp32, microphone, ota, socket
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_FILE,
@@ -452,7 +452,7 @@ async def to_code(config):
     cg.add(var.set_microphone_source(mic_source))
 
     cg.add_define("USE_MICRO_WAKE_WORD")
-    cg.add_define("USE_OTA_STATE_CALLBACK")
+    ota.request_ota_state_listeners()
 
     esp32.add_idf_component(name="espressif/esp-tflite-micro", ref="1.3.3~1")
 
diff --git a/esphome/components/ota/__init__.py b/esphome/components/ota/__init__.py
index eec39668db..387a307ab7 100644
--- a/esphome/components/ota/__init__.py
+++ b/esphome/components/ota/__init__.py
@@ -13,6 +13,8 @@ from esphome.const import (
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
 
+OTA_STATE_LISTENER_KEY = "ota_state_listener"
+
 CODEOWNERS = ["@esphome/core"]
 AUTO_LOAD = ["md5", "safe_mode"]
 
@@ -86,6 +88,7 @@ BASE_OTA_SCHEMA = cv.Schema(
 @coroutine_with_priority(CoroPriority.OTA_UPDATES)
 async def to_code(config):
     cg.add_define("USE_OTA")
+    CORE.add_job(final_step)
 
     if CORE.is_esp32 and CORE.using_arduino:
         cg.add_library("Update", None)
@@ -122,7 +125,24 @@ async def ota_to_code(var, config):
         await automation.build_automation(trigger, [(cg.uint8, "x")], conf)
         use_state_callback = True
     if use_state_callback:
-        cg.add_define("USE_OTA_STATE_CALLBACK")
+        request_ota_state_listeners()
+
+
+def request_ota_state_listeners() -> None:
+    """Request that OTA state listeners be compiled in.
+
+    Components that need to be notified about OTA state changes (start, progress,
+    complete, error) should call this function during their code generation.
+    This enables the add_state_listener() API on OTAComponent.
+    """
+    CORE.data[OTA_STATE_LISTENER_KEY] = True
+
+
+@coroutine_with_priority(CoroPriority.FINAL)
+async def final_step():
+    """Final code generation step to configure optional OTA features."""
+    if CORE.data.get(OTA_STATE_LISTENER_KEY, False):
+        cg.add_define("USE_OTA_STATE_LISTENER")
 
 
 FILTER_SOURCE_FILES = filter_source_files_from_platform(
diff --git a/esphome/components/ota/automation.h b/esphome/components/ota/automation.h
index 7e1a60f3ce..520cb293f0 100644
--- a/esphome/components/ota/automation.h
+++ b/esphome/components/ota/automation.h
@@ -1,5 +1,5 @@
 #pragma once
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
 #include "ota_backend.h"
 
 #include "esphome/core/automation.h"
@@ -7,69 +7,65 @@
 namespace esphome {
 namespace ota {
 
-class OTAStateChangeTrigger : public Trigger<OTAState> {
+class OTAStateChangeTrigger final : public Trigger<OTAState>, public OTAStateListener {
  public:
-  explicit OTAStateChangeTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (!parent->is_failed()) {
-        trigger(state);
-      }
-    });
+  explicit OTAStateChangeTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override { this->trigger(state); }
+};
+
+class OTAStartTrigger final : public Trigger<>, public OTAStateListener {
+ public:
+  explicit OTAStartTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override {
+    if (state == OTA_STARTED) {
+      this->trigger();
+    }
   }
 };
 
-class OTAStartTrigger : public Trigger<> {
+class OTAProgressTrigger final : public Trigger<float>, public OTAStateListener {
  public:
-  explicit OTAStartTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (state == OTA_STARTED && !parent->is_failed()) {
-        trigger();
-      }
-    });
+  explicit OTAProgressTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override {
+    if (state == OTA_IN_PROGRESS) {
+      this->trigger(progress);
+    }
   }
 };
 
-class OTAProgressTrigger : public Trigger<float> {
+class OTAEndTrigger final : public Trigger<>, public OTAStateListener {
  public:
-  explicit OTAProgressTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (state == OTA_IN_PROGRESS && !parent->is_failed()) {
-        trigger(progress);
-      }
-    });
+  explicit OTAEndTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override {
+    if (state == OTA_COMPLETED) {
+      this->trigger();
+    }
   }
 };
 
-class OTAEndTrigger : public Trigger<> {
+class OTAAbortTrigger final : public Trigger<>, public OTAStateListener {
  public:
-  explicit OTAEndTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (state == OTA_COMPLETED && !parent->is_failed()) {
-        trigger();
-      }
-    });
+  explicit OTAAbortTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override {
+    if (state == OTA_ABORT) {
+      this->trigger();
+    }
   }
 };
 
-class OTAAbortTrigger : public Trigger<> {
+class OTAErrorTrigger final : public Trigger<uint8_t>, public OTAStateListener {
  public:
-  explicit OTAAbortTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (state == OTA_ABORT && !parent->is_failed()) {
-        trigger();
-      }
-    });
-  }
-};
+  explicit OTAErrorTrigger(OTAComponent *parent) { parent->add_state_listener(this); }
 
-class OTAErrorTrigger : public Trigger<uint8_t> {
- public:
-  explicit OTAErrorTrigger(OTAComponent *parent) {
-    parent->add_on_state_callback([this, parent](OTAState state, float progress, uint8_t error) {
-      if (state == OTA_ERROR && !parent->is_failed()) {
-        trigger(error);
-      }
-    });
+  void on_ota_state(OTAState state, float progress, uint8_t error) override {
+    if (state == OTA_ERROR) {
+      this->trigger(error);
+    }
   }
 };
 
diff --git a/esphome/components/ota/ota_backend.cpp b/esphome/components/ota/ota_backend.cpp
index 30de4ec4b3..5f510b4f8b 100644
--- a/esphome/components/ota/ota_backend.cpp
+++ b/esphome/components/ota/ota_backend.cpp
@@ -3,7 +3,7 @@
 namespace esphome {
 namespace ota {
 
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
 OTAGlobalCallback *global_ota_callback{nullptr};  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
 
 OTAGlobalCallback *get_global_ota_callback() {
@@ -14,6 +14,10 @@ OTAGlobalCallback *get_global_ota_callback() {
 }
 
 void register_ota_platform(OTAComponent *ota_caller) { get_global_ota_callback()->register_ota(ota_caller); }
+
+void OTAComponentBridge::on_ota_state(OTAState state, float progress, uint8_t error) {
+  this->global_callback_->notify_global_listeners(state, progress, error, this->component_);
+}
 #endif
 
 }  // namespace ota
diff --git a/esphome/components/ota/ota_backend.h b/esphome/components/ota/ota_backend.h
index 64ee0b9f7c..e474316bb5 100644
--- a/esphome/components/ota/ota_backend.h
+++ b/esphome/components/ota/ota_backend.h
@@ -4,9 +4,7 @@
 #include "esphome/core/defines.h"
 #include "esphome/core/helpers.h"
 
-#ifdef USE_OTA_STATE_CALLBACK
-#include "esphome/core/automation.h"
-#endif
+#include <vector>
 
 namespace esphome {
 namespace ota {
@@ -60,62 +58,98 @@ class OTABackend {
   virtual bool supports_compression() = 0;
 };
 
-class OTAComponent : public Component {
-#ifdef USE_OTA_STATE_CALLBACK
+/** Listener interface for OTA state changes.
+ *
+ * Components can implement this interface to receive OTA state updates
+ * without the overhead of std::function callbacks.
+ */
+class OTAStateListener {
  public:
-  void add_on_state_callback(std::function<void(ota::OTAState, float, uint8_t)> &&callback) {
-    this->state_callback_.add(std::move(callback));
-  }
+  virtual void on_ota_state(OTAState state, float progress, uint8_t error) = 0;
+};
+
+class OTAComponent : public Component {
+#ifdef USE_OTA_STATE_LISTENER
+ public:
+  void add_state_listener(OTAStateListener *listener) { this->state_listeners_.push_back(listener); }
 
  protected:
-  /** Extended callback manager with deferred call support.
-   *
-   * This adds a call_deferred() method for thread-safe execution from other tasks.
-   */
-  class StateCallbackManager : public CallbackManager<void(OTAState, float, uint8_t)> {
-   public:
-    StateCallbackManager(OTAComponent *component) : component_(component) {}
-
-    /** Call callbacks with deferral to main loop (for thread safety).
-     *
-     * This should be used by OTA implementations that run in separate tasks
-     * (like web_server OTA) to ensure callbacks execute in the main loop.
-     */
-    void call_deferred(ota::OTAState state, float progress, uint8_t error) {
-      component_->defer([this, state, progress, error]() { this->call(state, progress, error); });
+  void notify_state_(OTAState state, float progress, uint8_t error) {
+    for (auto *listener : this->state_listeners_) {
+      listener->on_ota_state(state, progress, error);
     }
+  }
 
-   private:
-    OTAComponent *component_;
-  };
+  /** Notify state with deferral to main loop (for thread safety).
+   *
+   * This should be used by OTA implementations that run in separate tasks
+   * (like web_server OTA) to ensure listeners execute in the main loop.
+   */
+  void notify_state_deferred_(OTAState state, float progress, uint8_t error) {
+    this->defer([this, state, progress, error]() { this->notify_state_(state, progress, error); });
+  }
 
-  StateCallbackManager state_callback_{this};
+  std::vector<OTAStateListener *> state_listeners_;
 #endif
 };
 
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
+class OTAGlobalCallback;
+
+/** Listener interface for global OTA state changes (includes OTA component pointer).
+ *
+ * Used by OTAGlobalCallback to aggregate state from multiple OTA components.
+ */
+class OTAGlobalStateListener {
+ public:
+  virtual void on_ota_global_state(OTAState state, float progress, uint8_t error, OTAComponent *component) = 0;
+};
+
+/** Helper class to bridge per-component OTA state to global listeners.
+ *
+ * Each OTA component gets one of these registered as a listener. When that
+ * component fires state events, this bridge forwards them to all global listeners
+ * along with the component pointer.
+ */
+class OTAComponentBridge : public OTAStateListener {
+ public:
+  OTAComponentBridge(OTAGlobalCallback *global_callback, OTAComponent *component)
+      : global_callback_(global_callback), component_(component) {}
+
+  void on_ota_state(OTAState state, float progress, uint8_t error) override;
+
+ private:
+  OTAGlobalCallback *global_callback_;
+  OTAComponent *component_;
+};
+
 class OTAGlobalCallback {
  public:
   void register_ota(OTAComponent *ota_caller) {
-    ota_caller->add_on_state_callback([this, ota_caller](OTAState state, float progress, uint8_t error) {
-      this->state_callback_.call(state, progress, error, ota_caller);
-    });
+    // Create a bridge that forwards this component's events to global listeners
+    auto *bridge = new OTAComponentBridge(this, ota_caller);  // NOLINT(cppcoreguidelines-owning-memory)
+    ota_caller->add_state_listener(bridge);
   }
-  void add_on_state_callback(std::function<void(OTAState, float, uint8_t, OTAComponent *)> &&callback) {
-    this->state_callback_.add(std::move(callback));
+
+  void add_global_state_listener(OTAGlobalStateListener *listener) { this->global_listeners_.push_back(listener); }
+
+  void notify_global_listeners(OTAState state, float progress, uint8_t error, OTAComponent *component) {
+    for (auto *listener : this->global_listeners_) {
+      listener->on_ota_global_state(state, progress, error, component);
+    }
   }
 
  protected:
-  CallbackManager<void(OTAState, float, uint8_t, OTAComponent *)> state_callback_{};
+  std::vector<OTAGlobalStateListener *> global_listeners_;
 };
 
 OTAGlobalCallback *get_global_ota_callback();
 void register_ota_platform(OTAComponent *ota_caller);
 
 // OTA implementations should use:
-// - state_callback_.call() when already in main loop (e.g., esphome OTA)
-// - state_callback_.call_deferred() when in separate task (e.g., web_server OTA)
-// This ensures proper callback execution in all contexts.
+// - notify_state_() when already in main loop (e.g., esphome OTA)
+// - notify_state_deferred_() when in separate task (e.g., web_server OTA)
+// This ensures proper listener execution in all contexts.
 #endif
 std::unique_ptr<ota::OTABackend> make_ota_backend();
 
diff --git a/esphome/components/speaker/media_player/__init__.py b/esphome/components/speaker/media_player/__init__.py
index 062bff92f8..4ca57f2c4a 100644
--- a/esphome/components/speaker/media_player/__init__.py
+++ b/esphome/components/speaker/media_player/__init__.py
@@ -6,7 +6,7 @@ from pathlib import Path
 
 from esphome import automation, external_files
 import esphome.codegen as cg
-from esphome.components import audio, esp32, media_player, network, psram, speaker
+from esphome.components import audio, esp32, media_player, network, ota, psram, speaker
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BUFFER_SIZE,
@@ -342,7 +342,7 @@ async def to_code(config):
     var = await media_player.new_media_player(config)
     await cg.register_component(var, config)
 
-    cg.add_define("USE_OTA_STATE_CALLBACK")
+    ota.request_ota_state_listeners()
 
     cg.add(var.set_buffer_size(config[CONF_BUFFER_SIZE]))
 
diff --git a/esphome/components/web_server/ota/ota_web_server.cpp b/esphome/components/web_server/ota/ota_web_server.cpp
index 7929f3647f..30c4a59b8b 100644
--- a/esphome/components/web_server/ota/ota_web_server.cpp
+++ b/esphome/components/web_server/ota/ota_web_server.cpp
@@ -84,9 +84,9 @@ void OTARequestHandler::report_ota_progress_(AsyncWebServerRequest *request) {
     } else {
       ESP_LOGD(TAG, "OTA in progress: %" PRIu32 " bytes read", this->ota_read_length_);
     }
-#ifdef USE_OTA_STATE_CALLBACK
-    // Report progress - use call_deferred since we're in web server task
-    this->parent_->state_callback_.call_deferred(ota::OTA_IN_PROGRESS, percentage, 0);
+#ifdef USE_OTA_STATE_LISTENER
+    // Report progress - use notify_state_deferred_ since we're in web server task
+    this->parent_->notify_state_deferred_(ota::OTA_IN_PROGRESS, percentage, 0);
 #endif
     this->last_ota_progress_ = now;
   }
@@ -114,9 +114,9 @@ void OTARequestHandler::handleUpload(AsyncWebServerRequest *request, const Platf
     // Initialize OTA on first call
     this->ota_init_(filename.c_str());
 
-#ifdef USE_OTA_STATE_CALLBACK
-    // Notify OTA started - use call_deferred since we're in web server task
-    this->parent_->state_callback_.call_deferred(ota::OTA_STARTED, 0.0f, 0);
+#ifdef USE_OTA_STATE_LISTENER
+    // Notify OTA started - use notify_state_deferred_ since we're in web server task
+    this->parent_->notify_state_deferred_(ota::OTA_STARTED, 0.0f, 0);
 #endif
 
     // Platform-specific pre-initialization
@@ -134,9 +134,9 @@ void OTARequestHandler::handleUpload(AsyncWebServerRequest *request, const Platf
     this->ota_backend_ = ota::make_ota_backend();
     if (!this->ota_backend_) {
       ESP_LOGE(TAG, "Failed to create OTA backend");
-#ifdef USE_OTA_STATE_CALLBACK
-      this->parent_->state_callback_.call_deferred(ota::OTA_ERROR, 0.0f,
-                                                   static_cast<uint8_t>(ota::OTA_RESPONSE_ERROR_UNKNOWN));
+#ifdef USE_OTA_STATE_LISTENER
+      this->parent_->notify_state_deferred_(ota::OTA_ERROR, 0.0f,
+                                            static_cast<uint8_t>(ota::OTA_RESPONSE_ERROR_UNKNOWN));
 #endif
       return;
     }
@@ -148,8 +148,8 @@ void OTARequestHandler::handleUpload(AsyncWebServerRequest *request, const Platf
     if (error_code != ota::OTA_RESPONSE_OK) {
       ESP_LOGE(TAG, "OTA begin failed: %d", error_code);
       this->ota_backend_.reset();
-#ifdef USE_OTA_STATE_CALLBACK
-      this->parent_->state_callback_.call_deferred(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
+#ifdef USE_OTA_STATE_LISTENER
+      this->parent_->notify_state_deferred_(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
 #endif
       return;
     }
@@ -166,8 +166,8 @@ void OTARequestHandler::handleUpload(AsyncWebServerRequest *request, const Platf
       ESP_LOGE(TAG, "OTA write failed: %d", error_code);
       this->ota_backend_->abort();
       this->ota_backend_.reset();
-#ifdef USE_OTA_STATE_CALLBACK
-      this->parent_->state_callback_.call_deferred(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
+#ifdef USE_OTA_STATE_LISTENER
+      this->parent_->notify_state_deferred_(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
 #endif
       return;
     }
@@ -186,15 +186,15 @@ void OTARequestHandler::handleUpload(AsyncWebServerRequest *request, const Platf
     error_code = this->ota_backend_->end();
     if (error_code == ota::OTA_RESPONSE_OK) {
       this->ota_success_ = true;
-#ifdef USE_OTA_STATE_CALLBACK
-      // Report completion before reboot - use call_deferred since we're in web server task
-      this->parent_->state_callback_.call_deferred(ota::OTA_COMPLETED, 100.0f, 0);
+#ifdef USE_OTA_STATE_LISTENER
+      // Report completion before reboot - use notify_state_deferred_ since we're in web server task
+      this->parent_->notify_state_deferred_(ota::OTA_COMPLETED, 100.0f, 0);
 #endif
       this->schedule_ota_reboot_();
     } else {
       ESP_LOGE(TAG, "OTA end failed: %d", error_code);
-#ifdef USE_OTA_STATE_CALLBACK
-      this->parent_->state_callback_.call_deferred(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
+#ifdef USE_OTA_STATE_LISTENER
+      this->parent_->notify_state_deferred_(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(error_code));
 #endif
     }
     this->ota_backend_.reset();
@@ -232,7 +232,7 @@ void WebServerOTAComponent::setup() {
 
   // AsyncWebServer takes ownership of the handler and will delete it when the server is destroyed
   base->add_handler(new OTARequestHandler(this));  // NOLINT
-#ifdef USE_OTA_STATE_CALLBACK
+#ifdef USE_OTA_STATE_LISTENER
   // Register with global OTA callback system
   ota::register_ota_platform(this);
 #endif
diff --git a/esphome/core/defines.h b/esphome/core/defines.h
index f4026aad96..82963c767d 100644
--- a/esphome/core/defines.h
+++ b/esphome/core/defines.h
@@ -139,7 +139,7 @@
 #define USE_OTA_PASSWORD
 #define USE_OTA_SHA256
 #define ALLOW_OTA_DOWNGRADE_MD5
-#define USE_OTA_STATE_CALLBACK
+#define USE_OTA_STATE_LISTENER
 #define USE_OTA_VERSION 2
 #define USE_TIME_TIMEZONE
 #define USE_WIFI