Observers API Reference

This page provides detailed documentation for the observers in GUM.

Base Observer

gum.observers.Observer(name: Optional[str] = None)

Bases: ABC

Base class for all observers in the GUM system.

This abstract base class defines the interface for all observers that monitor user behavior. Observers are responsible for collecting data about user interactions and sending updates through an asynchronous queue.

Parameters:

Name	Type	Description	Default
name	Optional[str]	A custom name for the observer. If not provided, the class name will be used.	None

Attributes:

Name	Type	Description
update_queue	Queue	Queue for sending updates to the main GUM system.
_name	str	The name of the observer.
_running	bool	Flag indicating if the observer is currently running.
_task	Optional[Task]	Background task handle for the observer's worker.

Source code in gum/observers/observer.py

def __init__(self, name: Optional[str] = None) -> None:
    self.update_queue = asyncio.Queue()
    self._name = name or self.__class__.__name__

    # running flag + background task handle
    self._running = True
    self._task: asyncio.Task | None = asyncio.create_task(self._worker_wrapper())

Attributes

name: str property

Get the name of the observer.

Returns:

Name	Type	Description
str	str	The observer's name.

update_queue = asyncio.Queue() instance-attribute

Functions

get_update() async

Get the next update from the queue if available.

Returns:

Type	Description
	Optional[Update]: The next update from the queue, or None if the queue is empty.

Source code in gum/observers/observer.py

async def get_update(self):
    """Get the next update from the queue if available.

    Returns:
        Optional[Update]: The next update from the queue, or None if the queue is empty.
    """
    try:
        return self.update_queue.get_nowait()
    except asyncio.QueueEmpty:
        return None

set_session_factory(factory) -> None

Provide an async session factory for observers that need direct DB access.

Default is a no-op; override in subclasses that persist raw events (e.g. FocusObserver writing to focus_events).

Source code in gum/observers/observer.py

def set_session_factory(self, factory) -> None:
    """Provide an async session factory for observers that need direct DB access.

    Default is a no-op; override in subclasses that persist raw events
    (e.g. FocusObserver writing to focus_events).
    """
    pass

stop() -> None async

Stop the observer and clean up resources.

This method cancels the worker task and drains the update queue.

Source code in gum/observers/observer.py

async def stop(self) -> None:
    """Stop the observer and clean up resources.

    This method cancels the worker task and drains the update queue.
    """
    if self._task and not self._task.done():
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass
    # unblock any awaiters
    while not self.update_queue.empty():
        self.update_queue.get_nowait()

Screen Observer

gum.observers.Screen(model_name: str = 'gpt-4.1-mini', screenshots_dir: str = '~/.cache/gum/screenshots', skip_when_visible: Optional[str | list[str]] = None, transcription_prompt: Optional[str] = None, summary_prompt: Optional[str] = None, history_k: int = 10, debug: bool = False, api_key: str | None = None, api_base: str | None = None, inject_app_metadata: bool = True)

Bases: Observer

Observer that captures and analyzes screen content around user interactions.

Captures screenshots before and after user interactions (mouse clicks and scrolls) and uses a vision LLM to analyze the content. Also injects the currently focused application name and window title into the prompt to prevent the LLM from misidentifying the active application.

Parameters:

Name	Type	Description	Default
model_name	str	Vision model to use. Defaults to "gpt-4.1-mini".	'gpt-4.1-mini'
screenshots_dir	str	Directory to store screenshots.	'~/.cache/gum/screenshots'
skip_when_visible	Optional[str | list[str]]	App names to skip when visible.	None
transcription_prompt	Optional[str]	Custom transcription prompt.	None
summary_prompt	Optional[str]	Custom summary prompt.	None
history_k	int	Number of recent screenshots to keep in history.	10
debug	bool	Enable debug logging.	False
api_key	str | None	API key override.	None
api_base	str | None	API base URL override.	None

Source code in gum/observers/screen.py

def __init__(
    self,
    model_name: str = "gpt-4.1-mini",
    screenshots_dir: str = "~/.cache/gum/screenshots",
    skip_when_visible: Optional[str | list[str]] = None,
    transcription_prompt: Optional[str] = None,
    summary_prompt: Optional[str] = None,
    history_k: int = 10,
    debug: bool = False,
    api_key: str | None = None,
    api_base: str | None = None,
    inject_app_metadata: bool = True,
) -> None:
    self.screens_dir = os.path.abspath(os.path.expanduser(screenshots_dir))
    os.makedirs(self.screens_dir, exist_ok=True)

    self._guard = {skip_when_visible} if isinstance(skip_when_visible, str) else set(skip_when_visible or [])

    self.transcription_prompt = transcription_prompt or TRANSCRIPTION_PROMPT
    self.summary_prompt = summary_prompt or SUMMARY_PROMPT
    self.model_name = model_name

    self.debug = debug
    # Env var INJECT_APP_METADATA=false overrides the constructor param.
    env_val = os.getenv("INJECT_APP_METADATA", "").lower()
    if env_val == "false":
        self.inject_app_metadata = False
    elif env_val == "true":
        self.inject_app_metadata = True
    else:
        self.inject_app_metadata = inject_app_metadata

    # state shared with worker
    self._frames: Dict[int, Any] = {}
    self._frame_lock = asyncio.Lock()
    self._history: deque[str] = deque(maxlen=max(0, history_k))
    self._pending_event: Optional[dict] = None
    self._debounce_deadline: float | None = None
    self._generator = ScreenObservationGenerator(
        model_name=model_name,
        transcription_prompt=self.transcription_prompt,
        summary_prompt=self.summary_prompt,
        api_key=api_key,
        api_base=api_base,
    )
    self.client = self._generator.client

    super().__init__()

Attributes

client = self._generator.client instance-attribute

debug = debug instance-attribute

inject_app_metadata = False instance-attribute

model_name = model_name instance-attribute

screens_dir = os.path.abspath(os.path.expanduser(screenshots_dir)) instance-attribute

summary_prompt = summary_prompt or SUMMARY_PROMPT instance-attribute

transcription_prompt = transcription_prompt or TRANSCRIPTION_PROMPT instance-attribute

Functions