diff --git a/homeassistant/components/automation/__init__.py b/homeassistant/components/automation/__init__.py
index 89a2817e236..fa242ac1557 100644
--- a/homeassistant/components/automation/__init__.py
+++ b/homeassistant/components/automation/__init__.py
@@ -707,7 +707,10 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
 
             @callback
             def started_action() -> None:
-                self.hass.bus.async_fire(
+                # This is always a callback from a coro so there is no
+                # risk of this running in a thread which allows us to use
+                # async_fire_internal
+                self.hass.bus.async_fire_internal(
                     EVENT_AUTOMATION_TRIGGERED, event_data, context=trigger_context
                 )
 
diff --git a/homeassistant/core.py b/homeassistant/core.py
index 8471d2c4dcc..01329806e61 100644
--- a/homeassistant/core.py
+++ b/homeassistant/core.py
@@ -506,8 +506,8 @@ class HomeAssistant:
         setattr(self.loop, "_thread_ident", threading.get_ident())
 
         self.set_state(CoreState.starting)
-        self.bus.async_fire(EVENT_CORE_CONFIG_UPDATE)
-        self.bus.async_fire(EVENT_HOMEASSISTANT_START)
+        self.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE)
+        self.bus.async_fire_internal(EVENT_HOMEASSISTANT_START)
 
         if not self._tasks:
             pending: set[asyncio.Future[Any]] | None = None
@@ -540,8 +540,8 @@ class HomeAssistant:
             return
 
         self.set_state(CoreState.running)
-        self.bus.async_fire(EVENT_CORE_CONFIG_UPDATE)
-        self.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
+        self.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE)
+        self.bus.async_fire_internal(EVENT_HOMEASSISTANT_STARTED)
 
     def add_job(
         self, target: Callable[[*_Ts], Any] | Coroutine[Any, Any, Any], *args: *_Ts
@@ -1115,7 +1115,7 @@ class HomeAssistant:
         self.exit_code = exit_code
 
         self.set_state(CoreState.stopping)
-        self.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
+        self.bus.async_fire_internal(EVENT_HOMEASSISTANT_STOP)
         try:
             async with self.timeout.async_timeout(STOP_STAGE_SHUTDOWN_TIMEOUT):
                 await self.async_block_till_done()
@@ -1128,7 +1128,7 @@ class HomeAssistant:
 
         # Stage 3 - Final write
         self.set_state(CoreState.final_write)
-        self.bus.async_fire(EVENT_HOMEASSISTANT_FINAL_WRITE)
+        self.bus.async_fire_internal(EVENT_HOMEASSISTANT_FINAL_WRITE)
         try:
             async with self.timeout.async_timeout(FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT):
                 await self.async_block_till_done()
@@ -1141,7 +1141,7 @@ class HomeAssistant:
 
         # Stage 4 - Close
         self.set_state(CoreState.not_running)
-        self.bus.async_fire(EVENT_HOMEASSISTANT_CLOSE)
+        self.bus.async_fire_internal(EVENT_HOMEASSISTANT_CLOSE)
 
         # Make a copy of running_tasks since a task can finish
         # while we are awaiting canceled tasks to get their result
@@ -1390,7 +1390,7 @@ class _OneTimeListener(Generic[_DataT]):
         return f"<_OneTimeListener {self.listener_job.target}>"
 
 
-# Empty list, used by EventBus._async_fire
+# Empty list, used by EventBus.async_fire_internal
 EMPTY_LIST: list[Any] = []
 
 
@@ -1455,10 +1455,12 @@ class EventBus:
             raise MaxLengthExceeded(
                 event_type, "event_type", MAX_LENGTH_EVENT_EVENT_TYPE
             )
-        return self._async_fire(event_type, event_data, origin, context, time_fired)
+        return self.async_fire_internal(
+            event_type, event_data, origin, context, time_fired
+        )
 
     @callback
-    def _async_fire(
+    def async_fire_internal(
         self,
         event_type: EventType[_DataT] | str,
         event_data: _DataT | None = None,
@@ -1466,7 +1468,12 @@ class EventBus:
         context: Context | None = None,
         time_fired: float | None = None,
     ) -> None:
-        """Fire an event.
+        """Fire an event, for internal use only.
+
+        This method is intended to only be used by core internally
+        and should not be considered a stable API. We will make
+        breaking change to this function in the future and it
+        should not be used in integrations.
 
         This method must be run in the event loop.
         """
@@ -2112,7 +2119,7 @@ class StateMachine:
             "old_state": old_state,
             "new_state": None,
         }
-        self._bus._async_fire(  # pylint: disable=protected-access
+        self._bus.async_fire_internal(
             EVENT_STATE_CHANGED,
             state_changed_data,
             context=context,
@@ -2225,7 +2232,7 @@ class StateMachine:
             # mypy does not understand this is only possible if old_state is not None
             old_last_reported = old_state.last_reported  # type: ignore[union-attr]
             old_state.last_reported = now  # type: ignore[union-attr]
-            self._bus._async_fire(  # pylint: disable=protected-access
+            self._bus.async_fire_internal(
                 EVENT_STATE_REPORTED,
                 {
                     "entity_id": entity_id,
@@ -2268,7 +2275,7 @@ class StateMachine:
             "old_state": old_state,
             "new_state": state,
         }
-        self._bus._async_fire(  # pylint: disable=protected-access
+        self._bus.async_fire_internal(
             EVENT_STATE_CHANGED,
             state_changed_data,
             context=context,
@@ -2622,7 +2629,7 @@ class ServiceRegistry:
             domain, service, processed_data, context, return_response
         )
 
-        self._hass.bus._async_fire(  # pylint: disable=protected-access
+        self._hass.bus.async_fire_internal(
             EVENT_CALL_SERVICE,
             {
                 ATTR_DOMAIN: domain,
@@ -2948,7 +2955,7 @@ class Config:
 
         self._update(source=ConfigSource.STORAGE, **kwargs)
         await self._async_store()
-        self.hass.bus.async_fire(EVENT_CORE_CONFIG_UPDATE, kwargs)
+        self.hass.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE, kwargs)
 
         _raise_issue_if_historic_currency(self.hass, self.currency)
         _raise_issue_if_no_country(self.hass, self.country)
diff --git a/homeassistant/data_entry_flow.py b/homeassistant/data_entry_flow.py
index 7e7019681af..f628879a7fd 100644
--- a/homeassistant/data_entry_flow.py
+++ b/homeassistant/data_entry_flow.py
@@ -442,7 +442,7 @@ class FlowManager(abc.ABC, Generic[_FlowResultT, _HandlerT]):
                 )
             ):
                 # Tell frontend to reload the flow state.
-                self.hass.bus.async_fire(
+                self.hass.bus.async_fire_internal(
                     EVENT_DATA_ENTRY_FLOW_PROGRESSED,
                     {"handler": flow.handler, "flow_id": flow_id, "refresh": True},
                 )
diff --git a/homeassistant/helpers/script.py b/homeassistant/helpers/script.py
index 62c781ae629..d925bf215ab 100644
--- a/homeassistant/helpers/script.py
+++ b/homeassistant/helpers/script.py
@@ -784,7 +784,7 @@ class _ScriptRun:
                 )
 
         trace_set_result(event=self._action[CONF_EVENT], event_data=event_data)
-        self._hass.bus.async_fire(
+        self._hass.bus.async_fire_internal(
             self._action[CONF_EVENT], event_data, context=self._context
         )
 
diff --git a/homeassistant/setup.py b/homeassistant/setup.py
index 5772fce6955..fab70e31d9d 100644
--- a/homeassistant/setup.py
+++ b/homeassistant/setup.py
@@ -459,7 +459,9 @@ async def _async_setup_component(
     # Cleanup
     hass.data[DATA_SETUP].pop(domain, None)
 
-    hass.bus.async_fire(EVENT_COMPONENT_LOADED, EventComponentLoaded(component=domain))
+    hass.bus.async_fire_internal(
+        EVENT_COMPONENT_LOADED, EventComponentLoaded(component=domain)
+    )
 
     return True