Skip to content

Event Hook Setup

on_setup() is called once when the event hook is registered, before any events fire. Use it to initialise state in self.environment and load any external configuration.

async def on_setup(self) -> None:
    self.environment.counts = {
        EventType.AGENT_REGISTERED: 0,
        EventType.AGENT_CHECKED_IN: 0,
        EventType.AGENT_TASKED: 0,
        EventType.AGENT_TASK_COMPLETED: 0,
        EventType.AGENT_DEREGISTERED: 0,
    }
    self.logger.info("Agent Activity Tracker is ready.")

self.environment

self.environment is a types.SimpleNamespace created fresh each time the hook is instantiated. Use it to carry mutable state across on_triggered() calls:

# Initialise in on_setup
self.environment.counts = {}
self.environment.last_agent_name = None

# Read and mutate in on_triggered
self.environment.counts[event.event_type] += 1

Do not store runtime state as direct instance attributes. Using self.environment keeps state clearly separated from class-level configuration and from the framework-managed attributes on self.

self.logger

self.logger is a loguru logger bound with the hook's name and ID. Use {} placeholder syntax rather than f-strings:

self.logger.info("Processing event: {}.", event.event_type)
self.logger.warning("Unexpected event type: {}.", event.event_type)

Loading config files

If your hook needs external configuration (like webhook_sender), load it in on_setup(). Use self.root_directory to locate sibling files:

async def on_setup(self) -> None:
    config_path = self.root_directory / "config.json"
    with config_path.open("r") as f:
        self.environment.config = json.load(f)

See below for what to do if config_path does not exist.

Signalling setup failures

Raise EventHookSetupError from on_setup() to signal that the event hook cannot be set up. This aborts registration: the event hook is not loaded, on_triggered() is never called for it, and the registry surfaces the failure as an EventHookSetupError (from consortium.server.exceptions.consortium_exceptions.event_hooks_consortium_exceptions) carrying your message and any structured detail you provide.

from consortium.framework.signal_exceptions.event_hooks_signal_exceptions import (
    EventHookSetupError,
)


async def on_setup(self) -> None:
    config_path = self.root_directory / "config.json"
    if not config_path.exists():
        raise EventHookSetupError(
            message=f"config.json not found at '{config_path}'.",
            detail={"config_path": str(config_path)},
        )
    with config_path.open("r") as f:
        self.environment.config = json.load(f)

Do not catch and log setup failures internally, and do not let a bare, unexpected exception escape on_setup() unremarked. Either way the event hook still fails to load: raising EventHookSetupError explicitly gives the caller a clean, structured error instead of a hook that appears loaded but is silently missing state, or a raw traceback with no framework context.

Continue to Event Hook Triggering to react to events.