From 417ffb8fa23d9f01b9ad7f31c5d050652ae1bd98 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Tue, 18 Aug 2026 15:25:00 +0200
Subject: [PATCH 01/13] base of event logic

---
 ayon_api/events.py | 556 +++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 508 insertions(+), 48 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 79aac8223..2447f9d49 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -1,51 +1,511 @@
-import copy
+from __future__ import annotations
 
+from dataclasses import dataclass
+import inspect
+import logging
+import os
+import re
+from typing import Any, Callable
+import weakref
 
-class ServerEvent(object):
-    def __init__(
+
+@dataclass
+class Event:
+    topic: str
+
+
+@dataclass
+class ServerEvent(Event):
+    sender: str | None
+    event_hash: str | None
+    project_name: str | None
+    username: str | None
+    dependencies: list[str] | None
+    description: str | None
+    summary: str | None
+    payload: dict[str, Any] | None
+    finished: bool = True
+    store: bool = True
+
+
+def is_func_signature_supported(func, *args, **kwargs):
+    """Check if a function signature supports passed args and kwargs.
+
+    This check does not actually call the function, just look if function can
+    be called with the arguments.
+
+    Notes:
+        This does NOT check if the function would work with passed arguments
+            only if they can be passed in. If function have *args, **kwargs
+            in parameters, this will always return 'True'.
+
+    Example:
+        >>> def my_function(my_number):
+        ...     return my_number + 1
+        ...
+        >>> is_func_signature_supported(my_function, 1)
+        True
+        >>> is_func_signature_supported(my_function, 1, 2)
+        False
+        >>> is_func_signature_supported(my_function, my_number=1)
+        True
+        >>> is_func_signature_supported(my_function, number=1)
+        False
+        >>> is_func_signature_supported(my_function, "string")
+        True
+        >>> def my_other_function(*args, **kwargs):
+        ...     my_function(*args, **kwargs)
+        ...
+        >>> is_func_signature_supported(
+        ...     my_other_function,
+        ...     "string",
+        ...     1,
+        ...     other=None
+        ... )
+        True
+
+    Args:
+        func (Callable): A function where the signature should be tested.
+        *args (Any): Positional arguments for function signature.
+        **kwargs (Any): Keyword arguments for function signature.
+
+    Returns:
+        bool: Function can pass in arguments.
+
+    """
+    sig = inspect.signature(func)
+    try:
+        sig.bind(*args, **kwargs)
+        return True
+    except TypeError:
+        pass
+    return False
+
+
+def _get_func_ref(func: Callable) -> weakref.ref:
+    if inspect.ismethod(func):
+        return weakref.WeakMethod(func)
+    return weakref.ref(func)
+
+
+def _get_func_info(func: Callable) -> tuple[str, str]:
+    path = "<unknown path>"
+    if func is None:
+        return "<unknown>", path
+
+    if hasattr(func, "__name__"):
+        name = func.__name__
+    else:
+        name = str(func)
+
+    # Get path to file and fallback to '<unknown path>' if fails
+    # NOTE This was added because of 'partial' functions which is handled,
+    #   but who knows what else can cause this to fail?
+    try:
+        path = os.path.abspath(inspect.getfile(func))
+    except TypeError:
+        pass
+
+    return name, path
+
+
+class weakref_partial:
+    """Partial function with weak reference to the wrapped function.
+
+    Can be used as 'functools.partial' but it will store weak reference to
+        function. That means that the function must be reference counted
+        to avoid garbage collecting the function itself.
+
+        When the referenced functions is garbage collected then calling the
+        weakref partial (no matter the args/kwargs passed) will do nothing.
+        It will fail silently, returning `None`. The `is_valid()` method can
+        be used to detect whether the reference is still valid.
+
+    Is useful for object methods. In that case the callback is
+        deregistered when object is destroyed.
+
+    Warnings:
+        Values passed as *args and **kwargs are stored strongly in memory.
+            That may "keep alive" objects that should be already destroyed.
+            It is recommended to pass only immutable objects like 'str',
+            'bool', 'int' etc.
+
+    Args:
+        func (Callable): Function to wrap.
+        *args: Arguments passed to the wrapped function.
+        **kwargs: Keyword arguments passed to the wrapped function.
+    """
+
+    def __init__(self, func: Callable, *args, **kwargs) -> None:
+        self._func_ref: weakref.ref = _get_func_ref(func)
+        self._args: tuple = args
+        self._kwargs: dict = kwargs
+
+    def __call__(self, *args, **kwargs) -> Any:
+        func = self._func_ref()
+        if func is None:
+            return None
+
+        new_args = tuple(list(self._args) + list(args))
+        new_kwargs = dict(self._kwargs)
+        new_kwargs.update(kwargs)
+        return func(*new_args, **new_kwargs)
+
+    def get_func(self) -> Callable | None:
+        """Get wrapped function.
+
+        Returns:
+            Callable | None: Wrapped function or None if it was destroyed.
+
+        """
+        return self._func_ref()
+
+    def is_valid(self) -> bool:
+        """Check if wrapped function is still valid.
+
+        Returns:
+            bool: Is wrapped function still valid.
+
+        """
+        return self._func_ref() is not None
+
+    def validate_signature(self, *args, **kwargs) -> bool:
+        """Validate if passed arguments are supported by wrapped function.
+
+        Returns:
+            bool: Are passed arguments supported by wrapped function.
+
+        """
+        func = self._func_ref()
+        if func is None:
+            return False
+
+        new_args = tuple(list(self._args) + list(args))
+        new_kwargs = dict(self._kwargs)
+        new_kwargs.update(kwargs)
+        return is_func_signature_supported(
+            func, *new_args, **new_kwargs
+        )
+
+
+class EventCallback:
+    """Callback registered to a topic.
+
+    The callback function is registered to a topic. Topic is a string which
+    may contain '*' that will be handled as "any characters".
+
+    # Examples:
+    - "workfile.save"   Callback will be triggered if the event topic is
+                        exactly "workfile.save" .
+    - "workfile.*"      Callback will be triggered an event topic starts with
+                        "workfile." so "workfile.save" and "workfile.open"
+                        will trigger the callback.
+    - "*"               Callback will listen to all events.
+
+    Callback can be function or method. In both cases it should expect one
+    or none arguments. When 1 argument is expected then the processed 'Event'
+    object is passed in.
+
+    The callbacks are validated against their reference counter, that is
+        achieved using 'weakref' module. That means that the callback must
+        be stored in memory somewhere. e.g. lambda functions are not
+        supported as valid callback.
+
+    You can use 'weakref_partial' functions. In that case is partial object
+        stored in the callback object and reference counter is checked for
+        the wrapped function.
+
+    Args:
+        topic (str): Topic which will be listened.
+        func (Callable): Callback to a topic.
+        order (Union[int, None]): Order of callback. Lower number means higher
+            priority.
+
+    Raises:
+        TypeError: When passed function is not a callable object.
+    """
+
+    def __init__(self, topic: str, func: Callable, order: int) -> None:
+        if not callable(func):
+            raise TypeError(
+                f"Registered callback is not callable. \"{func}\""
+            )
+
+        self._validate_order(order)
+
+        self._log = None
+        self._topic: str = topic
+        self._order: int = order
+        self._enabled: bool = True
+        # Replace '*' with any character regex and escape rest of text
+        #   - when callback is registered for '*' topic it will receive all
+        #       events
+        #   - it is possible to register to a partial topis 'my.event.*'
+        #       - it will receive all matching event topics
+        #           e.g. 'my.event.start' and 'my.event.end'
+        topic_regex_str = "^{}$".format(
+            ".+".join(
+                re.escape(part)
+                for part in topic.split("*")
+            )
+        )
+        topic_regex = re.compile(topic_regex_str)
+        self._topic_regex: re.Pattern = topic_regex
+
+        # Callback function prep
+        if isinstance(func, weakref_partial):
+            partial_func = func
+            (name, path) = _get_func_info(func.get_func())
+            func_ref = None
+            expect_args = partial_func.validate_signature("fake")
+            expect_kwargs = partial_func.validate_signature(event="fake")
+
+        else:
+            partial_func = None
+            (name, path) = _get_func_info(func)
+            # Convert callback into references
+            #   - deleted functions won't cause crashes
+            func_ref = _get_func_ref(func)
+
+            # Get expected arguments from function spec
+            # - positional arguments are always preferred
+            expect_args = is_func_signature_supported(func, "fake")
+            expect_kwargs = is_func_signature_supported(func, event="fake")
+
+        self._func_ref: weakref.ref | None = func_ref
+        self._partial_func: weakref_partial | None = partial_func
+        self._ref_is_valid: bool = True
+        self._expect_args: bool = expect_args
+        self._expect_kwargs: bool = expect_kwargs
+
+        self._name: str = name
+        self._path: str = path
+
+    def __repr__(self) -> str:
+        return f"< {self.__class__.__name__} - {self._name} > {self._path}"
+
+    @property
+    def log(self) -> logging.Logger:
+        if self._log is None:
+            self._log = logging.getLogger(self.__class__.__name__)
+        return self._log
+
+    @property
+    def is_ref_valid(self) -> bool:
+        """
+
+        Returns:
+            bool: Is reference to callback valid.
+
+        """
+        self._validate_ref()
+        return self._ref_is_valid
+
+    def validate_ref(self) -> None:
+        """Validate if reference to callback is valid.
+
+        Deprecated:
+            Reference is always live checkd with 'is_ref_valid'.
+
+        """
+        # Trigger validate by getting 'is_valid'
+        _ = self.is_ref_valid
+
+    @property
+    def enabled(self) -> bool:
+        """Is callback enabled.
+
+        Returns:
+            bool: Is callback enabled.
+
+        """
+        return self._enabled
+
+    def set_enabled(self, enabled: bool) -> None:
+        """Change if callback is enabled.
+
+        Args:
+            enabled (bool): Change enabled state of the callback.
+
+        """
+        self._enabled = enabled
+
+    def deregister(self) -> None:
+        """Calling this function will cause that callback will be removed."""
+        self._ref_is_valid = False
+        self._partial_func = None
+        self._func_ref = None
+
+    def get_order(self) -> int:
+        """Get callback order.
+
+        Returns:
+            int: Callback order.
+
+        """
+        return self._order
+
+    def set_order(self, order: int) -> None:
+        """Change callback order.
+
+        Args:
+            order (int): Order of callback. Lower number means
+                higher priority.
+
+        """
+        self._validate_order(order)
+        self._order = order
+
+    order = property(get_order, set_order)
+
+    def topic_matches(self, topic: str) -> bool:
+        """Check if event topic matches callback's topic.
+
+        Args:
+            topic (str): Topic name.
+
+        Returns:
+            bool: Topic matches callback's topic.
+
+        """
+        return self._topic_regex.match(topic)
+
+    def process_event(self, event: Event) -> None:
+        """Process event.
+
+        Args:
+            event(Event): Event that was triggered.
+
+        """
+        # Skip if callback is not enabled
+        if not self._enabled:
+            return
+
+        # Get reference and skip if is not available
+        callback = self._get_callback()
+        if callback is None:
+            return
+
+        if not self.topic_matches(event.topic):
+            return
+
+        # Try to execute callback
+        try:
+            if self._expect_args:
+                callback(event)
+
+            elif self._expect_kwargs:
+                callback(event=event)
+
+            else:
+                callback()
+
+        except Exception:
+            self.log.warning(
+                f"Failed to execute event callback {repr(self)}",
+                exc_info=True
+            )
+
+    def _validate_order(self, order: int) -> None:
+        if isinstance(order, int):
+            return
+
+        raise TypeError(f"Expected type 'int' got '{type(order)}'.")
+
+    def _get_callback(self) -> Callable | None:
+        if self._partial_func is not None:
+            return self._partial_func
+
+        if self._func_ref is not None:
+            return self._func_ref()
+        return None
+
+    def _validate_ref(self) -> None:
+        if self._ref_is_valid is False:
+            return
+
+        if self._func_ref is not None:
+            self._ref_is_valid = self._func_ref() is not None
+
+        elif self._partial_func is not None:
+            self._ref_is_valid = self._partial_func.is_valid()
+
+        else:
+            self._ref_is_valid = False
+
+        if not self._ref_is_valid:
+            self._func_ref = None
+            self._partial_func = None
+
+
+class EventSystem:
+    """Encapsulate event handling into an object.
+
+    System wraps registered callbacks and triggered events into single object,
+    so it is possible to create multiple independent systems that have their
+    topics and callbacks.
+
+    Callbacks are stored by order of their registration, but it is possible to
+    manually define order of callbacks using 'order' argument within
+    'add_callback'.
+    """
+
+    default_order: int = 100
+
+    def __init__(self) -> None:
+        self._registered_callbacks: list[EventCallback] = []
+
+    def add_callback(
         self,
-        topic,
-        sender=None,
-        event_hash=None,
-        project_name=None,
-        username=None,
-        dependencies=None,
-        description=None,
-        summary=None,
-        payload=None,
-        finished=True,
-        store=True,
-    ):
-        if dependencies is None:
-            dependencies = []
-        if payload is None:
-            payload = {}
-        if summary is None:
-            summary = {}
-
-        self.topic = topic
-        self.sender = sender
-        self.event_hash = event_hash
-        self.project_name = project_name
-        self.username = username
-        self.dependencies = dependencies
-        self.description = description
-        self.summary = summary
-        self.payload = payload
-        self.finished = finished
-        self.store = store
-
-    def to_data(self):
-        return {
-            "topic": self.topic,
-            "sender": self.sender,
-            "hash": self.event_hash,
-            "project": self.project_name,
-            "user": self.username,
-            "dependencies": copy.deepcopy(self.dependencies),
-            "description": self.description,
-            "summary": copy.deepcopy(self.summary),
-            "payload": self.payload,
-            "finished": self.finished,
-            "store": self.store
-        }
+        topic: str,
+        callback: Callable | weakref_partial,
+        order: int | None = None,
+    ) -> EventCallback:
+        """Register callback in event system.
+
+        Args:
+            topic (str): Topic for EventCallback.
+            callback (Callable | weakref_partial): Function or method
+                that will be called when topic is triggered.
+            order (int | None): Order of callback. Lower number means
+                higher priority.
+
+        Returns:
+            EventCallback: Created callback object which can be used to
+                stop listening.
+
+        """
+        if order is None:
+            order = self.default_order
+
+        callback = EventCallback(topic, callback, order)
+        self._registered_callbacks.append(callback)
+        return callback
+
+    def emit_event(self, event: Event) -> None:
+        """Emit event object.
+
+        Args:
+            event (Event): Prepared event with topic and data.
+
+        """
+        self._process_event(event)
+
+    def clear_callbacks(self) -> None:
+        """Clear all registered callbacks."""
+        self._registered_callbacks = []
+
+    def _process_event(self, event: Event) -> None:
+        """Process event topic and trigger callbacks.
+
+        Args:
+            event (Event): Prepared event with topic and data.
+
+        """
+        callbacks = tuple(sorted(
+            self._registered_callbacks, key=lambda x: x.order
+        ))
+        for callback in callbacks:
+            callback.process_event(event)
+            if not callback.is_ref_valid:
+                self._registered_callbacks.remove(callback)

From d0487ce2902e1ef4077c57c722a87013de18f8da Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Fri, 21 Aug 2026 12:47:08 +0200
Subject: [PATCH 02/13] add websocket client to requirements

---
 pyproject.toml | 2 ++
 setup.py       | 1 +
 2 files changed, 3 insertions(+)

diff --git a/pyproject.toml b/pyproject.toml
index a4cea5722..e60ace302 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,6 +16,7 @@ classifiers = [
 dependencies = [
     "requests >= 2.27.1",
     "Unidecode >= 1.3.0",
+    "websocket-client >= 1.7.0",
 ]
 
 [project.urls]
@@ -42,6 +43,7 @@ packages = [
 python = ">=3.6.5"
 requests = "^2.27"
 Unidecode = "^1.3"
+websocket-client = "^1.7"
 
 [tool.poetry.group.dev.dependencies]
 sphinx = "*"
diff --git a/setup.py b/setup.py
index 402a48546..c85921c26 100644
--- a/setup.py
+++ b/setup.py
@@ -31,6 +31,7 @@
     install_requires=[
         "requests >= 2.27.1",
         "Unidecode >= 1.3.0",
+        "websocket-client >= 1.7.0",
         "appdirs >=1, <2",
     ],
     keywords=["AYON", "ynput", "OpenPype", "vfx"],

From ca2bf45c5e849ea9befc15ca9548b83c03bd87b8 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Fri, 21 Aug 2026 17:46:01 +0200
Subject: [PATCH 03/13] initial implementation of websockets event hub

---
 ayon_api/events.py     | 245 ++++++++++++++++++++++++++++++++++++-----
 ayon_api/server_api.py |  63 +++++++++++
 2 files changed, 280 insertions(+), 28 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 2447f9d49..54d3fceb8 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -1,31 +1,66 @@
 from __future__ import annotations
 
+import atexit
 from dataclasses import dataclass
 import inspect
+import json
 import logging
 import os
 import re
+import threading
+import typing
 from typing import Any, Callable
 import weakref
 
+from ._api import get_server_api_connection
+
+if typing.TYPE_CHECKING:
+    from websocket import WebSocket
+
+    from .server_api import ServerAPI
+
 
 @dataclass
 class Event:
     topic: str
+    data: dict[str, Any]
+    id: str | None = None
+    sender: str | None = None
+    event_hash: str | None = None
+    project_name: str | None = None
+    dependencies: list[str] | None = None
+    description: str | None = None
+    summary: str | None = None
+    payload: dict[str, Any] | None = None
+    status: str | None = None
+    store: bool | None = None
+
+    def __getitem__(self, key) -> Any:
+        return self.data[key]
+
+    def get(self, key, default=None) -> Any:
+        return self.data.get(key, default)
+
+    @classmethod
+    def from_ws_message(cls, data: dict[str, Any]) -> Event | None:
+        topic = data.get("topic")
+        if topic is None:
+            return None
 
-
-@dataclass
-class ServerEvent(Event):
-    sender: str | None
-    event_hash: str | None
-    project_name: str | None
-    username: str | None
-    dependencies: list[str] | None
-    description: str | None
-    summary: str | None
-    payload: dict[str, Any] | None
-    finished: bool = True
-    store: bool = True
+        return cls(
+            topic=topic,
+            data=data,
+            id=data.get("id"),
+            sender=data.get("sender"),
+            event_hash=data.get("eventHash"),
+            project_name=data.get("project"),
+            dependencies=data.get("dependsOn"),
+            description=data.get("description"),
+            summary=data.get("summary"),
+            payload=data.get("payload"),
+            status=data.get("status"),
+            store=data.get("store"),
+        )
 
 
 def is_func_signature_supported(func, *args, **kwargs):
@@ -195,12 +230,12 @@ class EventCallback:
     may contain '*' that will be handled as "any characters".
 
     # Examples:
-    - "workfile.save"   Callback will be triggered if the event topic is
-                        exactly "workfile.save" .
-    - "workfile.*"      Callback will be triggered an event topic starts with
-                        "workfile." so "workfile.save" and "workfile.open"
-                        will trigger the callback.
-    - "*"               Callback will listen to all events.
+    - "entity.folder.attr_changed" - Callback will be triggered if the event
+        topic is exactly "entity.folder.attr_changed".
+    - "entity.*" - Callback will be triggered an event topic starts with
+        "entity." so "entity.folder.created" and "entity.version.created"
+        will trigger the callback.
+    - "*" Callback will listen to all events.
 
     Callback can be function or method. In both cases it should expect one
     or none arguments. When 1 argument is expected then the processed 'Event'
@@ -218,19 +253,24 @@ class EventCallback:
     Args:
         topic (str): Topic which will be listened.
         func (Callable): Callback to a topic.
-        order (Union[int, None]): Order of callback. Lower number means higher
+        order (int | None): Order of callback. Lower number means higher
             priority.
 
     Raises:
         TypeError: When passed function is not a callable object.
     """
+    default_order: int = 100
 
-    def __init__(self, topic: str, func: Callable, order: int) -> None:
+    def __init__(
+        self, topic: str, func: Callable, order: int | None = None
+    ) -> None:
         if not callable(func):
             raise TypeError(
                 f"Registered callback is not callable. \"{func}\""
             )
 
+        if order is None:
+            order = self.default_order
         self._validate_order(order)
 
         self._log = None
@@ -290,6 +330,10 @@ def log(self) -> logging.Logger:
             self._log = logging.getLogger(self.__class__.__name__)
         return self._log
 
+    @property
+    def topic(self) -> str:
+        return self._topic
+
     @property
     def is_ref_valid(self) -> bool:
         """
@@ -438,7 +482,13 @@ def _validate_ref(self) -> None:
             self._partial_func = None
 
 
-class EventSystem:
+@dataclass
+class _LoopState:
+    running: bool = False
+    stop_event: threading.Event = threading.Event()
+
+
+class EventHub:
     """Encapsulate event handling into an object.
 
     System wraps registered callbacks and triggered events into single object,
@@ -450,10 +500,18 @@ class EventSystem:
     'add_callback'.
     """
 
-    default_order: int = 100
+    def __init__(self, connection: ServerAPI | None = None) -> None:
+        if connection is None:
+            connection = get_server_api_connection()
 
-    def __init__(self) -> None:
+        self._connection: ServerAPI = connection
+        self._ws_connection: WebSocket | None = None
+        self._loop_state: _LoopState = _LoopState()
+        self._loop_thread: threading.Thread | None = None
+        self._registered_topics: set[str] = set()
         self._registered_callbacks: list[EventCallback] = []
+        self._internal_callbacks: list[EventCallback] = []
+        atexit.register(self.stop)
 
     def add_callback(
         self,
@@ -475,13 +533,21 @@ def add_callback(
                 stop listening.
 
         """
-        if order is None:
-            order = self.default_order
-
         callback = EventCallback(topic, callback, order)
-        self._registered_callbacks.append(callback)
+        self.add_callbacks([callback])
         return callback
 
+    def add_callbacks(self, callbacks: list[EventCallback]) -> None:
+        """Register callback in event system.
+
+        Args:
+            callbacks (list[EventCallback]): List of EventCallback
+                objects to register.
+
+        """
+        self._registered_callbacks.extend(callbacks)
+        self._update_topics()
+
     def emit_event(self, event: Event) -> None:
         """Emit event object.
 
@@ -495,6 +561,124 @@ def clear_callbacks(self) -> None:
         """Clear all registered callbacks."""
         self._registered_callbacks = []
 
+    def stop(self) -> None:
+        if self._loop_thread is None:
+            return
+
+        if self._ws_connection is not None:
+            self._ws_connection.close()
+        self._loop_state.stop_event.set()
+        self._loop_thread.join()
+
+    def _update_topics(self) -> None:
+        """Subscribe to topics in server based on registered callbacks.
+
+        Server does not allow wildcards in the topic but does validate
+            start of the topic so 'entity.folder.*' is not allowed
+            but 'entity.folder.' does work.
+
+        In case the wildcard is used at the start of the topic we have to
+            subscribe to all topics.
+
+        """
+        topics = set()
+        for callback in self._registered_callbacks:
+            topic = callback.topic
+            if topic == "*":
+                topics.add(topic)
+                continue
+            parts = topic.split("*", maxsplit=1)
+            if len(parts) == 1:
+                topics.add(topic)
+                continue
+
+            part = parts[0]
+            if part:
+                topics.add(part)
+            else:
+                topics.add("*")
+
+        if topics == self._registered_topics:
+            return
+
+        self._registered_topics = topics
+
+        ws_connection = self._get_ws_connection()
+        subscribe_payload = {
+            "topic": "auth",
+            "token": self._connection.get_token(),
+            "subscribe": list(self._registered_topics),
+        }
+        ws_connection.send(json.dumps(subscribe_payload))
+
+    def _get_ws_connection(self) -> WebSocket:
+        if self._ws_connection is None:
+            return self._create_ws_connection()
+        # TODO validate if is still alive and re-create if needed
+        if not self._ws_connection.connected:
+            return self._create_ws_connection()
+        return self._ws_connection
+
+    def _create_ws_connection(self) -> WebSocket:
+        self.stop()
+        self._loop_state.stop_event = threading.Event()
+
+        ws_connection = self._connection.create_websocket()
+        self._ws_connection = ws_connection
+
+        callbacks = [
+            EventCallback(
+                "server.started", self._on_server_started,
+            ),
+            EventCallback(
+                "server.restart_requested", self._on_server_restart,
+            ),
+        ]
+        old_c, self._internal_callbacks = self._internal_callbacks, callbacks
+        for callback in old_c:
+            callback.deregister()
+
+        self.add_callbacks(callbacks)
+
+        loop_thread = threading.Thread(target=self._thread_loop)
+        self._loop_thread = loop_thread
+        loop_thread.start()
+
+        return ws_connection
+
+    def _thread_loop(self) -> None:
+        self._loop_state.running = True
+        try:
+            con = self._get_ws_connection()
+            while con.connected:
+                if self._loop_state.stop_event.is_set():
+                    break
+
+                message = con.recv()
+                if not message:
+                    continue
+
+                event_data = json.loads(message)
+                event = Event.from_ws_message(event_data)
+                if event is not None:
+                    self.emit_event(event)
+
+        finally:
+            self._loop_state.running = True
+            self._loop_state.stop_event.set()
+
+    def _on_server_started(self, event: Event) -> None:
+        """Handle server started event.
+
+        Args:
+            event (Event): Event object with topic and data.
+
+        """
+        print("HeHe?")
+
+    def _on_server_restart(self, event: Event) -> None:
+        print("Server is restarting")
+
     def _process_event(self, event: Event) -> None:
         """Process event topic and trigger callbacks.
 
@@ -505,7 +689,12 @@ def _process_event(self, event: Event) -> None:
         callbacks = tuple(sorted(
             self._registered_callbacks, key=lambda x: x.order
         ))
+        any_removed = False
         for callback in callbacks:
             callback.process_event(event)
             if not callback.is_ref_valid:
+                any_removed = True
                 self._registered_callbacks.remove(callback)
+
+        if any_removed:
+            self._update_topics()
diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index 9981554c9..137fdaa9f 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -11,6 +11,7 @@
 import re
 import io
 import json
+import ssl
 import time
 import logging
 import platform
@@ -22,6 +23,7 @@
 )
 
 import requests
+import websocket
 
 from .constants import (
     SERVER_RETRIES_ENV_KEY,
@@ -94,6 +96,8 @@
 )
 
 if typing.TYPE_CHECKING:
+    from websocket import WebSocket
+
     from .typing import (
         ServerVersion,
         AnyEntityDict,
@@ -380,9 +384,68 @@ def get_base_url(self) -> str:
     def get_rest_url(self) -> str:
         return self._rest_url
 
+    def get_websocket_url(
+        self, endpoint: str = "ws"
+    ) -> str:
+        """Prepare websocket url from endpoint and optional params."""
+        endpoint = (endpoint or "").strip()
+
+        if self._base_url.startswith("https"):
+            base_url = f"wss{self._base_url[5:]}"
+        elif self._base_url.startswith("http"):
+            base_url = f"ws{self._base_url[4:]}"
+        else:
+            raise ValueError(f"Invalid scheme in base URL: {self._base_url}")
+
+        base_url = base_url.rstrip("/")
+        endpoint = endpoint.lstrip("/")
+        return f"{base_url}/{endpoint}"
+
     base_url = property(get_base_url)
     rest_url = property(get_rest_url)
 
+    def create_websocket(
+        self,
+        endpoint: str = "ws",
+        timeout: float | None = None,
+        headers: dict[str, Any] | None = None,
+        sslopt: dict[str, Any] | None = None,
+        **kwargs,
+    ) -> WebSocket:
+        """Create websocket connection to AYON server."""
+        ws_url = self.get_websocket_url(endpoint)
+        ws_headers = self.get_headers()
+        ws_headers.pop("Content-Type", None)
+        if headers:
+            ws_headers.update(headers)
+
+        ws_kwargs = copy.deepcopy(kwargs)
+        if timeout is None:
+            timeout = self.timeout
+        if timeout:
+            ws_kwargs["timeout"] = timeout
+
+        if ws_headers:
+            ws_kwargs["header"] = [
+                f"{key}: {value}"
+                for key, value in ws_headers.items()
+                if value is not None
+            ]
+
+        prepared_sslopt = copy.deepcopy(sslopt) if sslopt else {}
+        if self._ssl_verify is False:
+            prepared_sslopt.setdefault("cert_reqs", ssl.CERT_NONE)
+        elif isinstance(self._ssl_verify, str):
+            prepared_sslopt.setdefault("ca_certs", self._ssl_verify)
+
+        if self._cert:
+            prepared_sslopt.setdefault("certfile", self._cert)
+
+        if ws_url.startswith("wss://") and prepared_sslopt:
+            ws_kwargs["sslopt"] = prepared_sslopt
+
+        return websocket.create_connection(ws_url, **ws_kwargs)
+
     def get_ssl_verify(self) -> bool | str | None:
         """Enable ssl verification.
 

From c18100badd1f0a24a7e67a551b2738fb4fbd5e62 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Mon, 24 Aug 2026 10:41:53 +0200
Subject: [PATCH 04/13] add websocket to public api

---
 automated_api.py       |  9 ++++-
 ayon_api/__init__.py   |  4 ++
 ayon_api/_api.py       | 71 +++++++++++++++++++++------------
 ayon_api/events.py     |  2 +-
 ayon_api/server_api.py | 89 +++++++++++++++++++++---------------------
 5 files changed, 102 insertions(+), 73 deletions(-)

diff --git a/automated_api.py b/automated_api.py
index b2536606f..c41eee8b8 100644
--- a/automated_api.py
+++ b/automated_api.py
@@ -24,8 +24,13 @@
 requests.__dict__["Response"] = type(
     "Response", (), {"__module__": "requests"}
 )
+websocket = type(sys)("websocket")
+websocket.__dict__["WebSocket"] = type(
+    "WebSocket", (), {"__module__": "websocket"}
+)
 
 sys.modules["requests"] = requests
+sys.modules["websocket"] = websocket
 sys.modules["unidecode"] = type(sys)("unidecode")
 
 CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -113,8 +118,8 @@ def prepare_docstring(func):
 
     line_char = ""
     if "\n" in docstring:
-        line_char = "\n"
-    return f'"""{docstring}{line_char}\n"""'
+        line_char = "\n\n"
+    return f'"""{docstring}{line_char}"""'
 
 
 def _find_obj(obj_full, api_globals):
diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py
index aef55d7e4..5772863b0 100644
--- a/ayon_api/__init__.py
+++ b/ayon_api/__init__.py
@@ -34,6 +34,7 @@
     as_username,
     get_base_url,
     get_rest_url,
+    get_websocket_url,
     get_ssl_verify,
     set_ssl_verify,
     get_cert,
@@ -52,6 +53,7 @@
     set_sender,
     get_sender_type,
     set_sender_type,
+    create_websocket,
     get_info,
     get_server_version,
     get_server_version_tuple,
@@ -354,6 +356,7 @@
     "as_username",
     "get_base_url",
     "get_rest_url",
+    "get_websocket_url",
     "get_ssl_verify",
     "set_ssl_verify",
     "get_cert",
@@ -372,6 +375,7 @@
     "set_sender",
     "get_sender_type",
     "set_sender_type",
+    "create_websocket",
     "get_info",
     "get_server_version",
     "get_server_version_tuple",
diff --git a/ayon_api/_api.py b/ayon_api/_api.py
index 58de1cbb4..c4dff37bb 100644
--- a/ayon_api/_api.py
+++ b/ayon_api/_api.py
@@ -43,6 +43,9 @@
 
 if typing.TYPE_CHECKING:
     from typing import Union, Literal
+
+    from websocket import WebSocket
+
     from .typing import (
         ServerVersion,
         ActivityType,
@@ -453,6 +456,16 @@ def get_rest_url() -> str:
     return con.get_rest_url()
 
 
+def get_websocket_url(
+    endpoint: str,
+) -> str:
+    """Prepare websocket url from endpoint."""
+    con = get_server_api_connection()
+    return con.get_websocket_url(
+        endpoint=endpoint,
+    )
+
+
 def get_ssl_verify() -> bool | str | None:
     """Enable ssl verification.
 
@@ -705,6 +718,25 @@ def set_sender_type(
     )
 
 
+def create_websocket(
+    endpoint: str,
+    *,
+    timeout: float | None = None,
+    headers: dict[str, Any] | None = None,
+    sslopt: dict[str, Any] | None = None,
+    **kwargs,
+) -> WebSocket:
+    """Create a websocket connection to AYON server."""
+    con = get_server_api_connection()
+    return con.create_websocket(
+        endpoint=endpoint,
+        timeout=timeout,
+        headers=headers,
+        sslopt=sslopt,
+        **kwargs,
+    )
+
+
 def get_info() -> dict[str, Any]:
     """Get information about current used api key.
 
@@ -752,15 +784,13 @@ def get_server_version_tuple() -> ServerVersion:
 
 
 def is_product_base_type_supported() -> bool:
-    """Product base types are available on server.
-    """
+    """Product base types are available on server."""
     con = get_server_api_connection()
     return con.is_product_base_type_supported()
 
 
 def links_graphql_support_data() -> bool:
-    """Links data can be received by GraphQl.
-    """
+    """Links data can be received by GraphQl."""
     con = get_server_api_connection()
     return con.links_graphql_support_data()
 
@@ -1369,8 +1399,7 @@ def delete_project_file(
     project_name: str,
     file_id: str,
 ) -> None:
-    """Delete project file.
-    """
+    """Delete project file."""
     con = get_server_api_connection()
     return con.delete_project_file(
         project_name=project_name,
@@ -2617,8 +2646,7 @@ def create_activity_reaction(
     activity_id: str,
     reaction: str,
 ) -> None:
-    """React to activity.
-    """
+    """React to activity."""
     con = get_server_api_connection()
     return con.create_activity_reaction(
         project_name=project_name,
@@ -2672,8 +2700,7 @@ def get_raw_entity_watchers(
     entity_id: str,
     entity_type: str,
 ) -> dict[str, Any]:
-    """Get entity watchers (raw response).
-    """
+    """Get entity watchers (raw response)."""
     con = get_server_api_connection()
     return con.get_raw_entity_watchers(
         project_name=project_name,
@@ -2687,8 +2714,7 @@ def get_entity_watchers(
     entity_id: str,
     entity_type: str,
 ) -> list[str]:
-    """List watchers of an entity.
-    """
+    """List watchers of an entity."""
     con = get_server_api_connection()
     return con.get_entity_watchers(
         project_name=project_name,
@@ -2703,8 +2729,7 @@ def set_entity_watchers(
     entity_type: str,
     watchers: list[str],
 ):
-    """Change watchers of an entity.
-    """
+    """Change watchers of an entity."""
     con = get_server_api_connection()
     return con.set_entity_watchers(
         project_name=project_name,
@@ -4433,8 +4458,7 @@ def delete_project(
 
 
 def get_raw_project_folders() -> dict[str, Any]:
-    """Get project folders (raw data).
-    """
+    """Get project folders (raw data)."""
     con = get_server_api_connection()
     return con.get_raw_project_folders()
 
@@ -4449,8 +4473,7 @@ def create_project_folder(
     parent_id: str | None = None,
     data: dict[str, Any] | None = None,
 ) -> str:
-    """Create project folder.
-    """
+    """Create project folder."""
     con = get_server_api_connection()
     return con.create_project_folder(
         label=label,
@@ -4477,8 +4500,7 @@ def update_project_folder(
 def set_project_folders_order(
     folder_ids: list[str],
 ) -> None:
-    """Set project folders order.
-    """
+    """Set project folders order."""
     con = get_server_api_connection()
     return con.set_project_folders_order(
         folder_ids=folder_ids,
@@ -4489,8 +4511,7 @@ def assign_projects_to_project_folder(
     folder_id: str,
     project_names: list[str],
 ) -> None:
-    """Assign project folder to project.
-    """
+    """Assign project folder to project."""
     con = get_server_api_connection()
     return con.assign_projects_to_project_folder(
         folder_id=folder_id,
@@ -4501,8 +4522,7 @@ def assign_projects_to_project_folder(
 def delete_project_folder(
     folder_id: str,
 ):
-    """Delete project folder.
-    """
+    """Delete project folder."""
     con = get_server_api_connection()
     return con.delete_project_folder(
         folder_id=folder_id,
@@ -8491,8 +8511,7 @@ def delete_entity_list_folder(
     project_name: str,
     entity_list_folder_id: str,
 ) -> None:
-    """Delete entity list folder.
-    """
+    """Delete entity list folder."""
     con = get_server_api_connection()
     return con.delete_entity_list_folder(
         project_name=project_name,
diff --git a/ayon_api/events.py b/ayon_api/events.py
index 54d3fceb8..44c40fdb7 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -623,7 +623,7 @@ def _create_ws_connection(self) -> WebSocket:
         self.stop()
         self._loop_state.stop_event = threading.Event()
 
-        ws_connection = self._connection.create_websocket()
+        ws_connection = self._connection.create_websocket("ws")
         self._ws_connection = ws_connection
 
         callbacks = [
diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index 137fdaa9f..8db50881e 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -385,9 +385,9 @@ def get_rest_url(self) -> str:
         return self._rest_url
 
     def get_websocket_url(
-        self, endpoint: str = "ws"
+        self, endpoint: str
     ) -> str:
-        """Prepare websocket url from endpoint and optional params."""
+        """Prepare websocket url from endpoint."""
         endpoint = (endpoint or "").strip()
 
         if self._base_url.startswith("https"):
@@ -404,48 +404,6 @@ def get_websocket_url(
     base_url = property(get_base_url)
     rest_url = property(get_rest_url)
 
-    def create_websocket(
-        self,
-        endpoint: str = "ws",
-        timeout: float | None = None,
-        headers: dict[str, Any] | None = None,
-        sslopt: dict[str, Any] | None = None,
-        **kwargs,
-    ) -> WebSocket:
-        """Create websocket connection to AYON server."""
-        ws_url = self.get_websocket_url(endpoint)
-        ws_headers = self.get_headers()
-        ws_headers.pop("Content-Type", None)
-        if headers:
-            ws_headers.update(headers)
-
-        ws_kwargs = copy.deepcopy(kwargs)
-        if timeout is None:
-            timeout = self.timeout
-        if timeout:
-            ws_kwargs["timeout"] = timeout
-
-        if ws_headers:
-            ws_kwargs["header"] = [
-                f"{key}: {value}"
-                for key, value in ws_headers.items()
-                if value is not None
-            ]
-
-        prepared_sslopt = copy.deepcopy(sslopt) if sslopt else {}
-        if self._ssl_verify is False:
-            prepared_sslopt.setdefault("cert_reqs", ssl.CERT_NONE)
-        elif isinstance(self._ssl_verify, str):
-            prepared_sslopt.setdefault("ca_certs", self._ssl_verify)
-
-        if self._cert:
-            prepared_sslopt.setdefault("certfile", self._cert)
-
-        if ws_url.startswith("wss://") and prepared_sslopt:
-            ws_kwargs["sslopt"] = prepared_sslopt
-
-        return websocket.create_connection(ws_url, **ws_kwargs)
-
     def get_ssl_verify(self) -> bool | str | None:
         """Enable ssl verification.
 
@@ -928,6 +886,49 @@ def close_session(self) -> None:
         self._session_functions_mapping = {}
         session.close()
 
+    def create_websocket(
+        self,
+        endpoint: str,
+        *,
+        timeout: float | None = None,
+        headers: dict[str, Any] | None = None,
+        sslopt: dict[str, Any] | None = None,
+        **kwargs,
+    ) -> WebSocket:
+        """Create a websocket connection to AYON server."""
+        ws_url = self.get_websocket_url(endpoint)
+        ws_headers = self.get_headers()
+        ws_headers.pop("Content-Type", None)
+        if headers:
+            ws_headers.update(headers)
+
+        ws_kwargs = copy.deepcopy(kwargs)
+        if timeout is None:
+            timeout = self.timeout
+        if timeout:
+            ws_kwargs["timeout"] = timeout
+
+        if ws_headers:
+            ws_kwargs["header"] = [
+                f"{key}: {value}"
+                for key, value in ws_headers.items()
+                if value is not None
+            ]
+
+        prepared_sslopt = copy.deepcopy(sslopt) if sslopt else {}
+        if self._ssl_verify is False:
+            prepared_sslopt.setdefault("cert_reqs", ssl.CERT_NONE)
+        elif isinstance(self._ssl_verify, str):
+            prepared_sslopt.setdefault("ca_certs", self._ssl_verify)
+
+        if self._cert:
+            prepared_sslopt.setdefault("certfile", self._cert)
+
+        if ws_url.startswith("wss://") and prepared_sslopt:
+            ws_kwargs["sslopt"] = prepared_sslopt
+
+        return websocket.create_connection(ws_url, **ws_kwargs)
+
     def _update_session_headers(self) -> None:
         if self._session is None:
             return

From 1e577b076113e85444e0c3f3718a72d87d995f13 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Wed, 26 Aug 2026 16:58:51 +0200
Subject: [PATCH 05/13] rework reconnecting

---
 ayon_api/events.py     | 199 ++++++++++++++++++++++++++++++-----------
 ayon_api/server_api.py |   1 -
 2 files changed, 146 insertions(+), 54 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 44c40fdb7..3d5ffb27e 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -1,18 +1,25 @@
 from __future__ import annotations
 
 import atexit
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 import inspect
 import json
 import logging
 import os
 import re
+import socket
+import time
 import threading
 import typing
 from typing import Any, Callable
 import weakref
 
-from ._api import get_server_api_connection
+from websocket import (
+    ABNF,
+    WebSocketProtocolException,
+    WebSocketConnectionClosedException,
+    WebSocketTimeoutException,
+)
 
 if typing.TYPE_CHECKING:
     from websocket import WebSocket
@@ -484,8 +491,12 @@ def _validate_ref(self) -> None:
 
 @dataclass
 class _LoopState:
+    started: bool = False
     running: bool = False
     stop_event: threading.Event = threading.Event()
+    auth_required: bool = False
+    registered_topics: set[str] = field(default_factory=set)
+    server_is_restarting: bool = False
 
 
 class EventHub:
@@ -502,22 +513,25 @@ class EventHub:
 
     def __init__(self, connection: ServerAPI | None = None) -> None:
         if connection is None:
+            from ._api import get_server_api_connection
+
             connection = get_server_api_connection()
 
         self._connection: ServerAPI = connection
         self._ws_connection: WebSocket | None = None
         self._loop_state: _LoopState = _LoopState()
         self._loop_thread: threading.Thread | None = None
-        self._registered_topics: set[str] = set()
         self._registered_callbacks: list[EventCallback] = []
         self._internal_callbacks: list[EventCallback] = []
-        atexit.register(self.stop)
+        atexit.register(self._stop)
 
     def add_callback(
         self,
         topic: str,
         callback: Callable | weakref_partial,
         order: int | None = None,
+        *,
+        create_connection: bool = True,
     ) -> EventCallback:
         """Register callback in event system.
 
@@ -527,6 +541,8 @@ def add_callback(
                 that will be called when topic is triggered.
             order (int | None): Order of callback. Lower number means
                 higher priority.
+            create_connection (bool): Create websocket connection if
+                not already created.
 
         Returns:
             EventCallback: Created callback object which can be used to
@@ -534,19 +550,30 @@ def add_callback(
 
         """
         callback = EventCallback(topic, callback, order)
-        self.add_callbacks([callback])
+        self.add_callbacks(
+            [callback], create_connection=create_connection
+        )
         return callback
 
-    def add_callbacks(self, callbacks: list[EventCallback]) -> None:
+    def add_callbacks(
+        self,
+        callbacks: list[EventCallback],
+        *,
+        create_connection: bool = True,
+    ) -> None:
         """Register callback in event system.
 
         Args:
             callbacks (list[EventCallback]): List of EventCallback
                 objects to register.
+            create_connection (bool): Create websocket connection if
+                not already created.
 
         """
         self._registered_callbacks.extend(callbacks)
         self._update_topics()
+        if create_connection:
+            self.start()
 
     def emit_event(self, event: Event) -> None:
         """Emit event object.
@@ -557,18 +584,32 @@ def emit_event(self, event: Event) -> None:
         """
         self._process_event(event)
 
-    def clear_callbacks(self) -> None:
-        """Clear all registered callbacks."""
-        self._registered_callbacks = []
+    def start(self) -> None:
+        """Start event loop.
 
-    def stop(self) -> None:
-        if self._loop_thread is None:
+        This will create websocket connection and start listening to events.
+        """
+        if self._loop_state.started:
             return
 
+        self._loop_state.started = True
+        self._create_connection_thread()
+
+    def stop(self) -> None:
+        self._stop()
+
+    def _stop(self) -> None:
+        loop_thread, self._loop_thread = self._loop_thread, None
+        self._loop_state.stop_event.set()
+
+        if self._loop_state.started:
+            self._loop_state.started = False
+
         if self._ws_connection is not None:
             self._ws_connection.close()
-        self._loop_state.stop_event.set()
-        self._loop_thread.join()
+
+        if loop_thread is not None:
+            loop_thread.join()
 
     def _update_topics(self) -> None:
         """Subscribe to topics in server based on registered callbacks.
@@ -598,38 +639,28 @@ def _update_topics(self) -> None:
             else:
                 topics.add("*")
 
-        if topics == self._registered_topics:
+        if topics == self._loop_state.registered_topics:
             return
 
-        self._registered_topics = topics
+        self._loop_state.registered_topics = topics
+        self._loop_state.auth_required = True
 
-        ws_connection = self._get_ws_connection()
-        subscribe_payload = {
-            "topic": "auth",
-            "token": self._connection.get_token(),
-            "subscribe": list(self._registered_topics),
-        }
-        ws_connection.send(json.dumps(subscribe_payload))
+    def _create_ws_connection(self) -> WebSocket:
+        if self._ws_connection is not None:
+            if self._ws_connection.connected:
+                return self._ws_connection
+            self._ws_connection = None
 
-    def _get_ws_connection(self) -> WebSocket:
-        if self._ws_connection is None:
-            return self._create_ws_connection()
-        # TODO validate if is still alive and re-create if needed
-        if not self._ws_connection.connected:
-            return self._create_ws_connection()
-        return self._ws_connection
+        con = self._connection.create_websocket("ws")
+        self._ws_connection = con
+        return con
 
-    def _create_ws_connection(self) -> WebSocket:
-        self.stop()
+    def _create_connection_thread(self) -> None:
+        self._stop()
+        self._loop_state.started = True
         self._loop_state.stop_event = threading.Event()
 
-        ws_connection = self._connection.create_websocket("ws")
-        self._ws_connection = ws_connection
-
         callbacks = [
-            EventCallback(
-                "server.started", self._on_server_started,
-            ),
             EventCallback(
                 "server.restart_requested", self._on_server_restart,
             ),
@@ -644,17 +675,87 @@ def _create_ws_connection(self) -> WebSocket:
         self._loop_thread = loop_thread
         loop_thread.start()
 
-        return ws_connection
-
     def _thread_loop(self) -> None:
         self._loop_state.running = True
+        con: WebSocket | None
+        new_connection: bool = True
         try:
-            con = self._get_ws_connection()
-            while con.connected:
+            while True:
+                con = self._ws_connection
                 if self._loop_state.stop_event.is_set():
+                    if con is not None and con.connected:
+                        con.close()
+                    self._ws_connection = None
                     break
 
-                message = con.recv()
+                if con is None:
+                    try:
+                        con = self._create_ws_connection()
+                    except (
+                        WebSocketTimeoutException,
+                        WebSocketConnectionClosedException,
+                        socket.error,
+                    ):
+                        time.sleep(0.5)
+                        continue
+                    self._loop_state.server_is_restarting = False
+                    self._loop_state.auth_required = True
+                    new_connection = True
+
+                if self._loop_state.auth_required:
+                    token = self._connection.get_token()
+                    subscribe_payload = {
+                        "topic": "auth",
+                        "token": token,
+                        "subscribe": list(
+                            self._loop_state.registered_topics
+                        ),
+                    }
+                    try:
+                        con.send(json.dumps(subscribe_payload))
+                    except WebSocketConnectionClosedException:
+                        self._ws_connection = None
+                        if not new_connection:
+                            self.emit_event(Event(
+                                topic="connection.closed", data={}
+                            ))
+                        continue
+
+                    self._loop_state.auth_required = False
+                    continue
+
+                try:
+                    op_code, message = con.recv_data()
+                except (
+                    WebSocketProtocolException,
+                    WebSocketConnectionClosedException,
+                ):
+                    self._ws_connection = None
+                    if not new_connection:
+                        self.emit_event(
+                            Event(topic="connection.closed", data={})
+                        )
+                    continue
+
+                if op_code == ABNF.OPCODE_CLOSE:
+                    self._ws_connection = None
+                    # NOTE if is new connection then token is probably invalid
+                    # - question is what to do in that case?
+                    if not new_connection:
+                        self.emit_event(
+                            Event(topic="connection.closed", data={})
+                        )
+                    continue
+
+                if new_connection:
+                    new_connection = False
+                    self.emit_event(
+                        Event(topic="connection.opened", data={})
+                    )
+
+                if op_code != ABNF.OPCODE_TEXT:
+                    continue
+
                 if not message:
                     continue
 
@@ -664,20 +765,12 @@ def _thread_loop(self) -> None:
                     self.emit_event(event)
 
         finally:
-            self._loop_state.running = True
+            self._loop_state.server_is_restarting = False
+            self._loop_state.running = False
             self._loop_state.stop_event.set()
 
-    def _on_server_started(self, event: Event) -> None:
-        """Handle server started event.
-
-        Args:
-            event (Event): Event object with topic and data.
-
-        """
-        print("HeHe?")
-
     def _on_server_restart(self, event: Event) -> None:
-        print("Server is restarting")
+        self._loop_state.server_is_restarting = True
 
     def _process_event(self, event: Event) -> None:
         """Process event topic and trigger callbacks.
diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index 8db50881e..a151d3f63 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -2740,7 +2740,6 @@ def _validate_operations_result(
         if result.get("success"):
             return None
 
-        print(result)
         for op_result in result["operations"]:
             if op_result["success"]:
                 continue

From 4f1abf7627bd71226c45cdd576dd631f849007a9 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Wed, 26 Aug 2026 17:14:55 +0200
Subject: [PATCH 06/13] add event hub to ServerAPI

---
 ayon_api/__init__.py   |  2 ++
 ayon_api/_api.py       | 14 ++++++++++++++
 ayon_api/events.py     | 34 ++++++++++++++++++++++++++++------
 ayon_api/server_api.py | 14 ++++++++++++++
 4 files changed, 58 insertions(+), 6 deletions(-)

diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py
index 5772863b0..7d700ce99 100644
--- a/ayon_api/__init__.py
+++ b/ayon_api/__init__.py
@@ -62,6 +62,7 @@
     get_users,
     get_user_by_name,
     get_user,
+    get_event_hub,
     raw_post,
     raw_put,
     raw_patch,
@@ -384,6 +385,7 @@
     "get_users",
     "get_user_by_name",
     "get_user",
+    "get_event_hub",
     "raw_post",
     "raw_put",
     "raw_patch",
diff --git a/ayon_api/_api.py b/ayon_api/_api.py
index c4dff37bb..186084365 100644
--- a/ayon_api/_api.py
+++ b/ayon_api/_api.py
@@ -93,6 +93,7 @@
         EntityListAttributeDefinitionDict,
         AdvancedFilterDict,
     )
+    from .events import EventHub
 
 
 class GlobalServerAPI(ServerAPI):
@@ -874,6 +875,19 @@ def get_user(
     )
 
 
+def get_event_hub() -> EventHub:
+    """Get event hub for connection.
+
+    Event hub is used to subscribe to events from server.
+
+    Returns:
+        EventHub: Event hub instance.
+
+    """
+    con = get_server_api_connection()
+    return con.get_event_hub()
+
+
 def raw_post(
     entrypoint: str,
     **kwargs,
diff --git a/ayon_api/events.py b/ayon_api/events.py
index 3d5ffb27e..4c26c9bfb 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -525,13 +525,35 @@ def __init__(self, connection: ServerAPI | None = None) -> None:
         self._internal_callbacks: list[EventCallback] = []
         atexit.register(self._stop)
 
+    def is_running(self) -> bool:
+        """Check if event loop is running.
+
+        Returns:
+            bool: Is event loop running.
+
+        """
+        return self._loop_state.running
+
+    def is_connected(self) -> bool:
+        """Check if event loop is connected to server.
+
+        Returns:
+            bool: Is event loop connected to server.
+
+        """
+        con = self._ws_connection
+        if con is None:
+            return False
+
+        return con.connected
+
     def add_callback(
         self,
         topic: str,
         callback: Callable | weakref_partial,
         order: int | None = None,
         *,
-        create_connection: bool = True,
+        connect: bool = True,
     ) -> EventCallback:
         """Register callback in event system.
 
@@ -541,7 +563,7 @@ def add_callback(
                 that will be called when topic is triggered.
             order (int | None): Order of callback. Lower number means
                 higher priority.
-            create_connection (bool): Create websocket connection if
+            connect (bool): Create websocket connection if
                 not already created.
 
         Returns:
@@ -551,7 +573,7 @@ def add_callback(
         """
         callback = EventCallback(topic, callback, order)
         self.add_callbacks(
-            [callback], create_connection=create_connection
+            [callback], connect=connect
         )
         return callback
 
@@ -559,20 +581,20 @@ def add_callbacks(
         self,
         callbacks: list[EventCallback],
         *,
-        create_connection: bool = True,
+        connect: bool = True,
     ) -> None:
         """Register callback in event system.
 
         Args:
             callbacks (list[EventCallback]): List of EventCallback
                 objects to register.
-            create_connection (bool): Create websocket connection if
+            connect (bool): Create websocket connection if
                 not already created.
 
         """
         self._registered_callbacks.extend(callbacks)
         self._update_topics()
-        if create_connection:
+        if connect:
             self.start()
 
     def emit_event(self, event: Event) -> None:
diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index a151d3f63..e51c4c52c 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -48,6 +48,7 @@
 )
 from .graphql import INTROSPECTION_QUERY
 from .graphql_queries import users_graphql_query
+from .events import EventHub
 from .exceptions import (
     FailedOperations,
     UnauthorizedError,
@@ -374,6 +375,8 @@ def __init__(
             self.validate_server_availability()
             self.create_session()
 
+        self._event_hub: EventHub = EventHub(self)
+
     @property
     def log(self) -> logging.Logger:
         return self._log
@@ -1221,6 +1224,17 @@ def get_headers(
                 headers["Authorization"] = f"Bearer {self._token_info.token}"
         return headers
 
+    def get_event_hub(self) -> EventHub:
+        """Get event hub for connection.
+
+        Event hub is used to subscribe to events from server.
+
+        Returns:
+            EventHub: Event hub instance.
+
+        """
+        return self._event_hub
+
     def login(
         self,
         username: str,

From 1c0b4090a42ebd22e02290c3846dffcda285894f Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Wed, 26 Aug 2026 17:15:03 +0200
Subject: [PATCH 07/13] fix automated api

---
 automated_api.py | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/automated_api.py b/automated_api.py
index c41eee8b8..c5289fe4b 100644
--- a/automated_api.py
+++ b/automated_api.py
@@ -25,9 +25,16 @@
     "Response", (), {"__module__": "requests"}
 )
 websocket = type(sys)("websocket")
-websocket.__dict__["WebSocket"] = type(
-    "WebSocket", (), {"__module__": "websocket"}
-)
+websocket.__dict__.update({
+    name: type(name, (), {"__module__": "websocket"})
+    for name in (
+        "WebSocket",
+        "ABNF",
+        "WebSocketProtocolException",
+        "WebSocketConnectionClosedException",
+        "WebSocketTimeoutException",
+    )
+})
 
 sys.modules["requests"] = requests
 sys.modules["websocket"] = websocket

From 05066aae3eca6278412d6ec7cecdcfae85451f0e Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Thu, 27 Aug 2026 12:37:41 +0200
Subject: [PATCH 08/13] unify on clase logic and validate token

---
 ayon_api/events.py | 59 ++++++++++++++++++++++++----------------------
 1 file changed, 31 insertions(+), 28 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 4c26c9bfb..30a03c937 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -30,7 +30,7 @@
 @dataclass
 class Event:
     topic: str
-    data: dict[str, Any]
+    data: dict[str, Any] = field(default_factory=dict)
     id: str | None = None
     sender: str | None = None
     event_hash: str | None = None
@@ -495,6 +495,7 @@ class _LoopState:
     running: bool = False
     stop_event: threading.Event = threading.Event()
     auth_required: bool = False
+    new_connection: bool = True
     registered_topics: set[str] = field(default_factory=set)
     server_is_restarting: bool = False
 
@@ -699,18 +700,37 @@ def _create_connection_thread(self) -> None:
 
     def _thread_loop(self) -> None:
         self._loop_state.running = True
+
+        def _on_close(handle_token: bool = True):
+            self._ws_connection = None
+            if not self._loop_state.new_connection:
+                self.emit_event(Event("connection.closed"))
+
+            if not handle_token:
+                return
+
+            if self._connection.has_valid_token:
+                self._connection.validate_token()
+            if not self._connection.has_valid_token:
+                self.emit_event(Event("auth.failed"))
+
         con: WebSocket | None
-        new_connection: bool = True
         try:
             while True:
                 con = self._ws_connection
                 if self._loop_state.stop_event.is_set():
                     if con is not None and con.connected:
                         con.close()
-                    self._ws_connection = None
+                    _on_close(handle_token=False)
                     break
 
                 if con is None:
+                    # NOTE: Don't even try to connect if token is invalid,
+                    #   wait for change
+                    if not self._connection.has_valid_token:
+                        time.sleep(0.5)
+                        continue
+
                     try:
                         con = self._create_ws_connection()
                     except (
@@ -722,13 +742,12 @@ def _thread_loop(self) -> None:
                         continue
                     self._loop_state.server_is_restarting = False
                     self._loop_state.auth_required = True
-                    new_connection = True
+                    self._loop_state.new_connection = True
 
                 if self._loop_state.auth_required:
-                    token = self._connection.get_token()
                     subscribe_payload = {
                         "topic": "auth",
-                        "token": token,
+                        "token": self._connection.access_token,
                         "subscribe": list(
                             self._loop_state.registered_topics
                         ),
@@ -736,11 +755,7 @@ def _thread_loop(self) -> None:
                     try:
                         con.send(json.dumps(subscribe_payload))
                     except WebSocketConnectionClosedException:
-                        self._ws_connection = None
-                        if not new_connection:
-                            self.emit_event(Event(
-                                topic="connection.closed", data={}
-                            ))
+                        _on_close()
                         continue
 
                     self._loop_state.auth_required = False
@@ -752,28 +767,16 @@ def _thread_loop(self) -> None:
                     WebSocketProtocolException,
                     WebSocketConnectionClosedException,
                 ):
-                    self._ws_connection = None
-                    if not new_connection:
-                        self.emit_event(
-                            Event(topic="connection.closed", data={})
-                        )
+                    _on_close()
                     continue
 
                 if op_code == ABNF.OPCODE_CLOSE:
-                    self._ws_connection = None
-                    # NOTE if is new connection then token is probably invalid
-                    # - question is what to do in that case?
-                    if not new_connection:
-                        self.emit_event(
-                            Event(topic="connection.closed", data={})
-                        )
+                    _on_close()
                     continue
 
-                if new_connection:
-                    new_connection = False
-                    self.emit_event(
-                        Event(topic="connection.opened", data={})
-                    )
+                if self._loop_state.new_connection:
+                    self._loop_state.new_connection = False
+                    self.emit_event(Event("connection.opened"))
 
                 if op_code != ABNF.OPCODE_TEXT:
                     continue

From 4c40ef73041d99139634aa183c4cd7842791b0a3 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Mon, 14 Sep 2026 10:52:15 +0200
Subject: [PATCH 09/13] capture more exceptions

---
 ayon_api/events.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 30a03c937..7febb6825 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -766,6 +766,7 @@ def _on_close(handle_token: bool = True):
                 except (
                     WebSocketProtocolException,
                     WebSocketConnectionClosedException,
+                    WebSocketTimeoutException,
                 ):
                     _on_close()
                     continue

From 316cd7c8f4d1dfc870ccafb024c18045c6ec2362 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Mon, 12 Oct 2026 15:14:44 +0200
Subject: [PATCH 10/13] fix the even hub

---
 ayon_api/events.py | 492 +++++++++++++++++++++++++++++++--------------
 1 file changed, 340 insertions(+), 152 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index 7febb6825..b44fab949 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -7,19 +7,15 @@
 import logging
 import os
 import re
-import socket
-import time
 import threading
 import typing
 from typing import Any, Callable
 import weakref
 
-from websocket import (
-    ABNF,
-    WebSocketProtocolException,
-    WebSocketConnectionClosedException,
-    WebSocketTimeoutException,
-)
+import requests
+from websocket import ABNF
+
+from .exceptions import UnauthorizedError
 
 if typing.TYPE_CHECKING:
     from websocket import WebSocket
@@ -489,27 +485,75 @@ def _validate_ref(self) -> None:
             self._partial_func = None
 
 
+# Server sends 'heartbeat' message to authorized clients when there was no
+#   other message for ~5 seconds. If nothing arrives for this amount of time
+#   the connection is considered dead.
+_RECV_TIMEOUT: float = 15.0
+# Timeout for opening websocket connection and for the auth check request.
+_CONNECT_TIMEOUT: float = 5.0
+# Delays between reconnection attempts (last value is used repeatedly).
+_RECONNECT_DELAYS: tuple[float, ...] = (0.5, 1.0, 2.0, 5.0)
+# Delay between reconnection attempts when the token is known to be invalid.
+_AUTH_FAILED_DELAY: float = 30.0
+# How long to wait for the loop thread on stop. The thread is daemon so
+#   it can't block process exit even if it does not finish in time.
+_STOP_TIMEOUT: float = 1.0
+
+
+def _abort_ws(ws_con: WebSocket) -> None:
+    """Close websocket without waiting for server response.
+
+    'WebSocket.close' sends close frame and waits for the response
+        (up to 3 seconds) which would block the caller.
+
+    """
+    for method_name in ("abort", "shutdown"):
+        method = getattr(ws_con, method_name, None)
+        if method is None:
+            continue
+        try:
+            method()
+            return
+        except Exception:
+            pass
+
+
 @dataclass
 class _LoopState:
-    started: bool = False
-    running: bool = False
-    stop_event: threading.Event = threading.Event()
+    stop_event: threading.Event = field(default_factory=threading.Event)
+    # Subscribe message has to be sent
     auth_required: bool = False
-    new_connection: bool = True
+    # Some message was received with current websocket connection
+    received_message: bool = False
+    # Connection state reported via events
+    #   - 'None' means unknown (nothing was reported yet)
+    connected: bool | None = None
+    auth_failed: bool = False
+    failed_attempts: int = 0
     registered_topics: set[str] = field(default_factory=set)
     server_is_restarting: bool = False
 
 
 class EventHub:
-    """Encapsulate event handling into an object.
+    """Receive AYON server events via websocket and trigger callbacks.
 
-    System wraps registered callbacks and triggered events into single object,
-    so it is possible to create multiple independent systems that have their
-    topics and callbacks.
+    Websocket connection is handled in a daemon thread, so it never blocks
+        the caller (e.g. DCC main thread) nor the process exit.
+
+    Warnings:
+        Callbacks are triggered in the websocket thread! Use
+            'WSEventsModel' (or own queue) to process them in main thread,
+            especially if a callback touches Qt widgets.
+
+    Internal topics:
+        - 'connection.opened' first message received after (re)connection.
+        - 'connection.closed' connection was lost or could not be created.
+        - 'auth.failed' server rejected the token.
+
+    Args:
+        connection (ServerAPI | None): Server connection. Global connection
+            is used if not passed.
 
-    Callbacks are stored by order of their registration, but it is possible to
-    manually define order of callbacks using 'order' argument within
-    'add_callback'.
     """
 
     def __init__(self, connection: ServerAPI | None = None) -> None:
@@ -518,12 +562,21 @@ def __init__(self, connection: ServerAPI | None = None) -> None:
 
             connection = get_server_api_connection()
 
+        self._log = logging.getLogger(self.__class__.__name__)
         self._connection: ServerAPI = connection
         self._ws_connection: WebSocket | None = None
         self._loop_state: _LoopState = _LoopState()
         self._loop_thread: threading.Thread | None = None
+        self._thread_lock = threading.Lock()
+        self._callbacks_lock = threading.RLock()
+        self._state_lock = threading.Lock()
         self._registered_callbacks: list[EventCallback] = []
-        self._internal_callbacks: list[EventCallback] = []
+        self._internal_callbacks: list[EventCallback] = [
+            EventCallback(
+                "server.restart_requested", self._on_server_restart,
+            ),
+        ]
+        self.add_callbacks(self._internal_callbacks, connect=False)
         atexit.register(self._stop)
 
     def is_running(self) -> bool:
@@ -533,7 +586,8 @@ def is_running(self) -> bool:
             bool: Is event loop running.
 
         """
-        return self._loop_state.running
+        thread = self._loop_thread
+        return thread is not None and thread.is_alive()
 
     def is_connected(self) -> bool:
         """Check if event loop is connected to server.
@@ -542,11 +596,62 @@ def is_connected(self) -> bool:
             bool: Is event loop connected to server.
 
         """
-        con = self._ws_connection
-        if con is None:
-            return False
+        return self._loop_state.connected is True
 
-        return con.connected
+    def get_connection_state(self) -> bool | None:
+        """Connection state as was reported by events.
+
+        Returns:
+            bool | None: 'True' if connected, 'False' if disconnected and
+                'None' if state is not known yet.
+
+        """
+        return self._loop_state.connected
+
+    def is_auth_failed(self) -> bool:
+        return self._loop_state.auth_failed
+
+    def is_server_restarting(self) -> bool:
+        return self._loop_state.server_is_restarting
+
+    def check_server_available(self) -> bool | None:
+        """Quick synchronous check if server is available.
+
+        Websocket connection is created in a background thread, so it takes
+            a while until the connection state is known. This check can be
+            used e.g. when a tool is opened to find out that the server is
+            not available before the tool tries to fetch data from it.
+
+        Request is sent only if the connection state is not known yet. When
+            the server is not available, the state is changed to
+            disconnected and 'connection.closed' event is emitted.
+
+        Returns:
+            bool | None: 'False' if server is not available, 'True' if
+                websocket is connected, 'None' if server did respond but
+                websocket connection is not established yet.
+
+        """
+        state = self._loop_state.connected
+        if state is not None:
+            return state
+
+        # Use 'get_info' so timeout and retries configured on the
+        #   connection are respected
+        try:
+            self._connection.get_info()
+            available = True
+        except UnauthorizedError:
+            # Server is running but rejected the token
+            available = True
+        except Exception:
+            available = False
+
+        if available:
+            return self._loop_state.connected
+
+        self._set_connected(False)
+        return False
 
     def add_callback(
         self,
@@ -593,8 +698,9 @@ def add_callbacks(
                 not already created.
 
         """
-        self._registered_callbacks.extend(callbacks)
-        self._update_topics()
+        with self._callbacks_lock:
+            self._registered_callbacks.extend(callbacks)
+            self._update_topics()
         if connect:
             self.start()
 
@@ -610,29 +716,46 @@ def emit_event(self, event: Event) -> None:
     def start(self) -> None:
         """Start event loop.
 
-        This will create websocket connection and start listening to events.
+        Websocket connection is created in a daemon thread, this method
+            does not block.
+
         """
-        if self._loop_state.started:
-            return
+        with self._thread_lock:
+            if self.is_running():
+                return
 
-        self._loop_state.started = True
-        self._create_connection_thread()
+            stop_event = threading.Event()
+            self._loop_state.stop_event = stop_event
+            loop_thread = threading.Thread(
+                target=self._thread_loop,
+                args=(stop_event,),
+                name="AYONEventHub",
+                # Daemon thread must be used! Non-daemon threads are joined
+                #   by the interpreter BEFORE 'atexit' callbacks are called,
+                #   so the process would hang on exit.
+                daemon=True,
+            )
+            self._loop_thread = loop_thread
+            loop_thread.start()
 
     def stop(self) -> None:
         self._stop()
 
     def _stop(self) -> None:
-        loop_thread, self._loop_thread = self._loop_thread, None
-        self._loop_state.stop_event.set()
-
-        if self._loop_state.started:
-            self._loop_state.started = False
+        with self._thread_lock:
+            loop_thread, self._loop_thread = self._loop_thread, None
+            self._loop_state.stop_event.set()
+            ws_con = self._ws_connection
 
-        if self._ws_connection is not None:
-            self._ws_connection.close()
+        # Unblock 'recv' in the loop thread
+        if ws_con is not None:
+            _abort_ws(ws_con)
 
-        if loop_thread is not None:
-            loop_thread.join()
+        if (
+            loop_thread is not None
+            and loop_thread is not threading.current_thread()
+        ):
+            loop_thread.join(_STOP_TIMEOUT)
 
     def _update_topics(self) -> None:
         """Subscribe to topics in server based on registered callbacks.
@@ -646,7 +769,10 @@ def _update_topics(self) -> None:
 
         """
         topics = set()
-        for callback in self._registered_callbacks:
+        with self._callbacks_lock:
+            callbacks = list(self._registered_callbacks)
+
+        for callback in callbacks:
             topic = callback.topic
             if topic == "*":
                 topics.add(topic)
@@ -669,133 +795,186 @@ def _update_topics(self) -> None:
         self._loop_state.auth_required = True
 
     def _create_ws_connection(self) -> WebSocket:
-        if self._ws_connection is not None:
-            if self._ws_connection.connected:
-                return self._ws_connection
-            self._ws_connection = None
+        ws_con = self._connection.create_websocket("ws")
+        # Server sends heartbeat periodically, so timeout on receive means
+        #   that the connection is dead.
+        ws_con.settimeout(_RECV_TIMEOUT)
+        return ws_con
 
-        con = self._connection.create_websocket("ws")
-        self._ws_connection = con
-        return con
+    def _close_ws_connection(self) -> None:
+        with self._thread_lock:
+            ws_con, self._ws_connection = self._ws_connection, None
+        if ws_con is not None:
+            _abort_ws(ws_con)
 
-    def _create_connection_thread(self) -> None:
-        self._stop()
-        self._loop_state.started = True
-        self._loop_state.stop_event = threading.Event()
+    def _check_token(self) -> bool | None:
+        """Check if token is valid without affecting the server connection.
 
-        callbacks = [
-            EventCallback(
-                "server.restart_requested", self._on_server_restart,
-            ),
-        ]
-        old_c, self._internal_callbacks = self._internal_callbacks, callbacks
-        for callback in old_c:
-            callback.deregister()
+        Do NOT use 'ServerAPI.validate_token' here. It marks the token
+            as invalid on any non-200 response (e.g. 502 while server is
+            restarting) which would break all requests of the process.
 
-        self.add_callbacks(callbacks)
-
-        loop_thread = threading.Thread(target=self._thread_loop)
-        self._loop_thread = loop_thread
-        loop_thread.start()
+        Returns:
+            bool | None: 'True' if valid, 'False' if server rejected the
+                token, 'None' if it is not possible to tell (server is
+                not reachable).
 
-    def _thread_loop(self) -> None:
-        self._loop_state.running = True
+        """
+        token = self._connection.access_token
+        if not token:
+            return False
 
-        def _on_close(handle_token: bool = True):
-            self._ws_connection = None
-            if not self._loop_state.new_connection:
-                self.emit_event(Event("connection.closed"))
+        ssl_verify = self._connection.get_ssl_verify()
+        if ssl_verify is None:
+            ssl_verify = True
+        url = f"{self._connection.get_base_url()}/api/users/me"
+        status_codes = set()
+        for headers in (
+            {"Authorization": f"Bearer {token}"},
+            {"X-Api-Key": token},
+        ):
+            try:
+                response = requests.get(
+                    url,
+                    headers=headers,
+                    verify=ssl_verify,
+                    cert=self._connection.get_cert(),
+                    timeout=_CONNECT_TIMEOUT,
+                )
+            except Exception:
+                return None
+            if response.status_code == 200:
+                return True
+            status_codes.add(response.status_code)
+
+        if status_codes <= {401, 403}:
+            return False
+        return None
 
-            if not handle_token:
+    def _set_connected(self, connected: bool) -> None:
+        # Can be called from websocket thread and from main thread
+        #   ('check_server_available').
+        state = self._loop_state
+        with self._state_lock:
+            if state.connected is connected:
                 return
+            state.connected = connected
+            if connected:
+                state.failed_attempts = 0
+                state.server_is_restarting = False
+                state.auth_failed = False
+
+        if connected:
+            self.emit_event(Event("connection.opened"))
+        else:
+            self.emit_event(Event("connection.closed"))
+
+    def _on_connection_lost(self, stop_event: threading.Event) -> None:
+        state = self._loop_state
+        received_message = state.received_message
+        state.received_message = False
+        self._close_ws_connection()
+        if stop_event.is_set():
+            return
 
-            if self._connection.has_valid_token:
-                self._connection.validate_token()
-            if not self._connection.has_valid_token:
+        self._set_connected(False)
+
+        # Server closes unauthorized clients after few seconds without
+        #   sending anything, authorized clients receive heartbeat.
+        if not received_message and not state.auth_failed:
+            if self._check_token() is False:
+                state.auth_failed = True
                 self.emit_event(Event("auth.failed"))
 
-        con: WebSocket | None
+        if state.auth_failed:
+            delay = _AUTH_FAILED_DELAY
+        else:
+            idx = min(state.failed_attempts, len(_RECONNECT_DELAYS) - 1)
+            delay = _RECONNECT_DELAYS[idx]
+        state.failed_attempts += 1
+        stop_event.wait(delay)
+
+    def _thread_loop(self, stop_event: threading.Event) -> None:
         try:
-            while True:
-                con = self._ws_connection
-                if self._loop_state.stop_event.is_set():
-                    if con is not None and con.connected:
-                        con.close()
-                    _on_close(handle_token=False)
-                    break
-
-                if con is None:
-                    # NOTE: Don't even try to connect if token is invalid,
-                    #   wait for change
-                    if not self._connection.has_valid_token:
-                        time.sleep(0.5)
-                        continue
-
-                    try:
-                        con = self._create_ws_connection()
-                    except (
-                        WebSocketTimeoutException,
-                        WebSocketConnectionClosedException,
-                        socket.error,
-                    ):
-                        time.sleep(0.5)
-                        continue
-                    self._loop_state.server_is_restarting = False
-                    self._loop_state.auth_required = True
-                    self._loop_state.new_connection = True
-
-                if self._loop_state.auth_required:
-                    subscribe_payload = {
-                        "topic": "auth",
-                        "token": self._connection.access_token,
-                        "subscribe": list(
-                            self._loop_state.registered_topics
-                        ),
-                    }
-                    try:
-                        con.send(json.dumps(subscribe_payload))
-                    except WebSocketConnectionClosedException:
-                        _on_close()
-                        continue
-
-                    self._loop_state.auth_required = False
+            self._loop(stop_event)
+        except Exception:
+            self._log.warning("AYON event hub loop crashed", exc_info=True)
+        finally:
+            self._close_ws_connection()
+            self._loop_state.received_message = False
+
+    def _loop(self, stop_event: threading.Event) -> None:
+        state = self._loop_state
+        while not stop_event.is_set():
+            ws_con = self._ws_connection
+            if ws_con is None:
+                if not self._connection.access_token:
+                    stop_event.wait(1.0)
                     continue
 
                 try:
-                    op_code, message = con.recv_data()
-                except (
-                    WebSocketProtocolException,
-                    WebSocketConnectionClosedException,
-                    WebSocketTimeoutException,
-                ):
-                    _on_close()
+                    ws_con = self._create_ws_connection()
+                except Exception:
+                    # Server is not available (connection refused, timeout,
+                    #   502 from proxy during restart, ssl error...)
+                    self._on_connection_lost(stop_event)
                     continue
 
-                if op_code == ABNF.OPCODE_CLOSE:
-                    _on_close()
-                    continue
+                with self._thread_lock:
+                    if stop_event.is_set():
+                        _abort_ws(ws_con)
+                        break
+                    self._ws_connection = ws_con
+                state.received_message = False
+                state.auth_required = True
+
+            if state.auth_required:
+                state.auth_required = False
+                subscribe_payload = {
+                    "topic": "auth",
+                    "token": self._connection.access_token,
+                    "subscribe": list(state.registered_topics),
+                }
+                try:
+                    ws_con.send(json.dumps(subscribe_payload))
+                except Exception:
+                    self._on_connection_lost(stop_event)
+                continue
 
-                if self._loop_state.new_connection:
-                    self._loop_state.new_connection = False
-                    self.emit_event(Event("connection.opened"))
+            try:
+                op_code, message = ws_con.recv_data()
+            except Exception:
+                # Timeout (no heartbeat), closed connection, socket error...
+                self._on_connection_lost(stop_event)
+                continue
 
-                if op_code != ABNF.OPCODE_TEXT:
-                    continue
+            if op_code == ABNF.OPCODE_CLOSE:
+                self._on_connection_lost(stop_event)
+                continue
 
-                if not message:
-                    continue
+            if not state.received_message:
+                state.received_message = True
+                self._set_connected(True)
 
+            if op_code != ABNF.OPCODE_TEXT or not message:
+                continue
+
+            try:
                 event_data = json.loads(message)
-                event = Event.from_ws_message(event_data)
-                if event is not None:
-                    self.emit_event(event)
+            except ValueError:
+                self._log.debug(
+                    "Failed to parse websocket message", exc_info=True
+                )
+                continue
 
-        finally:
-            self._loop_state.server_is_restarting = False
-            self._loop_state.running = False
-            self._loop_state.stop_event.set()
+            if not isinstance(event_data, dict):
+                continue
 
-    def _on_server_restart(self, event: Event) -> None:
+            event = Event.from_ws_message(event_data)
+            if event is not None:
+                self.emit_event(event)
+
+    def _on_server_restart(self) -> None:
         self._loop_state.server_is_restarting = True
 
     def _process_event(self, event: Event) -> None:
@@ -805,15 +984,24 @@ def _process_event(self, event: Event) -> None:
             event (Event): Prepared event with topic and data.
 
         """
-        callbacks = tuple(sorted(
-            self._registered_callbacks, key=lambda x: x.order
-        ))
-        any_removed = False
+        with self._callbacks_lock:
+            callbacks = tuple(sorted(
+                self._registered_callbacks, key=lambda x: x.order
+            ))
+
+        removed = []
         for callback in callbacks:
             callback.process_event(event)
             if not callback.is_ref_valid:
-                any_removed = True
-                self._registered_callbacks.remove(callback)
+                removed.append(callback)
 
-        if any_removed:
+        if not removed:
+            return
+
+        with self._callbacks_lock:
+            for callback in removed:
+                try:
+                    self._registered_callbacks.remove(callback)
+                except ValueError:
+                    pass
             self._update_topics()

From 8aa7c9ab63b3206157b681137f99e5fe466ff329 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Mon, 12 Oct 2026 15:53:27 +0200
Subject: [PATCH 11/13] use constants for event hub topics

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 ayon_api/events.py | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/ayon_api/events.py b/ayon_api/events.py
index b44fab949..3addfb162 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -555,6 +555,12 @@ class EventHub:
             is used if not passed.
 
     """
+    # Topics emitted by the hub itself
+    TOPIC_CONNECTION_OPENED: str = "connection.opened"
+    TOPIC_CONNECTION_CLOSED: str = "connection.closed"
+    TOPIC_AUTH_FAILED: str = "auth.failed"
+    # Topic sent by server
+    TOPIC_SERVER_RESTART_REQUESTED: str = "server.restart_requested"
 
     def __init__(self, connection: ServerAPI | None = None) -> None:
         if connection is None:
@@ -573,7 +579,8 @@ def __init__(self, connection: ServerAPI | None = None) -> None:
         self._registered_callbacks: list[EventCallback] = []
         self._internal_callbacks: list[EventCallback] = [
             EventCallback(
-                "server.restart_requested", self._on_server_restart,
+                self.TOPIC_SERVER_RESTART_REQUESTED,
+                self._on_server_restart,
             ),
         ]
         self.add_callbacks(self._internal_callbacks, connect=False)
@@ -865,9 +872,9 @@ def _set_connected(self, connected: bool) -> None:
                 state.auth_failed = False
 
         if connected:
-            self.emit_event(Event("connection.opened"))
+            self.emit_event(Event(self.TOPIC_CONNECTION_OPENED))
         else:
-            self.emit_event(Event("connection.closed"))
+            self.emit_event(Event(self.TOPIC_CONNECTION_CLOSED))
 
     def _on_connection_lost(self, stop_event: threading.Event) -> None:
         state = self._loop_state
@@ -884,7 +891,7 @@ def _on_connection_lost(self, stop_event: threading.Event) -> None:
         if not received_message and not state.auth_failed:
             if self._check_token() is False:
                 state.auth_failed = True
-                self.emit_event(Event("auth.failed"))
+                self.emit_event(Event(self.TOPIC_AUTH_FAILED))
 
         if state.auth_failed:
             delay = _AUTH_FAILED_DELAY

From 8ca19dff0c011be1bae2ee825c5b81d3092bbfdb Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:39:43 +0200
Subject: [PATCH 12/13] keep the print in the code

---
 ayon_api/server_api.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index e51c4c52c..77a96f033 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -2754,6 +2754,7 @@ def _validate_operations_result(
         if result.get("success"):
             return None
 
+        print(result)
         for op_result in result["operations"]:
             if op_result["success"]:
                 continue

From 830cfa0c5916f3e31116e738eecd3c4ab797a4c7 Mon Sep 17 00:00:00 2001
From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com>
Date: Wed, 14 Oct 2026 17:04:11 +0200
Subject: [PATCH 13/13] Event hub: lazy creation and token changes

- Token change ('set_token', 'reset_token', 'login', 'logout')
  re-authenticates the websocket and wakes the hub from auth failure
  backoff.
- 'ServerAPI' creates its hub on demand.
- 'EventHub' requires connection.
- One 'atexit' callback stops all hubs, hubs are stored weakly so stopped
  hubs and their connections can be garbage collected.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 ayon_api/_api.py                  |   8 +-
 ayon_api/events.py                |  79 +++++++++++---
 ayon_api/server_api.py            |  82 +++++++++-----
 tests/test_event_hub_lifecycle.py | 176 ++++++++++++++++++++++++++++++
 4 files changed, 300 insertions(+), 45 deletions(-)
 create mode 100644 tests/test_event_hub_lifecycle.py

diff --git a/ayon_api/_api.py b/ayon_api/_api.py
index dde262703..444f085af 100644
--- a/ayon_api/_api.py
+++ b/ayon_api/_api.py
@@ -878,7 +878,13 @@ def get_user(
 def get_event_hub() -> EventHub:
     """Get event hub for connection.
 
-    Event hub is used to subscribe to events from server.
+    Event hub is used to subscribe to events from server. The hub is
+        created on first call, websocket connection is created when
+        a callback is registered or when the hub is started.
+
+    Notes:
+        Running hub keeps reference to the connection, call
+            'EventHub.stop' when the connection is not used anymore.
 
     Returns:
         EventHub: Event hub instance.
diff --git a/ayon_api/events.py b/ayon_api/events.py
index 3addfb162..1dda11969 100644
--- a/ayon_api/events.py
+++ b/ayon_api/events.py
@@ -532,6 +532,21 @@ class _LoopState:
     failed_attempts: int = 0
     registered_topics: set[str] = field(default_factory=set)
     server_is_restarting: bool = False
+    # Interrupts waiting between reconnection attempts
+    wake_event: threading.Event = field(default_factory=threading.Event)
+
+
+# Hubs stopped on process exit, stored weakly so a hub that is not running
+#   can be garbage collected
+_EVENT_HUBS: weakref.WeakSet[EventHub] = weakref.WeakSet()
+
+
+def _stop_event_hubs() -> None:
+    for event_hub in list(_EVENT_HUBS):
+        event_hub.stop()
+
+
+atexit.register(_stop_event_hubs)
 
 
 class EventHub:
@@ -541,9 +556,9 @@ class EventHub:
         the caller (e.g. DCC main thread) nor the process exit.
 
     Warnings:
-        Callbacks are triggered in the websocket thread! Use
-            'WSEventsModel' (or own queue) to process them in main thread,
-            especially if a callback touches Qt widgets.
+        Callbacks are triggered in the websocket thread! Use own queue
+            to process them in main thread, especially if a callback
+            touches Qt widgets.
 
     Internal topics:
         - 'connection.opened' first message received after (re)connection.
@@ -551,8 +566,7 @@ class EventHub:
         - 'auth.failed' server rejected the token.
 
     Args:
-        connection (ServerAPI | None): Server connection. Global connection
-            is used if not passed.
+        connection (ServerAPI): Server connection.
 
     """
     # Topics emitted by the hub itself
@@ -562,12 +576,7 @@ class EventHub:
     # Topic sent by server
     TOPIC_SERVER_RESTART_REQUESTED: str = "server.restart_requested"
 
-    def __init__(self, connection: ServerAPI | None = None) -> None:
-        if connection is None:
-            from ._api import get_server_api_connection
-
-            connection = get_server_api_connection()
-
+    def __init__(self, connection: ServerAPI) -> None:
         self._log = logging.getLogger(self.__class__.__name__)
         self._connection: ServerAPI = connection
         self._ws_connection: WebSocket | None = None
@@ -584,7 +593,7 @@ def __init__(self, connection: ServerAPI | None = None) -> None:
             ),
         ]
         self.add_callbacks(self._internal_callbacks, connect=False)
-        atexit.register(self._stop)
+        _EVENT_HUBS.add(self)
 
     def is_running(self) -> bool:
         """Check if event loop is running.
@@ -748,10 +757,43 @@ def start(self) -> None:
     def stop(self) -> None:
         self._stop()
 
+    def reconnect(self) -> None:
+        """Drop current websocket connection and connect again.
+
+        Should be called when websocket connection should be recreated,
+            e.g. server url of the connection did change.
+
+        """
+        state = self._loop_state
+        state.auth_failed = False
+        state.failed_attempts = 0
+        self._close_ws_connection()
+        state.wake_event.set()
+
+    def reauthenticate(self) -> None:
+        """Authenticate websocket connection with current token.
+
+        Should be called when token of the server connection did change.
+            Websocket connection is kept if is open, otherwise next
+            connection attempt happens immediately.
+
+        """
+        if not self._connection.access_token:
+            # Token was removed, server would close the connection anyway
+            self.reconnect()
+            return
+
+        state = self._loop_state
+        state.auth_failed = False
+        state.failed_attempts = 0
+        state.auth_required = True
+        state.wake_event.set()
+
     def _stop(self) -> None:
         with self._thread_lock:
             loop_thread, self._loop_thread = self._loop_thread, None
             self._loop_state.stop_event.set()
+            self._loop_state.wake_event.set()
             ws_con = self._ws_connection
 
         # Unblock 'recv' in the loop thread
@@ -870,6 +912,7 @@ def _set_connected(self, connected: bool) -> None:
                 state.failed_attempts = 0
                 state.server_is_restarting = False
                 state.auth_failed = False
+                state.wake_event.clear()
 
         if connected:
             self.emit_event(Event(self.TOPIC_CONNECTION_OPENED))
@@ -899,7 +942,15 @@ def _on_connection_lost(self, stop_event: threading.Event) -> None:
             idx = min(state.failed_attempts, len(_RECONNECT_DELAYS) - 1)
             delay = _RECONNECT_DELAYS[idx]
         state.failed_attempts += 1
-        stop_event.wait(delay)
+        self._wait(stop_event, delay)
+
+    def _wait(self, stop_event: threading.Event, timeout: float) -> None:
+        """Wait until timeout, stop or wake request."""
+        if stop_event.is_set():
+            return
+        wake_event = self._loop_state.wake_event
+        wake_event.wait(timeout)
+        wake_event.clear()
 
     def _thread_loop(self, stop_event: threading.Event) -> None:
         try:
@@ -916,7 +967,7 @@ def _loop(self, stop_event: threading.Event) -> None:
             ws_con = self._ws_connection
             if ws_con is None:
                 if not self._connection.access_token:
-                    stop_event.wait(1.0)
+                    self._wait(stop_event, 1.0)
                     continue
 
                 try:
diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py
index f2130fecf..7cb03c4d4 100644
--- a/ayon_api/server_api.py
+++ b/ayon_api/server_api.py
@@ -375,7 +375,8 @@ def __init__(
             self.validate_server_availability()
             self.create_session()
 
-        self._event_hub: EventHub = EventHub(self)
+        # Created on demand by 'get_event_hub'
+        self._event_hub: EventHub | None = None
 
     @property
     def log(self) -> logging.Logger:
@@ -828,11 +829,16 @@ def validate_token(self) -> bool:
         return self._token_info.is_valid
 
     def set_token(self, token: str | None) -> None:
-        self.reset_token()
+        self._reset_token()
         self._token_info.token = token
         self.validate_token()
+        self._token_changed()
 
     def reset_token(self) -> None:
+        self._reset_token()
+        self._token_changed()
+
+    def _reset_token(self) -> None:
         self._token_info.token = None
         self._token_info.is_service = None
         self._token_info.is_valid = None
@@ -1227,14 +1233,27 @@ def get_headers(
     def get_event_hub(self) -> EventHub:
         """Get event hub for connection.
 
-        Event hub is used to subscribe to events from server.
+        Event hub is used to subscribe to events from server. The hub is
+            created on first call, websocket connection is created when
+            a callback is registered or when the hub is started.
+
+        Notes:
+            Running hub keeps reference to the connection, call
+                'EventHub.stop' when the connection is not used anymore.
 
         Returns:
             EventHub: Event hub instance.
 
         """
+        if self._event_hub is None:
+            self._event_hub = EventHub(self)
         return self._event_hub
 
+    def _token_changed(self) -> None:
+        """Let event hub know that token did change."""
+        if self._event_hub is not None:
+            self._event_hub.reauthenticate()
+
     def login(
         self,
         username: str,
@@ -1266,34 +1285,37 @@ def login(
                     self.create_session()
                 return
 
-        self.reset_token()
-
-        self.validate_server_availability()
-
-        response = self.raw_post(
-            "auth/login",
-            json=dict(
-                name=username,
-                password=password,
-            ),
-            handle_invalid_token=False,
-        )
-        if response.status_code != 200:
-            _detail = response.data.get("detail")
-            details = ""
-            if _detail:
-                details = f" {_detail}"
-
-            raise AuthenticationError(f"Login failed {details}")
-
-        self._token_info.token = response["token"]
-        # Should be valid if was just logged in
-        self._token_info.is_valid = True
-        # Service token can't be obtained by login, so it is not service token
-        self._token_info.is_service = False
+        self._reset_token()
+        try:
+            self.validate_server_availability()
 
-        if not self.has_valid_token:
-            raise AuthenticationError("Invalid credentials")
+            response = self.raw_post(
+                "auth/login",
+                json=dict(
+                    name=username,
+                    password=password,
+                ),
+                handle_invalid_token=False,
+            )
+            if response.status_code != 200:
+                _detail = response.data.get("detail")
+                details = ""
+                if _detail:
+                    details = f" {_detail}"
+
+                raise AuthenticationError(f"Login failed {details}")
+
+            self._token_info.token = response["token"]
+            # Should be valid if was just logged in
+            self._token_info.is_valid = True
+            # Service token can't be obtained by login, so it is not
+            #   service token
+            self._token_info.is_service = False
+
+            if not self.has_valid_token:
+                raise AuthenticationError("Invalid credentials")
+        finally:
+            self._token_changed()
 
         if create_session:
             self.create_session()
diff --git a/tests/test_event_hub_lifecycle.py b/tests/test_event_hub_lifecycle.py
new file mode 100644
index 000000000..3db7e3444
--- /dev/null
+++ b/tests/test_event_hub_lifecycle.py
@@ -0,0 +1,176 @@
+"""Lifecycle of event hub. Does not require running AYON server.
+
+Websocket connection is replaced with a fake socket which sends
+heartbeat messages, the same way as AYON server does.
+"""
+import gc
+import json
+import threading
+import time
+import weakref
+
+import pytest
+from websocket import ABNF, WebSocketConnectionClosedException
+
+from ayon_api.events import EventHub
+from ayon_api.server_api import ServerAPI
+
+WAIT_TIMEOUT = 5.0
+
+
+def wait_for(predicate, timeout=WAIT_TIMEOUT):
+    start = time.time()
+    while time.time() - start < timeout:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return False
+
+
+class FakeWebSocket:
+    """Fake websocket connection.
+
+    Sends heartbeat until aborted. If 'rejected_tokens' contains token
+        sent in auth message, the connection is closed by "server" without
+        sending any message.
+    """
+
+    def __init__(self, base_url, rejected_tokens):
+        self.base_url = base_url
+        self.sent = []
+        self._rejected_tokens = rejected_tokens
+        self._aborted = threading.Event()
+
+    @property
+    def tokens(self):
+        return [message["token"] for message in self.sent]
+
+    def settimeout(self, timeout):
+        pass
+
+    def send(self, data):
+        message = json.loads(data)
+        self.sent.append(message)
+        if message["token"] in self._rejected_tokens:
+            self._aborted.set()
+
+    def recv_data(self):
+        if self._aborted.wait(0.02):
+            raise WebSocketConnectionClosedException("closed")
+        if not self.sent:
+            # Server does not send anything before auth message
+            return self.recv_data()
+        return ABNF.OPCODE_TEXT, json.dumps({"topic": "heartbeat"})
+
+    def abort(self):
+        self._aborted.set()
+
+
+class Recorder:
+    """Callback storing received topics, must be kept in memory."""
+
+    def __init__(self):
+        self.topics = []
+
+    def __call__(self, event):
+        self.topics.append(event.topic)
+
+    def count(self, topic):
+        return self.topics.count(topic)
+
+
+class _Sockets(list):
+    rejected_tokens = None
+
+
+@pytest.fixture
+def sockets(monkeypatch):
+    created = _Sockets()
+    rejected_tokens = set()
+
+    def create_websocket(self, endpoint, **kwargs):
+        ws_con = FakeWebSocket(self.get_base_url(), rejected_tokens)
+        created.append(ws_con)
+        return ws_con
+
+    def check_token(self):
+        return self._connection.access_token not in rejected_tokens
+
+    monkeypatch.setattr(ServerAPI, "create_websocket", create_websocket)
+    monkeypatch.setattr(ServerAPI, "validate_token", lambda self: True)
+    monkeypatch.setattr(EventHub, "_check_token", check_token)
+    created.rejected_tokens = rejected_tokens
+    return created
+
+
+@pytest.fixture
+def con():
+    return ServerAPI("http://server-a:0", "token-a", create_session=False)
+
+
+@pytest.fixture
+def hubs():
+    created = []
+    yield created
+    for hub in created:
+        hub.stop()
+
+
+def _start_hub(hub, hubs):
+    hubs.append(hub)
+    recorder = Recorder()
+    hub.add_callback("*", recorder)
+    return recorder
+
+
+def test_event_hub_is_created_on_demand(con):
+    assert con._event_hub is None
+    hub = con.get_event_hub()
+    assert hub is con.get_event_hub()
+    assert not hub.is_running()
+
+
+def test_set_token_reauthenticates_open_connection(con, sockets, hubs):
+    recorder = _start_hub(con.get_event_hub(), hubs)
+    assert wait_for(lambda: recorder.count(EventHub.TOPIC_CONNECTION_OPENED))
+    assert sockets[0].tokens == ["token-a"]
+
+    con.set_token("token-b")
+
+    assert wait_for(lambda: sockets[0].tokens[-1] == "token-b")
+    # Same websocket connection is used
+    assert len(sockets) == 1
+    assert recorder.count(EventHub.TOPIC_CONNECTION_CLOSED) == 0
+
+
+def test_token_change_wakes_hub_after_auth_failure(con, sockets, hubs):
+    sockets.rejected_tokens.add("token-a")
+    hub = con.get_event_hub()
+    recorder = _start_hub(hub, hubs)
+    assert wait_for(lambda: recorder.count(EventHub.TOPIC_AUTH_FAILED))
+    assert hub.is_auth_failed()
+
+    # Without wake up the hub would wait 30 seconds before next attempt
+    con.set_token("token-b")
+
+    assert wait_for(
+        lambda: recorder.count(EventHub.TOPIC_CONNECTION_OPENED), timeout=2.0
+    )
+    assert not hub.is_auth_failed()
+    assert sockets[-1].tokens == ["token-b"]
+
+
+def test_stopped_hub_is_garbage_collected(sockets):
+    con = ServerAPI("http://server-a:0", "token-a", create_session=False)
+    hub = con.get_event_hub()
+    recorder = Recorder()
+    hub.add_callback("*", recorder)
+    assert wait_for(lambda: recorder.count(EventHub.TOPIC_CONNECTION_OPENED))
+    hub.stop()
+
+    con_ref = weakref.ref(con)
+    hub_ref = weakref.ref(hub)
+    del con, hub
+    gc.collect()
+    assert hub_ref() is None
+    assert con_ref() is None