App Module

The App module provides functions for initializing and securing apps that interact with the Unique platform.

Overview

The unique_toolkit.app module encompasses functions for: - Initializing the SDK and logging - Handling events from the platform - Verifying webhook signatures - Building FastAPI applications - Running async tasks in parallel

Components

Settings

unique_toolkit.app.unique_settings.UniqueSettings

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

class UniqueSettings:
    def __init__(
        self,
        auth: UniqueAuth,
        app: UniqueApp,
        api: UniqueApi,
        *,
        chat_event_filter_options: UniqueChatEventFilterOptions | None = None,
        env_file: Path | None = None,
    ):
        self._app = app
        self._auth = auth
        self._api = api
        self._chat_event_filter_options = chat_event_filter_options
        self._env_file: Path | None = (
            env_file if (env_file and env_file.exists()) else None
        )

    @classmethod
    def _find_env_file(cls, filename: str = "unique.env") -> Path:
        """Find environment file using cross-platform fallback locations.

        Search order:
        1. UNIQUE_ENV_FILE environment variable
        2. Current working directory
        3. User config directory (cross-platform via platformdirs)

        Args:
            filename: Name of the environment file (default: 'unique.env')

        Returns:
            Path to the environment file.

        Raises:
            EnvFileNotFoundError: If no environment file is found in any location.
        """
        locations = [
            # 1. Explicit environment variable
            Path(env_path) if (env_path := os.environ.get("UNIQUE_ENV_FILE")) else None,
            # 2. Current working directory
            Path.cwd() / filename,
            # 3. User config directory (cross-platform)
            Path(user_config_dir("unique", "unique-toolkit")) / filename,
        ]

        for location in locations:
            if location and location.exists() and location.is_file():
                return location

        # If no file found, provide helpful error message
        searched_locations = [str(loc) for loc in locations if loc is not None]
        raise EnvFileNotFoundError(
            f"Environment file '{filename}' not found. Searched locations:\n"
            + "\n".join(f"  - {loc}" for loc in searched_locations)
            + "\n\nTo fix this:\n"
            + f"  1. Create {filename} in one of the above locations, or\n"
            + f"  2. Set UNIQUE_ENV_FILE environment variable to point to your {filename} file"
        )

    @classmethod
    def from_env(
        cls,
        env_file: Path | None = None,
    ) -> "UniqueSettings":
        """Initialize settings from environment variables and/or env file.

        Args:
            env_file: Optional path to environment file. If provided, will load variables from this file.

        Returns:
            UniqueSettings instance with values loaded from environment/env file.

        Raises:
            FileNotFoundError: If env_file is provided but does not exist.
            ValidationError: If required environment variables are missing.
        """
        if env_file and not env_file.exists():
            raise FileNotFoundError(f"Environment file not found: {env_file}")

        # Initialize settings with environment file if provided
        env_file_str = str(env_file) if env_file else None
        auth = UniqueAuth(_env_file=env_file_str)  # type: ignore[call-arg]
        app = UniqueApp(_env_file=env_file_str)  # type: ignore[call-arg]
        api = UniqueApi(_env_file=env_file_str)  # type: ignore[call-arg]
        event_filter_options = UniqueChatEventFilterOptions(_env_file=env_file_str)  # type: ignore[call-arg]
        return cls(
            auth=auth,
            app=app,
            api=api,
            chat_event_filter_options=event_filter_options,
            env_file=env_file,
        )

    @classmethod
    def from_env_auto(cls, filename: str = "unique.env") -> "UniqueSettings":
        """Initialize settings by automatically finding environment file.

        This method will automatically search for an environment file in standard locations
        and fall back to environment variables only if no file is found.

        Args:
            filename: Name of the environment file to search for (default: '.env')

        Returns:
            UniqueSettings instance with values loaded from found env file or environment variables.
        """
        try:
            env_file = cls._find_env_file(filename)
            logger.info(f"Environment file found at {env_file}")
            return cls.from_env(env_file=env_file)
        except EnvFileNotFoundError:
            logger.warning(
                f"Environment file '{filename}' not found. Falling back to environment variables only."
            )
            # Fall back to environment variables only
            return cls.from_env()

    def init_sdk(self) -> None:
        """Initialize the unique_sdk global configuration with these settings.

        This method configures the global unique_sdk module with the API key,
        app ID, and base URL from these settings.
        """
        unique_sdk.api_key = self._app.key.get_secret_value()
        unique_sdk.app_id = self._app.id.get_secret_value()
        unique_sdk.api_base = self._api.sdk_url()

    @classmethod
    def from_env_auto_with_sdk_init(
        cls, filename: str = "unique.env"
    ) -> "UniqueSettings":
        """Initialize settings and SDK in one convenient call.

        This method combines from_env_auto() and init_sdk() for the most common use case.

        Args:
            filename: Name of the environment file to search for (default: '.env')

        Returns:
            UniqueSettings instance with SDK already initialized.
        """
        settings = cls.from_env_auto(filename)
        settings.init_sdk()
        return settings

    def update_from_event(self, event: "BaseEvent") -> None:
        self._auth = UniqueAuth.from_event(event)

    @property
    def api(self) -> UniqueApi:
        return self._api

    @property
    def app(self) -> UniqueApp:
        return self._app

    @property
    def auth(self) -> UniqueAuth:
        return self._auth

    @auth.setter
    def auth(self, value: UniqueAuth) -> None:
        self._auth = value

    @property
    def chat_event_filter_options(self) -> UniqueChatEventFilterOptions | None:
        return self._chat_event_filter_options

_find_env_file(filename='unique.env') classmethod

Find environment file using cross-platform fallback locations.

Search order: 1. UNIQUE_ENV_FILE environment variable 2. Current working directory 3. User config directory (cross-platform via platformdirs)

Parameters:

Name	Type	Description	Default
filename	str	Name of the environment file (default: 'unique.env')	'unique.env'

Returns:

Type	Description
Path	Path to the environment file.

Raises:

Type	Description
EnvFileNotFoundError	If no environment file is found in any location.

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

@classmethod
def _find_env_file(cls, filename: str = "unique.env") -> Path:
    """Find environment file using cross-platform fallback locations.

    Search order:
    1. UNIQUE_ENV_FILE environment variable
    2. Current working directory
    3. User config directory (cross-platform via platformdirs)

    Args:
        filename: Name of the environment file (default: 'unique.env')

    Returns:
        Path to the environment file.

    Raises:
        EnvFileNotFoundError: If no environment file is found in any location.
    """
    locations = [
        # 1. Explicit environment variable
        Path(env_path) if (env_path := os.environ.get("UNIQUE_ENV_FILE")) else None,
        # 2. Current working directory
        Path.cwd() / filename,
        # 3. User config directory (cross-platform)
        Path(user_config_dir("unique", "unique-toolkit")) / filename,
    ]

    for location in locations:
        if location and location.exists() and location.is_file():
            return location

    # If no file found, provide helpful error message
    searched_locations = [str(loc) for loc in locations if loc is not None]
    raise EnvFileNotFoundError(
        f"Environment file '{filename}' not found. Searched locations:\n"
        + "\n".join(f"  - {loc}" for loc in searched_locations)
        + "\n\nTo fix this:\n"
        + f"  1. Create {filename} in one of the above locations, or\n"
        + f"  2. Set UNIQUE_ENV_FILE environment variable to point to your {filename} file"
    )

from_env(env_file=None) classmethod

Initialize settings from environment variables and/or env file.

Parameters:

Name	Type	Description	Default
env_file	Path | None	Optional path to environment file. If provided, will load variables from this file.	None

Returns:

Type	Description
UniqueSettings	UniqueSettings instance with values loaded from environment/env file.

Raises:

Type	Description
FileNotFoundError	If env_file is provided but does not exist.
ValidationError	If required environment variables are missing.

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

@classmethod
def from_env(
    cls,
    env_file: Path | None = None,
) -> "UniqueSettings":
    """Initialize settings from environment variables and/or env file.

    Args:
        env_file: Optional path to environment file. If provided, will load variables from this file.

    Returns:
        UniqueSettings instance with values loaded from environment/env file.

    Raises:
        FileNotFoundError: If env_file is provided but does not exist.
        ValidationError: If required environment variables are missing.
    """
    if env_file and not env_file.exists():
        raise FileNotFoundError(f"Environment file not found: {env_file}")

    # Initialize settings with environment file if provided
    env_file_str = str(env_file) if env_file else None
    auth = UniqueAuth(_env_file=env_file_str)  # type: ignore[call-arg]
    app = UniqueApp(_env_file=env_file_str)  # type: ignore[call-arg]
    api = UniqueApi(_env_file=env_file_str)  # type: ignore[call-arg]
    event_filter_options = UniqueChatEventFilterOptions(_env_file=env_file_str)  # type: ignore[call-arg]
    return cls(
        auth=auth,
        app=app,
        api=api,
        chat_event_filter_options=event_filter_options,
        env_file=env_file,
    )

from_env_auto(filename='unique.env') classmethod

Initialize settings by automatically finding environment file.

This method will automatically search for an environment file in standard locations and fall back to environment variables only if no file is found.

Parameters:

Name	Type	Description	Default
filename	str	Name of the environment file to search for (default: '.env')	'unique.env'

Returns:

Type	Description
UniqueSettings	UniqueSettings instance with values loaded from found env file or environment variables.

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

@classmethod
def from_env_auto(cls, filename: str = "unique.env") -> "UniqueSettings":
    """Initialize settings by automatically finding environment file.

    This method will automatically search for an environment file in standard locations
    and fall back to environment variables only if no file is found.

    Args:
        filename: Name of the environment file to search for (default: '.env')

    Returns:
        UniqueSettings instance with values loaded from found env file or environment variables.
    """
    try:
        env_file = cls._find_env_file(filename)
        logger.info(f"Environment file found at {env_file}")
        return cls.from_env(env_file=env_file)
    except EnvFileNotFoundError:
        logger.warning(
            f"Environment file '{filename}' not found. Falling back to environment variables only."
        )
        # Fall back to environment variables only
        return cls.from_env()

from_env_auto_with_sdk_init(filename='unique.env') classmethod

Initialize settings and SDK in one convenient call.

This method combines from_env_auto() and init_sdk() for the most common use case.

Parameters:

Name	Type	Description	Default
filename	str	Name of the environment file to search for (default: '.env')	'unique.env'

Returns:

Type	Description
UniqueSettings	UniqueSettings instance with SDK already initialized.

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

@classmethod
def from_env_auto_with_sdk_init(
    cls, filename: str = "unique.env"
) -> "UniqueSettings":
    """Initialize settings and SDK in one convenient call.

    This method combines from_env_auto() and init_sdk() for the most common use case.

    Args:
        filename: Name of the environment file to search for (default: '.env')

    Returns:
        UniqueSettings instance with SDK already initialized.
    """
    settings = cls.from_env_auto(filename)
    settings.init_sdk()
    return settings

init_sdk()

Initialize the unique_sdk global configuration with these settings.

This method configures the global unique_sdk module with the API key, app ID, and base URL from these settings.

Source code in unique_toolkit/unique_toolkit/app/unique_settings.py

def init_sdk(self) -> None:
    """Initialize the unique_sdk global configuration with these settings.

    This method configures the global unique_sdk module with the API key,
    app ID, and base URL from these settings.
    """
    unique_sdk.api_key = self._app.key.get_secret_value()
    unique_sdk.app_id = self._app.id.get_secret_value()
    unique_sdk.api_base = self._api.sdk_url()

Initialization

unique_toolkit.app.init_sdk.init_sdk(strict_all_vars=False)

Initialize the SDK.

Parameters:

Name	Type	Description	Default
strict_all_vars	bool	This method raises a ValueError if strict and no value is found in the environment. Defaults to False.	False

Source code in unique_toolkit/unique_toolkit/app/init_sdk.py

@deprecated("Use init_unique_sdk instead")
def init_sdk(
    strict_all_vars: bool = False,
):
    """Initialize the SDK.

    Args:
        strict_all_vars (bool, optional): This method raises a ValueError if strict and no value is found in the environment. Defaults to False.
    """

    unique_sdk.api_key = get_env("API_KEY", default="dummy", strict=strict_all_vars)
    unique_sdk.app_id = get_env("APP_ID", default="dummy", strict=strict_all_vars)
    unique_sdk.api_base = get_env("API_BASE", default=None, strict=strict_all_vars)

unique_toolkit.app.init_sdk.init_unique_sdk(*, unique_settings=None, env_file=None)

init_unique_sdk(*, env_file: Path | None = None)

init_unique_sdk(*, unique_settings: UniqueSettings)

Source code in unique_toolkit/unique_toolkit/app/init_sdk.py

def init_unique_sdk(
    *, unique_settings: UniqueSettings | None = None, env_file: Path | None = None
):
    if unique_settings:
        unique_sdk.api_key = unique_settings.app.key.get_secret_value()
        unique_sdk.app_id = unique_settings.app.id.get_secret_value()
        unique_sdk.api_base = unique_settings.api.sdk_url()
    elif env_file:
        unique_settings = UniqueSettings.from_env(env_file=env_file)
        unique_sdk.api_key = unique_settings.app.key.get_secret_value()
        unique_sdk.app_id = unique_settings.app.id.get_secret_value()
        unique_sdk.api_base = unique_settings.api.sdk_url()

unique_toolkit.app.init_logging.init_logging(config=unique_log_config)

Source code in unique_toolkit/unique_toolkit/app/init_logging.py

def init_logging(config: dict = unique_log_config):
    return dictConfig(config)

Event Schemas

unique_toolkit.app.schemas.ChatEvent

Bases: BaseEvent

Source code in unique_toolkit/unique_toolkit/app/schemas.py

class ChatEvent(BaseEvent):
    model_config = model_config

    payload: ChatEventPayload
    created_at: Optional[int] = None
    version: Optional[str] = None

    @classmethod
    def from_json_file(cls, file_path: Path) -> "ChatEvent":
        if not file_path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")
        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
        return cls.model_validate(data)

    def get_initial_debug_info(self) -> dict[str, Any]:
        """Get the debug information for the chat event"""

        # TODO: Make sure this coincides with what is shown in the first user message
        return {
            "user_metadata": self.payload.user_metadata,
            "tool_parameters": self.payload.tool_parameters,
            "chosen_module": self.payload.name,
            "assistant": {"id": self.payload.assistant_id},
        }

    @override
    def filter_event(
        self, *, filter_options: UniqueChatEventFilterOptions | None = None
    ) -> bool:
        # Empty string evals to False

        if filter_options is None:
            return False  # Don't filter when no options provided

        if not filter_options.assistant_ids and not filter_options.references_in_code:
            raise ConfigurationException(
                "No filter options provided, all events will be filtered! \n"
                "Please define: \n"
                " - 'UNIQUE_CHAT_EVENT_FILTER_OPTIONS_ASSISTANT_IDS' \n"
                " - 'UNIQUE_CHAT_EVENT_FILTER_OPTIONS_REFERENCES_IN_CODE' \n"
                "in your environment variables."
            )

        # Per reference in code there can be multiple assistants
        if (
            filter_options.assistant_ids
            and self.payload.assistant_id not in filter_options.assistant_ids
        ):
            return True

        if (
            filter_options.references_in_code
            and self.payload.name not in filter_options.references_in_code
        ):
            return True

        return super().filter_event(filter_options=filter_options)

get_initial_debug_info()

Get the debug information for the chat event

Source code in unique_toolkit/unique_toolkit/app/schemas.py

def get_initial_debug_info(self) -> dict[str, Any]:
    """Get the debug information for the chat event"""

    # TODO: Make sure this coincides with what is shown in the first user message
    return {
        "user_metadata": self.payload.user_metadata,
        "tool_parameters": self.payload.tool_parameters,
        "chosen_module": self.payload.name,
        "assistant": {"id": self.payload.assistant_id},
    }

unique_toolkit.app.schemas.Event

Bases: ChatEvent

Source code in unique_toolkit/unique_toolkit/app/schemas.py

@deprecated(
    """Use the more specific `ChatEvent` instead that has the same properties. \
This class will be removed in the next major version."""
)
class Event(ChatEvent):
    pass
    # The below should only affect type hints
    # event: EventName T
    # payload: EventPayload

    @classmethod
    def from_json_file(cls, file_path: Path) -> "Event":
        if not file_path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")
        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
        return cls.model_validate(data)

unique_toolkit.app.schemas.BaseEvent

Bases: BaseModel, Generic[FilterOptionsT]

Source code in unique_toolkit/unique_toolkit/app/schemas.py

class BaseEvent(BaseModel, Generic[FilterOptionsT]):
    model_config = model_config

    id: str
    event: str
    user_id: str
    company_id: str

    @classmethod
    def from_json_file(cls, file_path: Path) -> "BaseEvent":
        if not file_path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")
        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
        return cls.model_validate(data)

    def filter_event(self, *, filter_options: FilterOptionsT | None = None) -> bool:
        """Determine if event should be filtered out and be neglected."""
        return False

unique_toolkit.app.schemas.EventName

Bases: StrEnum

Source code in unique_toolkit/unique_toolkit/app/schemas.py

class EventName(StrEnum):
    EXTERNAL_MODULE_CHOSEN = "unique.chat.external-module.chosen"
    USER_MESSAGE_CREATED = "unique.chat.user-message.created"
    INGESTION_CONTENT_UPLOADED = "unique.ingestion.content.uploaded"
    INGESTION_CONTENT_FINISHED = "unique.ingestion.content.finished"
    MAGIC_TABLE_IMPORT_COLUMNS = "unique.magic-table.import-columns"
    MAGIC_TABLE_ADD_META_DATA = "unique.magic-table.add-meta-data"
    MAGIC_TABLE_ADD_DOCUMENT = "unique.magic-table.add-document"
    MAGIC_TABLE_DELETE_ROW = "unique.magic-table.delete-row"
    MAGIC_TABLE_DELETE_COLUMN = "unique.magic-table.delete-column"
    MAGIC_TABLE_UPDATE_CELL = "unique.magic-table.update-cell"

Verification

unique_toolkit.app.verification.verify_signature_and_construct_event(headers, payload, endpoint_secret, logger=logger, event_constructor=Event)

Verify the signature of a webhook and construct an event object.

Parameters:

Name	Type	Description	Default
headers	Dict[str, str]	The headers of the webhook request.	required
payload	bytes	The raw payload of the webhook request.	required
endpoint_secret	str	The secret used to verify the webhook signature.	required
logger	Logger	A logger instance for logging messages.	logger
event_constructor	Callable[..., T]	A callable that constructs an event object.	Event

Raises:

Type	Description
WebhookVerificationError	If there's an error during verification or event construction.

Source code in unique_toolkit/unique_toolkit/app/verification.py

def verify_signature_and_construct_event(
    headers: dict[str, str],
    payload: bytes,
    endpoint_secret: str,
    logger: logging.Logger = logger,
    event_constructor: Callable[..., T] = Event,
) -> T:
    """
    Verify the signature of a webhook and construct an event object.

    Args:
        headers (Dict[str, str]): The headers of the webhook request.
        payload (bytes): The raw payload of the webhook request.
        endpoint_secret (str): The secret used to verify the webhook signature.
        logger (logging.Logger): A logger instance for logging messages.
        event_constructor (Callable[..., T]): A callable that constructs an event object.
    Returns:
        T: The constructed event object.

    Raises:
        WebhookVerificationError: If there's an error during verification or event construction.
    """

    sig_header = headers.get("X-Unique-Signature")
    timestamp = headers.get("X-Unique-Created-At")

    if not sig_header or not timestamp:
        logger.error("⚠️  Webhook signature or timestamp headers missing.")
        raise WebhookVerificationError("Signature or timestamp headers missing")

    try:
        event = unique_sdk.Webhook.construct_event(
            payload,
            sig_header,
            timestamp,
            endpoint_secret,
        )
        logger.info("✅  Webhook signature verification successful.")
        return event_constructor(**event)
    except unique_sdk.SignatureVerificationError as e:
        logger.error("⚠️  Webhook signature verification failed. " + str(e))
        raise WebhookVerificationError(f"Signature verification failed: {str(e)}")

unique_toolkit.app.webhook.is_webhook_signature_valid(headers, payload, endpoint_secret, tolerance=300)

Verify webhook signature from Unique platform.

Parameters:

Name	Type	Description	Default
headers	dict[str, str]	Request headers with X-Unique-Signature and X-Unique-Created-At	required
payload	bytes	Raw request body bytes	required
endpoint_secret	str	App endpoint secret from Unique platform	required
tolerance	int	Max seconds between timestamp and now (default: 300)	300

Returns:

Type	Description
bool	True if signature is valid, False otherwise

Source code in unique_toolkit/unique_toolkit/app/webhook.py

def is_webhook_signature_valid(
    headers: dict[str, str],
    payload: bytes,
    endpoint_secret: str,
    tolerance: int = 300,
) -> bool:
    """
    Verify webhook signature from Unique platform.

    Args:
        headers: Request headers with X-Unique-Signature and X-Unique-Created-At
        payload: Raw request body bytes
        endpoint_secret: App endpoint secret from Unique platform
        tolerance: Max seconds between timestamp and now (default: 300)

    Returns:
        True if signature is valid, False otherwise
    """
    # Extract headers
    signature = headers.get("X-Unique-Signature") or headers.get("x-unique-signature")
    timestamp_str = headers.get("X-Unique-Created-At") or headers.get(
        "x-unique-created-at"
    )

    if not signature:
        _LOGGER.error("Missing X-Unique-Signature header")
        return False

    if not timestamp_str:
        _LOGGER.error("Missing X-Unique-Created-At header")
        return False

    # Convert timestamp to int
    try:
        timestamp = int(timestamp_str)
    except ValueError:
        _LOGGER.error(f"Invalid timestamp: {timestamp_str}")
        return False

    # Decode payload if bytes
    message = payload.decode("utf-8") if isinstance(payload, bytes) else payload

    # Compute expected signature: HMAC-SHA256(message, secret)
    expected_signature = hmac.new(
        endpoint_secret.encode("utf-8"),
        msg=message.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).hexdigest()

    # Compare signatures (constant-time to prevent timing attacks)
    if not hmac.compare_digest(expected_signature, signature):
        _LOGGER.error("Signature mismatch. Ensure you're using the raw request body.")
        return False

    # Check timestamp tolerance (prevent replay attacks)
    if tolerance and timestamp < time.time() - tolerance:
        _LOGGER.error(
            f"Timestamp outside tolerance ({tolerance}s). Possible replay attack."
        )
        return False

    _LOGGER.debug("✅ Webhook signature verified successfully")
    return True

FastAPI Factory

unique_toolkit.app.fast_api_factory.build_unique_custom_app(*, title='Unique Chat App', webhook_path='/webhook', settings, event_handler=default_event_handler, event_constructor=ChatEvent, subscribed_event_names=None)

Factory class for creating FastAPI apps with Unique webhook handling.

Source code in unique_toolkit/unique_toolkit/app/fast_api_factory.py

def build_unique_custom_app(
    *,
    title: str = "Unique Chat App",
    webhook_path: str = "/webhook",
    settings: UniqueSettings,
    event_handler: Callable[[T], int] = default_event_handler,
    event_constructor: Callable[..., T] = ChatEvent,
    subscribed_event_names: list[str] | None = None,
) -> "FastAPI":
    """Factory class for creating FastAPI apps with Unique webhook handling."""
    if FastAPI is None:
        raise ImportError(
            "FastAPI is not installed. Install it with: poetry install --with fastapi"
        )

    app = FastAPI(title=title)

    if subscribed_event_names is None:
        subscribed_event_names = [EventName.EXTERNAL_MODULE_CHOSEN]

    @app.get(path="/")
    async def health_check() -> JSONResponse:
        """Health check endpoint."""
        return JSONResponse(content={"status": "healthy", "service": title})

    @app.post(path=webhook_path)
    async def webhook_handler(
        request: Request, background_tasks: BackgroundTasks
    ) -> JSONResponse:
        """
        Webhook endpoint for receiving events from Unique platform.

        This endpoint:
        1. Verifies the webhook signature
        2. Constructs an event from the payload
        3. Calls the configured event handler
        """
        # Get raw body and headers
        body = await request.body()
        headers = dict(request.headers)

        from unique_toolkit.app.webhook import is_webhook_signature_valid

        if not is_webhook_signature_valid(
            headers=headers,
            payload=body,
            endpoint_secret=settings.app.endpoint_secret.get_secret_value(),
        ):
            return JSONResponse(
                status_code=status.HTTP_401_UNAUTHORIZED,
                content={"error": "Invalid webhook signature"},
            )

        try:
            event_data = json.loads(body.decode(encoding="utf-8"))
        except json.JSONDecodeError as e:
            logger.error(f"Error parsing event: {e}", exc_info=True)
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"error": f"Invalid event format: {e.msg}"},
            )

        if event_data["event"] not in subscribed_event_names:
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"error": "Not subscribed event"},
            )

        try:
            event = event_constructor(**event_data)
            if event.filter_event(filter_options=settings.chat_event_filter_options):
                return JSONResponse(
                    status_code=status.HTTP_200_OK,
                    content={"error": "Event filtered out"},
                )
        except ValidationError as e:
            # pydantic errors https://docs.pydantic.dev/2.10/errors/errors/
            logger.error(f"Validation error with model: {e.json()}", exc_info=True)
            raise e
        except ValueError as e:
            logger.error(f"Error deserializing event: {e}", exc_info=True)
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"error": "Invalid event"},
            )

        # Run the task in background so that we don't block for long running tasks
        background_tasks.add_task(event_handler, event)
        return JSONResponse(
            status_code=status.HTTP_200_OK, content={"message": "Event received"}
        )

    return app