diff --git a/libervia/backend/bridge/bridge_constructor/constructors/embedded/constructor.py b/libervia/backend/bridge/bridge_constructor/constructors/embedded/constructor.py
index 98cb9914..a56e9ba9 100644
--- a/libervia/backend/bridge/bridge_constructor/constructors/embedded/constructor.py
+++ b/libervia/backend/bridge/bridge_constructor/constructors/embedded/constructor.py
@@ -66,9 +66,7 @@ class EmbeddedConstructor(base_constructor.Constructor):
         d = self._methods_cbs["{name}"]({args_result})
         if callback is not None:
             d.addCallback({cb_or_lambda})
-        if errback is None:
-            d.addErrback(lambda failure_: log.error(failure_))
-        else:
+        if errback is not None:
             d.addErrback(errback)
         return d
         """.format(**completion)
diff --git a/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_frontend_template.py b/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_frontend_template.py
index edab2f50..bd04dde5 100644
--- a/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_frontend_template.py
+++ b/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_frontend_template.py
@@ -17,4 +17,4 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from libervia.backend.bridge.embedded import bridge
+from libervia.backend.bridge.embedded import bridge, TwistedBridge, AIOBridge
diff --git a/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_template.py b/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_template.py
index c299d80d..45c6be32 100644
--- a/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_template.py
+++ b/libervia/backend/bridge/bridge_constructor/constructors/embedded/embedded_template.py
@@ -21,11 +21,16 @@ from libervia.backend.core.log import getLogger
 
 log = getLogger(__name__)
 from libervia.backend.core import exceptions
+from libervia.frontends.bridge.bridge_frontend import BridgeException
+
+import asyncio
+from twisted.internet import defer
+from twisted.internet.error import ReactorAlreadyInstalledError
 
 
-class _Bridge(object):
+class _Bridge:
     def __init__(self):
-        log.debug("Init embedded bridge...")
+        log.debug("Init embedded bridge…")
         self._methods_cbs = {}
         self._signals_cbs = {"core": {}, "plugin": {}}
 
@@ -55,9 +60,7 @@ class _Bridge(object):
             d = self._methods_cbs[name](*args, **kwargs)
             if callback is not None:
                 d.addCallback(callback if out_sign else lambda __: callback())
-            if errback is None:
-                d.addErrback(lambda failure_: log.error(failure_))
-            else:
+            if errback is not None:
                 d.addErrback(errback)
             return d
         else:
@@ -114,12 +117,118 @@ class _Bridge(object):
 
 ##METHODS_PART##
 
-# we want the same instance for both core and frontend
-bridge = None
+# Singleton: shared instance between backend and frontend
+_bridge_instance = None
 
 
 def bridge():
-    global bridge
-    if bridge is None:
-        bridge = _Bridge()
-    return bridge
+    """Return the singleton _Bridge instance, creating it if necessary."""
+    global _bridge_instance
+    if _bridge_instance is None:
+        _bridge_instance = _Bridge()
+    return _bridge_instance
+
+
+class TwistedBridge:
+    """Embedded bridge wrapper that returns Twisted Deferreds.
+
+    Starts ``LiberviaBackend`` on first bridge connection.
+    """
+
+    def __init__(self):
+        self._connected = False
+
+    async def post_init(self):
+        pass
+
+    def _ensure_reactor(self):
+        """Ensure asyncioreactor is installed."""
+        from twisted.internet import asyncioreactor
+
+        # Python 3.14+ workaround: ensure an event loop exists before installing reactor
+        try:
+            asyncio.get_event_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+        try:
+            asyncioreactor.install()
+        except ReactorAlreadyInstalledError:
+            from twisted.internet import reactor
+
+            if not isinstance(reactor, asyncioreactor.AsyncioSelectorReactor):
+                raise exceptions.InternalError(
+                    f"Wrong reactor installed: {reactor}"
+                ) from None
+
+    def _start_backend(self):
+        """Start LiberviaBackend and return its initialised Deferred."""
+        from libervia.backend.core.main import LiberviaBackend
+
+        self._ensure_reactor()
+        backend = LiberviaBackend()
+        backend._init()
+        return backend.initialised
+
+    def bridge_connect(self):
+        """Connect to the embedded backend, starting it if necessary."""
+        if self._connected:
+            return defer.succeed(None)
+        self._connected = True
+        return self._start_backend()
+
+    def register_signal(self, name, handler, iface="core"):
+        """Register a signal handler on the embedded bridge."""
+        bridge().register_signal(name, handler, iface)
+
+    def _errback(self, failure_):
+        """Convert Failure to BridgeException."""
+        e = failure_.value
+        if isinstance(e, BridgeException):
+            raise e
+        try:
+            condition = e.condition
+        except AttributeError:
+            condition = ""
+        raise BridgeException(
+            type(e).__module__ + "." + type(e).__name__, str(e), condition
+        ) from e
+
+    def __getattr__(self, name):
+        """Delegate method calls, ensuring they return Deferreds."""
+        attr = getattr(bridge(), name)
+        if callable(attr) and not name.startswith("_"):
+
+            def deferred_wrapper(*args, **kwargs):
+                result = attr(*args, **kwargs)
+                if isinstance(result, defer.Deferred):
+                    result.addErrback(self._errback)
+                    return result
+                return defer.succeed(result)
+
+            return deferred_wrapper
+        return attr
+
+
+class AIOBridge(TwistedBridge):
+    """Embedded bridge wrapper that returns asyncio Futures."""
+
+    def bridge_connect(self):
+        """Connect to the embedded backend, starting it if necessary."""
+        if self._connected:
+            return asyncio.ensure_future(defer.succeed(None))
+        self._connected = True
+        d = self._start_backend()
+        return d.asFuture(asyncio.get_event_loop())
+
+    def __getattr__(self, name):
+        """Delegate method calls, converting Deferreds to Futures."""
+        attr = super().__getattr__(name)
+        if callable(attr) and not name.startswith("_"):
+
+            def aio_wrapper(*args, **kwargs):
+                return attr(*args, **kwargs).asFuture(asyncio.get_event_loop())
+
+            return aio_wrapper
+        return attr
diff --git a/libervia/backend/bridge/embedded.py b/libervia/backend/bridge/embedded.py
new file mode 100644
index 00000000..4d1324c2
--- /dev/null
+++ b/libervia/backend/bridge/embedded.py
@@ -0,0 +1,1486 @@
+#!/usr/bin/env python3
+
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from libervia.backend.core.log import getLogger
+
+log = getLogger(__name__)
+from libervia.backend.core import exceptions
+from libervia.frontends.bridge.bridge_frontend import BridgeException
+
+import asyncio
+from twisted.internet import defer
+from twisted.internet.error import ReactorAlreadyInstalledError
+
+
+class _Bridge:
+    def __init__(self):
+        log.debug("Init embedded bridge…")
+        self._methods_cbs = {}
+        self._signals_cbs = {"core": {}, "plugin": {}}
+
+    def bridge_connect(self, callback, errback):
+        callback()
+
+    def register_method(self, name, callback):
+        log.debug("registering embedded bridge method [{}]".format(name))
+        if name in self._methods_cbs:
+            raise exceptions.ConflictError("method {} is already regitered".format(name))
+        self._methods_cbs[name] = callback
+
+    def register_signal(self, functionName, handler, iface="core"):
+        iface_dict = self._signals_cbs[iface]
+        if functionName in iface_dict:
+            raise exceptions.ConflictError(
+                "signal {name} is already regitered for interface {iface}".format(
+                    name=functionName, iface=iface
+                )
+            )
+        iface_dict[functionName] = handler
+
+    def call_method(self, name, out_sign, async_, args, kwargs):
+        callback = kwargs.pop("callback", None)
+        errback = kwargs.pop("errback", None)
+        if async_:
+            d = self._methods_cbs[name](*args, **kwargs)
+            if callback is not None:
+                d.addCallback(callback if out_sign else lambda __: callback())
+            if errback is not None:
+                d.addErrback(errback)
+            return d
+        else:
+            try:
+                ret = self._methods_cbs[name](*args, **kwargs)
+            except Exception as e:
+                if errback is not None:
+                    errback(e)
+                else:
+                    raise e
+            else:
+                if callback is None:
+                    return ret
+                else:
+                    if out_sign:
+                        callback(ret)
+                    else:
+                        callback()
+
+    def send_signal(self, name, args, kwargs):
+        try:
+            cb = self._signals_cbs["plugin"][name]
+        except KeyError:
+            log.debug("ignoring signal {}: no callback registered".format(name))
+        else:
+            cb(*args, **kwargs)
+
+    def add_method(
+        self, name, int_suffix, in_sign, out_sign, method, async_=False, doc={}
+    ):
+        # FIXME: doc parameter is kept only temporary, the time to remove it from calls
+        log.debug("Adding method [{}] to embedded bridge".format(name))
+        self.register_method(name, method)
+        setattr(
+            self.__class__,
+            name,
+            lambda self_, *args, **kwargs: self.call_method(
+                name, out_sign, async_, args, kwargs
+            ),
+        )
+
+    def add_signal(self, name, int_suffix, signature, doc={}):
+        setattr(
+            self.__class__,
+            name,
+            lambda self_, *args, **kwargs: self.send_signal(name, args, kwargs),
+        )
+
+    ## signals ##
+
+    def _debug(self, action, params, profile):
+        try:
+            cb = self._signals_cbs["core"]["_debug"]
+        except KeyError:
+            log.warning("ignoring signal _debug: no callback registered")
+        else:
+            cb(action, params, profile)
+
+    def action_new(self, action_data, id, security_limit, profile):
+        try:
+            cb = self._signals_cbs["core"]["action_new"]
+        except KeyError:
+            log.warning("ignoring signal action_new: no callback registered")
+        else:
+            cb(action_data, id, security_limit, profile)
+
+    def connected(self, jid_s, profile):
+        try:
+            cb = self._signals_cbs["core"]["connected"]
+        except KeyError:
+            log.warning("ignoring signal connected: no callback registered")
+        else:
+            cb(jid_s, profile)
+
+    def contact_deleted(self, entity_jid, profile):
+        try:
+            cb = self._signals_cbs["core"]["contact_deleted"]
+        except KeyError:
+            log.warning("ignoring signal contact_deleted: no callback registered")
+        else:
+            cb(entity_jid, profile)
+
+    def contact_new(self, contact_jid, attributes, groups, profile):
+        try:
+            cb = self._signals_cbs["core"]["contact_new"]
+        except KeyError:
+            log.warning("ignoring signal contact_new: no callback registered")
+        else:
+            cb(contact_jid, attributes, groups, profile)
+
+    def disconnected(self, profile):
+        try:
+            cb = self._signals_cbs["core"]["disconnected"]
+        except KeyError:
+            log.warning("ignoring signal disconnected: no callback registered")
+        else:
+            cb(profile)
+
+    def entity_data_updated(self, jid, name, value, profile):
+        try:
+            cb = self._signals_cbs["core"]["entity_data_updated"]
+        except KeyError:
+            log.warning("ignoring signal entity_data_updated: no callback registered")
+        else:
+            cb(jid, name, value, profile)
+
+    def message_encryption_started(self, to_jid, encryption_data, profile_key):
+        try:
+            cb = self._signals_cbs["core"]["message_encryption_started"]
+        except KeyError:
+            log.warning(
+                "ignoring signal message_encryption_started: no callback registered"
+            )
+        else:
+            cb(to_jid, encryption_data, profile_key)
+
+    def message_encryption_stopped(self, to_jid, encryption_data, profile_key):
+        try:
+            cb = self._signals_cbs["core"]["message_encryption_stopped"]
+        except KeyError:
+            log.warning(
+                "ignoring signal message_encryption_stopped: no callback registered"
+            )
+        else:
+            cb(to_jid, encryption_data, profile_key)
+
+    def message_new(
+        self,
+        uid,
+        timestamp,
+        from_jid,
+        to_jid,
+        message,
+        subject,
+        mess_type,
+        extra,
+        profile,
+    ):
+        try:
+            cb = self._signals_cbs["core"]["message_new"]
+        except KeyError:
+            log.warning("ignoring signal message_new: no callback registered")
+        else:
+            cb(
+                uid,
+                timestamp,
+                from_jid,
+                to_jid,
+                message,
+                subject,
+                mess_type,
+                extra,
+                profile,
+            )
+
+    def message_update(self, uid, message_type, message_data, profile):
+        try:
+            cb = self._signals_cbs["core"]["message_update"]
+        except KeyError:
+            log.warning("ignoring signal message_update: no callback registered")
+        else:
+            cb(uid, message_type, message_data, profile)
+
+    def messages_new(self, messages, profile):
+        try:
+            cb = self._signals_cbs["core"]["messages_new"]
+        except KeyError:
+            log.warning("ignoring signal messages_new: no callback registered")
+        else:
+            cb(messages, profile)
+
+    def notification_deleted(self, id, profile):
+        try:
+            cb = self._signals_cbs["core"]["notification_deleted"]
+        except KeyError:
+            log.warning("ignoring signal notification_deleted: no callback registered")
+        else:
+            cb(id, profile)
+
+    def notification_new(
+        self,
+        id,
+        timestamp,
+        type,
+        body_plain,
+        body_rich,
+        title,
+        requires_action,
+        priority,
+        expire_at,
+        extra,
+        profile,
+    ):
+        try:
+            cb = self._signals_cbs["core"]["notification_new"]
+        except KeyError:
+            log.warning("ignoring signal notification_new: no callback registered")
+        else:
+            cb(
+                id,
+                timestamp,
+                type,
+                body_plain,
+                body_rich,
+                title,
+                requires_action,
+                priority,
+                expire_at,
+                extra,
+                profile,
+            )
+
+    def param_update(self, name, value, category, profile):
+        try:
+            cb = self._signals_cbs["core"]["param_update"]
+        except KeyError:
+            log.warning("ignoring signal param_update: no callback registered")
+        else:
+            cb(name, value, category, profile)
+
+    def presence_update(self, entity_jid, show, priority, statuses, profile):
+        try:
+            cb = self._signals_cbs["core"]["presence_update"]
+        except KeyError:
+            log.warning("ignoring signal presence_update: no callback registered")
+        else:
+            cb(entity_jid, show, priority, statuses, profile)
+
+    def profile_connection_error(self, error, profile):
+        try:
+            cb = self._signals_cbs["core"]["profile_connection_error"]
+        except KeyError:
+            log.warning(
+                "ignoring signal profile_connection_error: no callback registered"
+            )
+        else:
+            cb(error, profile)
+
+    def profile_ready(self, profile):
+        try:
+            cb = self._signals_cbs["core"]["profile_ready"]
+        except KeyError:
+            log.warning("ignoring signal profile_ready: no callback registered")
+        else:
+            cb(profile)
+
+    def progress_error(self, id, error, profile):
+        try:
+            cb = self._signals_cbs["core"]["progress_error"]
+        except KeyError:
+            log.warning("ignoring signal progress_error: no callback registered")
+        else:
+            cb(id, error, profile)
+
+    def progress_finished(self, id, metadata, profile):
+        try:
+            cb = self._signals_cbs["core"]["progress_finished"]
+        except KeyError:
+            log.warning("ignoring signal progress_finished: no callback registered")
+        else:
+            cb(id, metadata, profile)
+
+    def progress_started(self, id, metadata, profile):
+        try:
+            cb = self._signals_cbs["core"]["progress_started"]
+        except KeyError:
+            log.warning("ignoring signal progress_started: no callback registered")
+        else:
+            cb(id, metadata, profile)
+
+    def subscribe(self, sub_type, entity_jid, profile):
+        try:
+            cb = self._signals_cbs["core"]["subscribe"]
+        except KeyError:
+            log.warning("ignoring signal subscribe: no callback registered")
+        else:
+            cb(sub_type, entity_jid, profile)
+
+    ## methods ##
+
+    def action_launch(
+        self, callback_id, data, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        d = self._methods_cbs["action_launch"](callback_id, data, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def actions_get(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["actions_get"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def config_get(self, section, name, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["config_get"](section, name)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def connect(
+        self,
+        profile_key="@DEFAULT@",
+        password="",
+        options={},
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["connect"](profile_key, password, options)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def contact_add(
+        self, entity_jid, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["contact_add"](entity_jid, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def contact_del(
+        self, entity_jid, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        d = self._methods_cbs["contact_del"](entity_jid, profile_key)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def contact_get(self, arg_0, profile_key="@DEFAULT@", callback=None, errback=None):
+        d = self._methods_cbs["contact_get"](arg_0, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def contact_update(
+        self,
+        entity_jid,
+        name,
+        groups,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        try:
+            ret = self._methods_cbs["contact_update"](
+                entity_jid, name, groups, profile_key
+            )
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def contacts_get(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        d = self._methods_cbs["contacts_get"](profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def contacts_get_from_group(
+        self, group, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["contacts_get_from_group"](group, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def devices_infos_get(self, bare_jid, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["devices_infos_get"](bare_jid, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def disco_find_by_features(
+        self,
+        namespaces,
+        identities,
+        bare_jid=False,
+        service=True,
+        roster=True,
+        own_jid=True,
+        local_device=False,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["disco_find_by_features"](
+            namespaces,
+            identities,
+            bare_jid,
+            service,
+            roster,
+            own_jid,
+            local_device,
+            profile_key,
+        )
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def disco_infos(
+        self,
+        entity_jid,
+        node="",
+        use_cache=True,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["disco_infos"](entity_jid, node, use_cache, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def disco_items(
+        self,
+        entity_jid,
+        node="",
+        use_cache=True,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["disco_items"](entity_jid, node, use_cache, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def disconnect(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        d = self._methods_cbs["disconnect"](profile_key)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def encryption_namespace_get(self, arg_0, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["encryption_namespace_get"](arg_0)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def encryption_plugins_get(self, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["encryption_plugins_get"]()
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def encryption_trust_ui_get(
+        self, to_jid, namespace, profile_key, callback=None, errback=None
+    ):
+        d = self._methods_cbs["encryption_trust_ui_get"](to_jid, namespace, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def entities_data_get(self, jids, keys, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["entities_data_get"](jids, keys, profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def entity_data_get(self, jid, keys, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["entity_data_get"](jid, keys, profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def features_get(self, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["features_get"](profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def history_get(
+        self,
+        from_jid,
+        to_jid,
+        limit,
+        between=True,
+        filters="",
+        profile="@NONE@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["history_get"](
+            from_jid, to_jid, limit, between, filters, profile
+        )
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def image_check(self, arg_0, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["image_check"](arg_0)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def image_convert(self, source, dest, arg_2, extra, callback=None, errback=None):
+        d = self._methods_cbs["image_convert"](source, dest, arg_2, extra)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def image_generate_preview(
+        self, image_path, profile_key, callback=None, errback=None
+    ):
+        d = self._methods_cbs["image_generate_preview"](image_path, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def image_resize(self, image_path, width, height, callback=None, errback=None):
+        d = self._methods_cbs["image_resize"](image_path, width, height)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def init_pre_script(self, callback=None, errback=None):
+        d = self._methods_cbs["init_pre_script"]()
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def is_connected(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["is_connected"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def main_resource_get(
+        self, contact_jid, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["main_resource_get"](contact_jid, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def menu_help_get(self, menu_id, language, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["menu_help_get"](menu_id, language)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def menu_launch(
+        self,
+        menu_type,
+        path,
+        data,
+        security_limit,
+        profile_key,
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["menu_launch"](
+            menu_type, path, data, security_limit, profile_key
+        )
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def menus_get(self, language, security_limit, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["menus_get"](language, security_limit)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def message_encryption_get(self, to_jid, profile_key, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["message_encryption_get"](to_jid, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def message_encryption_start(
+        self,
+        to_jid,
+        namespace="",
+        replace=False,
+        profile_key="@NONE@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["message_encryption_start"](
+            to_jid, namespace, replace, profile_key
+        )
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def message_encryption_stop(self, to_jid, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["message_encryption_stop"](to_jid, profile_key)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def message_send(
+        self,
+        to_jid,
+        message,
+        subject={},
+        mess_type="auto",
+        extra={},
+        profile_key="@NONE@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["message_send"](
+            to_jid, message, subject, mess_type, extra, profile_key
+        )
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def namespaces_get(self, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["namespaces_get"]()
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def notification_add(
+        self,
+        type_,
+        body_plain,
+        body_rich,
+        title,
+        is_global,
+        requires_action,
+        arg_6,
+        priority,
+        expire_at,
+        extra,
+        callback=None,
+        errback=None,
+    ):
+        try:
+            ret = self._methods_cbs["notification_add"](
+                type_,
+                body_plain,
+                body_rich,
+                title,
+                is_global,
+                requires_action,
+                arg_6,
+                priority,
+                expire_at,
+                extra,
+            )
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def notification_delete(
+        self, id_, is_global, profile_key, callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["notification_delete"](id_, is_global, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def notifications_expired_clean(
+        self, limit_timestamp, profile_key, callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["notifications_expired_clean"](
+                limit_timestamp, profile_key
+            )
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def notifications_get(self, filters, profile_key, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["notifications_get"](filters, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def param_get_a(
+        self,
+        name,
+        category,
+        attribute="value",
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        try:
+            ret = self._methods_cbs["param_get_a"](name, category, attribute, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def param_get_a_async(
+        self,
+        name,
+        category,
+        attribute="value",
+        security_limit=-1,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["param_get_a_async"](
+            name, category, attribute, security_limit, profile_key
+        )
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def param_set(
+        self,
+        name,
+        value,
+        category,
+        security_limit=-1,
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        try:
+            ret = self._methods_cbs["param_set"](
+                name, value, category, security_limit, profile_key
+            )
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def param_ui_get(
+        self,
+        security_limit=-1,
+        app="",
+        extra="",
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["param_ui_get"](security_limit, app, extra, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def params_categories_get(self, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["params_categories_get"]()
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def params_register_app(
+        self, xml, security_limit=-1, app="", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["params_register_app"](xml, security_limit, app)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def params_template_load(self, filename, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["params_template_load"](filename)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def params_template_save(self, filename, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["params_template_save"](filename)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def params_values_from_category_get_async(
+        self,
+        category,
+        security_limit=-1,
+        app="",
+        extra="",
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        d = self._methods_cbs["params_values_from_category_get_async"](
+            category, security_limit, app, extra, profile_key
+        )
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def presence_set(
+        self,
+        to_jid="",
+        show="",
+        statuses={},
+        profile_key="@DEFAULT@",
+        callback=None,
+        errback=None,
+    ):
+        try:
+            ret = self._methods_cbs["presence_set"](to_jid, show, statuses, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def presence_statuses_get(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["presence_statuses_get"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def private_data_delete(self, namespace, key, arg_2, callback=None, errback=None):
+        d = self._methods_cbs["private_data_delete"](namespace, key, arg_2)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def private_data_get(self, namespace, key, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["private_data_get"](namespace, key, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def private_data_set(
+        self, namespace, key, data, profile_key, callback=None, errback=None
+    ):
+        d = self._methods_cbs["private_data_set"](namespace, key, data, profile_key)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def profile_create(
+        self, profile, password="", component="", callback=None, errback=None
+    ):
+        d = self._methods_cbs["profile_create"](profile, password, component)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def profile_delete_async(self, profile, callback=None, errback=None):
+        d = self._methods_cbs["profile_delete_async"](profile)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def profile_is_session_started(
+        self, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["profile_is_session_started"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def profile_name_get(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["profile_name_get"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def profile_set_default(self, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["profile_set_default"](profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def profile_start_session(
+        self, password="", profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        d = self._methods_cbs["profile_start_session"](password, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def profiles_list_get(
+        self, clients=True, components=False, callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["profiles_list_get"](clients, components)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def progress_get(self, id, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["progress_get"](id, profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def progress_get_all(self, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["progress_get_all"](profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def progress_get_all_metadata(self, profile, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["progress_get_all_metadata"](profile)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def ready_get(self, callback=None, errback=None):
+        d = self._methods_cbs["ready_get"]()
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def roster_resync(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        d = self._methods_cbs["roster_resync"](profile_key)
+        if callback is not None:
+            d.addCallback(lambda __: callback())
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def session_infos_get(self, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["session_infos_get"](profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def sub_waiting_get(self, profile_key="@DEFAULT@", callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["sub_waiting_get"](profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+    def subscription(
+        self, sub_type, entity, profile_key="@DEFAULT@", callback=None, errback=None
+    ):
+        try:
+            ret = self._methods_cbs["subscription"](sub_type, entity, profile_key)
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback()
+
+    def tor_infos_get(self, extra, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["tor_infos_get"](extra, profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def tor_mode_get(self, profile_key, callback=None, errback=None):
+        d = self._methods_cbs["tor_mode_get"](profile_key)
+        if callback is not None:
+            d.addCallback(callback)
+        if errback is not None:
+            d.addErrback(errback)
+        return d
+
+    def version_get(self, callback=None, errback=None):
+        try:
+            ret = self._methods_cbs["version_get"]()
+        except Exception as e:
+            if errback is not None:
+                errback(e)
+            else:
+                raise e
+        else:
+            if callback is None:
+                return ret
+            else:
+                callback(ret)
+
+
+# Singleton: shared instance between backend and frontend
+_bridge_instance = None
+
+
+def bridge():
+    """Return the singleton _Bridge instance, creating it if necessary."""
+    global _bridge_instance
+    if _bridge_instance is None:
+        _bridge_instance = _Bridge()
+    return _bridge_instance
+
+
+class TwistedBridge:
+    """Embedded bridge wrapper that returns Twisted Deferreds.
+
+    Starts ``LiberviaBackend`` on first bridge connection.
+    """
+
+    def __init__(self):
+        self._connected = False
+
+    async def post_init(self):
+        pass
+
+    def _ensure_reactor(self):
+        """Ensure asyncioreactor is installed."""
+        from twisted.internet import asyncioreactor
+
+        # Python 3.14+ workaround: ensure an event loop exists before installing reactor
+        try:
+            asyncio.get_event_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+        try:
+            asyncioreactor.install()
+        except ReactorAlreadyInstalledError:
+            from twisted.internet import reactor
+
+            if not isinstance(reactor, asyncioreactor.AsyncioSelectorReactor):
+                raise exceptions.InternalError(
+                    f"Wrong reactor installed: {reactor}"
+                ) from None
+
+    def _start_backend(self):
+        """Start LiberviaBackend and return its initialised Deferred."""
+        from libervia.backend.core.main import LiberviaBackend
+
+        self._ensure_reactor()
+        backend = LiberviaBackend()
+        backend._init()
+        return backend.initialised
+
+    def bridge_connect(self):
+        """Connect to the embedded backend, starting it if necessary."""
+        if self._connected:
+            return defer.succeed(None)
+        self._connected = True
+        return self._start_backend()
+
+    def register_signal(self, name, handler, iface="core"):
+        """Register a signal handler on the embedded bridge."""
+        bridge().register_signal(name, handler, iface)
+
+    def _errback(self, failure_):
+        """Convert Failure to BridgeException."""
+        e = failure_.value
+        if isinstance(e, BridgeException):
+            raise e
+        try:
+            condition = e.condition
+        except AttributeError:
+            condition = ""
+        raise BridgeException(
+            type(e).__module__ + "." + type(e).__name__, str(e), condition
+        ) from e
+
+    def __getattr__(self, name):
+        """Delegate method calls, ensuring they return Deferreds."""
+        attr = getattr(bridge(), name)
+        if callable(attr) and not name.startswith("_"):
+
+            def deferred_wrapper(*args, **kwargs):
+                result = attr(*args, **kwargs)
+                if isinstance(result, defer.Deferred):
+                    result.addErrback(self._errback)
+                    return result
+                return defer.succeed(result)
+
+            return deferred_wrapper
+        return attr
+
+
+class AIOBridge(TwistedBridge):
+    """Embedded bridge wrapper that returns asyncio Futures."""
+
+    def bridge_connect(self):
+        """Connect to the embedded backend, starting it if necessary."""
+        if self._connected:
+            return asyncio.ensure_future(defer.succeed(None))
+        self._connected = True
+        d = self._start_backend()
+        return d.asFuture(asyncio.get_event_loop())
+
+    def __getattr__(self, name):
+        """Delegate method calls, converting Deferreds to Futures."""
+        attr = super().__getattr__(name)
+        if callable(attr) and not name.startswith("_"):
+
+            def aio_wrapper(*args, **kwargs):
+                return attr(*args, **kwargs).asFuture(asyncio.get_event_loop())
+
+            return aio_wrapper
+        return attr
diff --git a/libervia/backend/tools/config.py b/libervia/backend/tools/config.py
index 75d14baf..7f87b1b2 100644
--- a/libervia/backend/tools/config.py
+++ b/libervia/backend/tools/config.py
@@ -232,8 +232,18 @@ class AutoBridgeSettings(ConfigOptions):
     type: Literal["auto"] = "auto"
 
 
+class EmbeddedBridgeSettings(ConfigOptions):
+    """Embedded Bridge.
+
+    Runs frontend and backend in the same process, backend is used like a library.
+    The backend is started automatically on first bridge connection.
+    """
+
+    type: Literal["embedded"] = "embedded"
+
+
 BridgeSettings = Annotated[
-    DBusBridgeSettings | PbBridgeSettings | AutoBridgeSettings,
+    DBusBridgeSettings | PbBridgeSettings | AutoBridgeSettings | EmbeddedBridgeSettings,
     Field(discriminator="type"),
 ]
 
diff --git a/libervia/frontends/bridge/embedded.py b/libervia/frontends/bridge/embedded.py
new file mode 100644
index 00000000..bd04dde5
--- /dev/null
+++ b/libervia/frontends/bridge/embedded.py
@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from libervia.backend.bridge.embedded import bridge, TwistedBridge, AIOBridge