API Reference

This reference is rendered from current signatures and docstrings by mkdocstrings. The public API inventory defines the deliberate stable import surface; conceptual behavior and Google mappings live in the task guides.

Core

Async, typed core event engine for Google Chat applications.

Dispatcher

Bases: Router

Root router and transport-independent event feed.

Source code in src/chattice/dispatcher/dispatcher.py

class Dispatcher(Router):
    """Root router and transport-independent event feed."""

    def lifespan(self, *resources: LifespanResource) -> Lifespan:
        """An async context manager starting resources in order and closing
        them in reverse (partial-start rollback included). Plug it into
        FastAPI via ``app.router.lifespan_context = dispatcher.lifespan(...)``.
        """
        return Lifespan(*resources)

    async def run_pubsub(
        self,
        subscription: str,
        *,
        bot: Bot | None = None,
        credentials: Credentials | None = None,
        credentials_provider: CredentialsProvider | None = None,
        max_concurrency: int = 10,
        max_outstanding_messages: int = 100,
        idempotency_storage: IdempotencyStorage | None = None,
        max_delivery_attempts: int = 5,
        stop_event: asyncio.Event | None = None,
    ) -> None:
        """Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

        Runs every delivery through THIS dispatcher's router/filter/
        middleware/DI pipeline. Handler answers go outbound through
        ``bot`` where semantics allow (text -> send_message, Card ->
        update_message/send_message); Dialog answers are rejected with
        ``CapabilityNotSupported`` (dialogs require the synchronous HTTP
        transport). Requires the ``chattice[pubsub]`` extra.

        Blocks until ``stop_event`` fires or SIGINT/SIGTERM; drains
        in-flight handlers before returning.
        """
        from chattice.transports.pubsub_runner import PubSubPullRunner

        runner = PubSubPullRunner(
            self,
            subscription,
            bot=bot,
            credentials=credentials,
            credentials_provider=credentials_provider,
            max_concurrency=max_concurrency,
            max_outstanding_messages=max_outstanding_messages,
            idempotency_storage=idempotency_storage,
            max_delivery_attempts=max_delivery_attempts,
            delayed_event_ms=self._runtime_diagnostics.delayed_event_ms,
            observability_hooks=self._observability_hooks,
        )
        await runner.run(stop_event=stop_event)

    def __init__(
        self,
        *,
        name: str = "dispatcher",
        bot: object | None = None,
        fsm_storage: BaseStorage | None = None,
        fsm_strategy: FSMStrategy = FSMStrategy.USER_IN_SPACE,
        # ``object``: implementations may provide ANY subset of the
        # optional ObservabilityHooks surface.
        observability_hooks: object | None = None,
        preview_features: Iterable[PreviewFeature] = (),
        strict_interactions: bool = False,
        runtime_diagnostics: RuntimeDiagnostics | None = None,
    ) -> None:
        super().__init__(name=name)
        self._is_dispatcher = True
        self._bot = bot
        self._fsm_storage = fsm_storage
        self._fsm_strategy = fsm_strategy
        self._observability_hooks = observability_hooks
        self._preview_capabilities = PreviewCapabilities(preview_features)
        self._strict_interactions = strict_interactions
        self._runtime_diagnostics = runtime_diagnostics or RuntimeDiagnostics()

    @property
    def preview_capabilities(self) -> PreviewCapabilities:
        """The immutable Developer Preview enrollment for typed routing."""
        return self._preview_capabilities

    async def feed_update(self, event: Event, **context: object) -> object:
        """Route one domain event and return the handler result unchanged."""
        if not isinstance(event, Event):
            raise TypeError("feed_update() accepts chattice Event instances only")
        _routing_logger.debug("event received: %s", _event_context(event))
        _routing_logger.debug("routing started: %s", _event_context(event))
        data = dict(context)
        if "bot" not in data and self._bot is not None:
            data["bot"] = self._bot
        contextual_bot = data.get("bot")
        # Configuration, not caller context, is the source of truth: a
        # feed_update() kwarg must not bypass explicit preview enrollment.
        data["preview_capabilities"] = self._preview_capabilities
        if self._fsm_storage is not None:
            data["state"] = FSMContext(
                self._fsm_storage, StorageKey.build(event, self._fsm_strategy)
            )
        result: object = None
        error: BaseException | None = None
        bot_token = _set_current_bot(contextual_bot)
        try:
            await _maybe_hook(self._observability_hooks, "before_event", event, data)
            try:
                outcome = await self._route_event(event, data)
                if outcome.handled:
                    result = outcome.result
            except BaseException as exc:
                error = exc
                if not isinstance(exc, Exception):
                    # CancelledError and friends bypass error routing but are
                    # still reported to the after_event hook, then re-raised.
                    raise
                error_event = ErrorEvent(
                    source_event=event,
                    exception=exc,
                    raw=event.raw,
                )
                try:
                    error_outcome = await self._route_pass(error_event, data, "error")
                except Exception as error_handler_failure:
                    if error_handler_failure is exc:
                        raise
                    _runtime_logger.error(
                        "error handler failed: exception_type=%s",
                        type(error_handler_failure).__name__,
                        exc_info=_runtime_logger.isEnabledFor(logging.DEBUG) or None,
                    )
                    raise error_handler_failure from exc
                if error_outcome.handled:
                    result = error_outcome.result
                else:
                    # An unhandled failure must never be silent: one safe,
                    # structured ERROR per delivery (no payload, no message).
                    _log_unhandled_failure(event, exc)
                    raise
        finally:
            try:
                await _maybe_hook(
                    self._observability_hooks,
                    "after_event",
                    event,
                    data,
                    result,
                    error,
                )
            finally:
                _reset_current_bot(bot_token)
        return result

    async def _route_event(
        self, event: Event, data: dict[str, object]
    ) -> _DispatchOutcome:
        for observer_name in self._specific_observer_names(event):
            outcome = await self._route_pass(event, data, observer_name)
            if outcome.handled or outcome.stopped:
                return outcome
        if not isinstance(event, ErrorEvent):
            outcome = await self._route_pass(event, data, "event")
            if not outcome.handled and not outcome.stopped:
                self._log_unmatched(event)
            return outcome
        return _DispatchOutcome()

    def _log_unmatched(self, event: Event) -> None:
        """Interactive events without a handler must not pass silently."""
        if isinstance(event, ActionEvent):
            fields = [f"event_type={event.event_type}", f"action={event.function_name}"]
            message_name = getattr(event.message, "name", None)
            if message_name:
                fields.append(f"message={message_name}")
            _routing_logger.debug(
                "registered actions: %s", self._registered_action_names()
            )
            _routing_logger.warning("unhandled interaction: %s", " ".join(fields))
            if self._strict_interactions:
                raise UnhandledInteractionError(
                    f"Unhandled interactive event: {' '.join(fields)}"
                )
            return
        if isinstance(event, CommandEvent):
            fields = [
                f"event_type={event.event_type}",
                f"command_id={event.command_id}",
            ]
            _routing_logger.warning("unhandled command: %s", " ".join(fields))
            if self._strict_interactions:
                raise UnhandledInteractionError(
                    f"Unhandled interactive event: {' '.join(fields)}"
                )
            return
        _routing_logger.debug("no handler matched: %s", _event_context(event))

    def _registered_action_names(self) -> str:
        names: list[str] = []
        for router, _ in self._walk():
            for name in router.action.registered_names:
                if name not in names:
                    names.append(name)
        return ", ".join(names) if names else "-"

    async def _route_pass(
        self,
        event: Event,
        data: dict[str, object],
        observer_name: str,
    ) -> _DispatchOutcome:
        for router_index, (router, middleware) in enumerate(self._walk()):
            observer = self._observer(router, observer_name)
            handlers = observer.handlers
            if not handlers:
                continue
            _routing_logger.debug(
                "observer selected: router=%s observer=%s router_path=router[%d]",
                router.name,
                observer_name,
                router_index,
            )
            observer_data = dict(data)
            try:
                matches = await evaluate_filters(observer.filters, event, observer_data)
            except SkipHandler:
                continue
            except StopPropagation:
                return _DispatchOutcome(stopped=True)
            except Exception as exc:
                _attach_failure_context(exc, stage="filter")
                raise
            if not matches:
                continue
            for handler in handlers:
                candidate_data = dict(observer_data)
                try:
                    try:
                        matches = await evaluate_filters(
                            handler.filters, event, candidate_data
                        )
                    except Exception as exc:
                        _attach_failure_context(
                            exc,
                            stage="filter",
                            handler=handler_qualified_name(handler.callback),
                        )
                        raise
                    if not matches:
                        continue
                    # `handler_started` must be visible BEFORE the callback
                    # runs: a hung handler must not look like silence.
                    _routing_logger.debug(
                        "handler_selected: router=%s observer=%s handler=%s "
                        "router_path=router[%d]",
                        router.name,
                        observer_name,
                        handler_qualified_name(handler.callback),
                        router_index,
                    )
                    _routing_logger.info(
                        "handler_started: handler=%s",
                        handler_qualified_name(handler.callback),
                    )
                    await _maybe_hook(
                        self._observability_hooks,
                        "before_handler",
                        event,
                        candidate_data,
                        handler_qualified_name(handler.callback),
                    )
                    started_at = time.perf_counter()
                    result = await self._invoke(
                        handler, middleware, event, candidate_data
                    )
                    duration_ms = (time.perf_counter() - started_at) * 1000
                    slow_handler_ms = self._runtime_diagnostics.slow_handler_ms
                    if slow_handler_ms is not None and duration_ms > slow_handler_ms:
                        _runtime_logger.warning(
                            "slow handler: handler=%s duration_ms=%.1f",
                            handler_qualified_name(handler.callback),
                            duration_ms,
                        )
                except SkipHandler:
                    continue
                except StopPropagation:
                    return _DispatchOutcome(stopped=True)
                except Exception as exc:
                    if _failure_stage(exc) == "unknown":
                        # handler/DI exceptions are tagged inside _invoke;
                        # anything untagged escaped a middleware frame.
                        _attach_failure_context(
                            exc,
                            stage="middleware",
                            handler=handler_qualified_name(handler.callback),
                        )
                    raise
                _routing_logger.info(
                    "handler_completed: handler=%s result_type=%s",
                    handler_qualified_name(handler.callback),
                    type(result).__name__,
                )
                await _maybe_hook(
                    self._observability_hooks,
                    "after_handler",
                    event,
                    candidate_data,
                    handler_qualified_name(handler.callback),
                    result,
                )
                return _DispatchOutcome(handled=True, result=result)
        return _DispatchOutcome()

    @staticmethod
    def _observer(router: Router, name: str) -> EventObserver:
        observer = getattr(router, name)
        if not isinstance(observer, EventObserver):
            raise TypeError(f"Router attribute {name!r} is not an EventObserver")
        return observer

    def _specific_observer_names(self, event: Event) -> tuple[str, ...]:
        if isinstance(event, MessageEvent):
            return ("message",)
        if isinstance(event, ActionEvent):
            if event.dialog is not None and (
                event.dialog.type == DialogEventType.SUBMIT_DIALOG
            ):
                return ("dialog_submit",)
            if event.dialog is not None and (
                event.dialog.type == DialogEventType.CANCEL_DIALOG
            ):
                return ("dialog_cancel",)
            return ("action",)
        if isinstance(event, CommandEvent):
            if event.kind is CommandKind.MESSAGE_ACTION:
                return ("message_action", "command")
            if event.kind is CommandKind.SLASH_COMMAND:
                return ("slash_command", "command")
            if event.kind is CommandKind.QUICK_COMMAND:
                return ("quick_command", "command")
            return ()
        if isinstance(event, AddedToSpaceEvent):
            return ("added_to_space",)
        if isinstance(event, RemovedFromSpaceEvent):
            return ("removed_from_space",)
        if isinstance(event, WidgetUpdatedEvent):
            return ("widget_updated",)
        if isinstance(event, AppHomeEvent):
            return ("app_home",)
        if isinstance(event, FormSubmitEvent):
            return ("form_submit",)
        if isinstance(event, UnknownEvent):
            return ("unknown_event",)
        if isinstance(event, ErrorEvent):
            return ("error",)
        return ()

    @staticmethod
    async def _invoke(
        handler: HandlerObject,
        middleware: tuple[MiddlewareLike, ...],
        event: Event,
        data: MutableMapping[str, object],
    ) -> object:
        async def resolved_handler(
            resolved_event: Event, resolved_data: MutableMapping[str, object]
        ) -> object:
            try:
                return await handler.plan.invoke(resolved_event, resolved_data)
            except DependencyResolutionError as exc:
                _attach_failure_context(
                    exc,
                    stage="dependency_resolution",
                    handler=handler_qualified_name(handler.callback),
                )
                raise
            except Exception as exc:
                _attach_failure_context(
                    exc,
                    stage="handler",
                    handler=handler_qualified_name(handler.callback),
                )
                raise

        next_handler: NextHandler = resolved_handler
        return await invoke_with_middleware(next_handler, middleware, event, data)

preview_capabilities property

The immutable Developer Preview enrollment for typed routing.

feed_update(event, **context) async

Route one domain event and return the handler result unchanged.

Source code in src/chattice/dispatcher/dispatcher.py

async def feed_update(self, event: Event, **context: object) -> object:
    """Route one domain event and return the handler result unchanged."""
    if not isinstance(event, Event):
        raise TypeError("feed_update() accepts chattice Event instances only")
    _routing_logger.debug("event received: %s", _event_context(event))
    _routing_logger.debug("routing started: %s", _event_context(event))
    data = dict(context)
    if "bot" not in data and self._bot is not None:
        data["bot"] = self._bot
    contextual_bot = data.get("bot")
    # Configuration, not caller context, is the source of truth: a
    # feed_update() kwarg must not bypass explicit preview enrollment.
    data["preview_capabilities"] = self._preview_capabilities
    if self._fsm_storage is not None:
        data["state"] = FSMContext(
            self._fsm_storage, StorageKey.build(event, self._fsm_strategy)
        )
    result: object = None
    error: BaseException | None = None
    bot_token = _set_current_bot(contextual_bot)
    try:
        await _maybe_hook(self._observability_hooks, "before_event", event, data)
        try:
            outcome = await self._route_event(event, data)
            if outcome.handled:
                result = outcome.result
        except BaseException as exc:
            error = exc
            if not isinstance(exc, Exception):
                # CancelledError and friends bypass error routing but are
                # still reported to the after_event hook, then re-raised.
                raise
            error_event = ErrorEvent(
                source_event=event,
                exception=exc,
                raw=event.raw,
            )
            try:
                error_outcome = await self._route_pass(error_event, data, "error")
            except Exception as error_handler_failure:
                if error_handler_failure is exc:
                    raise
                _runtime_logger.error(
                    "error handler failed: exception_type=%s",
                    type(error_handler_failure).__name__,
                    exc_info=_runtime_logger.isEnabledFor(logging.DEBUG) or None,
                )
                raise error_handler_failure from exc
            if error_outcome.handled:
                result = error_outcome.result
            else:
                # An unhandled failure must never be silent: one safe,
                # structured ERROR per delivery (no payload, no message).
                _log_unhandled_failure(event, exc)
                raise
    finally:
        try:
            await _maybe_hook(
                self._observability_hooks,
                "after_event",
                event,
                data,
                result,
                error,
            )
        finally:
            _reset_current_bot(bot_token)
    return result

lifespan(*resources)

An async context manager starting resources in order and closing them in reverse (partial-start rollback included). Plug it into FastAPI via app.router.lifespan_context = dispatcher.lifespan(...).

Source code in src/chattice/dispatcher/dispatcher.py

def lifespan(self, *resources: LifespanResource) -> Lifespan:
    """An async context manager starting resources in order and closing
    them in reverse (partial-start rollback included). Plug it into
    FastAPI via ``app.router.lifespan_context = dispatcher.lifespan(...)``.
    """
    return Lifespan(*resources)

run_pubsub(subscription, *, bot=None, credentials=None, credentials_provider=None, max_concurrency=10, max_outstanding_messages=100, idempotency_storage=None, max_delivery_attempts=5, stop_event=None) async

Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

Runs every delivery through THIS dispatcher's router/filter/ middleware/DI pipeline. Handler answers go outbound through bot where semantics allow (text -> send_message, Card -> update_message/send_message); Dialog answers are rejected with CapabilityNotSupported (dialogs require the synchronous HTTP transport). Requires the chattice[pubsub] extra.

Blocks until stop_event fires or SIGINT/SIGTERM; drains in-flight handlers before returning.

Source code in src/chattice/dispatcher/dispatcher.py

async def run_pubsub(
    self,
    subscription: str,
    *,
    bot: Bot | None = None,
    credentials: Credentials | None = None,
    credentials_provider: CredentialsProvider | None = None,
    max_concurrency: int = 10,
    max_outstanding_messages: int = 100,
    idempotency_storage: IdempotencyStorage | None = None,
    max_delivery_attempts: int = 5,
    stop_event: asyncio.Event | None = None,
) -> None:
    """Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

    Runs every delivery through THIS dispatcher's router/filter/
    middleware/DI pipeline. Handler answers go outbound through
    ``bot`` where semantics allow (text -> send_message, Card ->
    update_message/send_message); Dialog answers are rejected with
    ``CapabilityNotSupported`` (dialogs require the synchronous HTTP
    transport). Requires the ``chattice[pubsub]`` extra.

    Blocks until ``stop_event`` fires or SIGINT/SIGTERM; drains
    in-flight handlers before returning.
    """
    from chattice.transports.pubsub_runner import PubSubPullRunner

    runner = PubSubPullRunner(
        self,
        subscription,
        bot=bot,
        credentials=credentials,
        credentials_provider=credentials_provider,
        max_concurrency=max_concurrency,
        max_outstanding_messages=max_outstanding_messages,
        idempotency_storage=idempotency_storage,
        max_delivery_attempts=max_delivery_attempts,
        delayed_event_ms=self._runtime_diagnostics.delayed_event_ms,
        observability_hooks=self._observability_hooks,
    )
    await runner.run(stop_event=stop_event)

Router

A named collection of observers, middleware, and child routers.

Source code in src/chattice/dispatcher/router.py

class Router:
    """A named collection of observers, middleware, and child routers."""

    def __init__(self, *, name: str | None = None) -> None:
        self.name = name or "router"
        if not self.name.strip():
            raise ValueError("Router name cannot be empty")
        self.message = EventObserver("message")
        self.action = EventObserver("action", action_shortcut=True)
        self.command = EventObserver("command")
        self.slash_command = EventObserver("slash_command")
        self.quick_command = EventObserver("quick_command")
        self.message_action = EventObserver("message_action")
        self.added_to_space = EventObserver("added_to_space")
        self.removed_from_space = EventObserver("removed_from_space")
        self.widget_updated = EventObserver("widget_updated")
        self.app_home = EventObserver("app_home")
        self.form_submit = EventObserver("form_submit")
        self.dialog_submit = EventObserver("dialog_submit")
        self.dialog_cancel = EventObserver("dialog_cancel")
        self.event = EventObserver("event")
        self.unknown_event = EventObserver("unknown_event")
        self.error = EventObserver("error")
        self.middleware = MiddlewareManager()
        self._parent: Router | None = None
        self._children: list[Router] = []
        self._is_dispatcher = False

    @property
    def parent(self) -> Router | None:
        """The owning parent, or ``None`` for a detached/root router."""
        return self._parent

    @property
    def children(self) -> tuple[Router, ...]:
        """Child routers in deterministic inclusion order."""
        return tuple(self._children)

    def include_router(self, router: Router) -> Router:
        """Attach one detached router as a child."""
        if not isinstance(router, Router):
            raise TypeError("include_router() requires a Router")
        if router is self:
            raise RouterConfigurationError("A router cannot include itself")
        if router._is_dispatcher:
            raise RouterConfigurationError("A Dispatcher cannot be attached as a child")
        if router._parent is not None:
            raise RouterConfigurationError(
                f"Router {router.name!r} is already attached to {router._parent.name!r}"
            )
        if router._contains(self):
            raise RouterConfigurationError(
                f"Including {router.name!r} in {self.name!r} would create a cycle"
            )
        router._parent = self
        if any(child.name == router.name for child in self._children):
            logger.warning("duplicate router name: %s", router.name)
        self._children.append(router)
        return router

    def _contains(self, candidate: Router) -> bool:
        if self is candidate:
            return True
        return any(child._contains(candidate) for child in self._children)

    def _walk(
        self, inherited: tuple[MiddlewareLike, ...] = ()
    ) -> Iterator[tuple[Router, tuple[MiddlewareLike, ...]]]:
        middleware = (*inherited, *tuple(self.middleware))
        yield self, middleware
        for child in tuple(self._children):
            yield from child._walk(middleware)

    def __repr__(self) -> str:
        return (
            f"Router(name={self.name!r}, children={len(self._children)}, "
            f"parent={self._parent.name if self._parent else None!r})"
        )

children property

Child routers in deterministic inclusion order.

parent property

The owning parent, or None for a detached/root router.

include_router(router)

Attach one detached router as a child.

Source code in src/chattice/dispatcher/router.py

def include_router(self, router: Router) -> Router:
    """Attach one detached router as a child."""
    if not isinstance(router, Router):
        raise TypeError("include_router() requires a Router")
    if router is self:
        raise RouterConfigurationError("A router cannot include itself")
    if router._is_dispatcher:
        raise RouterConfigurationError("A Dispatcher cannot be attached as a child")
    if router._parent is not None:
        raise RouterConfigurationError(
            f"Router {router.name!r} is already attached to {router._parent.name!r}"
        )
    if router._contains(self):
        raise RouterConfigurationError(
            f"Including {router.name!r} in {self.name!r} would create a cycle"
        )
    router._parent = self
    if any(child.name == router.name for child in self._children):
        logger.warning("duplicate router name: %s", router.name)
    self._children.append(router)
    return router

Public dispatcher, router, and observer API.

Dispatcher

Bases: Router

Root router and transport-independent event feed.

Source code in src/chattice/dispatcher/dispatcher.py

class Dispatcher(Router):
    """Root router and transport-independent event feed."""

    def lifespan(self, *resources: LifespanResource) -> Lifespan:
        """An async context manager starting resources in order and closing
        them in reverse (partial-start rollback included). Plug it into
        FastAPI via ``app.router.lifespan_context = dispatcher.lifespan(...)``.
        """
        return Lifespan(*resources)

    async def run_pubsub(
        self,
        subscription: str,
        *,
        bot: Bot | None = None,
        credentials: Credentials | None = None,
        credentials_provider: CredentialsProvider | None = None,
        max_concurrency: int = 10,
        max_outstanding_messages: int = 100,
        idempotency_storage: IdempotencyStorage | None = None,
        max_delivery_attempts: int = 5,
        stop_event: asyncio.Event | None = None,
    ) -> None:
        """Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

        Runs every delivery through THIS dispatcher's router/filter/
        middleware/DI pipeline. Handler answers go outbound through
        ``bot`` where semantics allow (text -> send_message, Card ->
        update_message/send_message); Dialog answers are rejected with
        ``CapabilityNotSupported`` (dialogs require the synchronous HTTP
        transport). Requires the ``chattice[pubsub]`` extra.

        Blocks until ``stop_event`` fires or SIGINT/SIGTERM; drains
        in-flight handlers before returning.
        """
        from chattice.transports.pubsub_runner import PubSubPullRunner

        runner = PubSubPullRunner(
            self,
            subscription,
            bot=bot,
            credentials=credentials,
            credentials_provider=credentials_provider,
            max_concurrency=max_concurrency,
            max_outstanding_messages=max_outstanding_messages,
            idempotency_storage=idempotency_storage,
            max_delivery_attempts=max_delivery_attempts,
            delayed_event_ms=self._runtime_diagnostics.delayed_event_ms,
            observability_hooks=self._observability_hooks,
        )
        await runner.run(stop_event=stop_event)

    def __init__(
        self,
        *,
        name: str = "dispatcher",
        bot: object | None = None,
        fsm_storage: BaseStorage | None = None,
        fsm_strategy: FSMStrategy = FSMStrategy.USER_IN_SPACE,
        # ``object``: implementations may provide ANY subset of the
        # optional ObservabilityHooks surface.
        observability_hooks: object | None = None,
        preview_features: Iterable[PreviewFeature] = (),
        strict_interactions: bool = False,
        runtime_diagnostics: RuntimeDiagnostics | None = None,
    ) -> None:
        super().__init__(name=name)
        self._is_dispatcher = True
        self._bot = bot
        self._fsm_storage = fsm_storage
        self._fsm_strategy = fsm_strategy
        self._observability_hooks = observability_hooks
        self._preview_capabilities = PreviewCapabilities(preview_features)
        self._strict_interactions = strict_interactions
        self._runtime_diagnostics = runtime_diagnostics or RuntimeDiagnostics()

    @property
    def preview_capabilities(self) -> PreviewCapabilities:
        """The immutable Developer Preview enrollment for typed routing."""
        return self._preview_capabilities

    async def feed_update(self, event: Event, **context: object) -> object:
        """Route one domain event and return the handler result unchanged."""
        if not isinstance(event, Event):
            raise TypeError("feed_update() accepts chattice Event instances only")
        _routing_logger.debug("event received: %s", _event_context(event))
        _routing_logger.debug("routing started: %s", _event_context(event))
        data = dict(context)
        if "bot" not in data and self._bot is not None:
            data["bot"] = self._bot
        contextual_bot = data.get("bot")
        # Configuration, not caller context, is the source of truth: a
        # feed_update() kwarg must not bypass explicit preview enrollment.
        data["preview_capabilities"] = self._preview_capabilities
        if self._fsm_storage is not None:
            data["state"] = FSMContext(
                self._fsm_storage, StorageKey.build(event, self._fsm_strategy)
            )
        result: object = None
        error: BaseException | None = None
        bot_token = _set_current_bot(contextual_bot)
        try:
            await _maybe_hook(self._observability_hooks, "before_event", event, data)
            try:
                outcome = await self._route_event(event, data)
                if outcome.handled:
                    result = outcome.result
            except BaseException as exc:
                error = exc
                if not isinstance(exc, Exception):
                    # CancelledError and friends bypass error routing but are
                    # still reported to the after_event hook, then re-raised.
                    raise
                error_event = ErrorEvent(
                    source_event=event,
                    exception=exc,
                    raw=event.raw,
                )
                try:
                    error_outcome = await self._route_pass(error_event, data, "error")
                except Exception as error_handler_failure:
                    if error_handler_failure is exc:
                        raise
                    _runtime_logger.error(
                        "error handler failed: exception_type=%s",
                        type(error_handler_failure).__name__,
                        exc_info=_runtime_logger.isEnabledFor(logging.DEBUG) or None,
                    )
                    raise error_handler_failure from exc
                if error_outcome.handled:
                    result = error_outcome.result
                else:
                    # An unhandled failure must never be silent: one safe,
                    # structured ERROR per delivery (no payload, no message).
                    _log_unhandled_failure(event, exc)
                    raise
        finally:
            try:
                await _maybe_hook(
                    self._observability_hooks,
                    "after_event",
                    event,
                    data,
                    result,
                    error,
                )
            finally:
                _reset_current_bot(bot_token)
        return result

    async def _route_event(
        self, event: Event, data: dict[str, object]
    ) -> _DispatchOutcome:
        for observer_name in self._specific_observer_names(event):
            outcome = await self._route_pass(event, data, observer_name)
            if outcome.handled or outcome.stopped:
                return outcome
        if not isinstance(event, ErrorEvent):
            outcome = await self._route_pass(event, data, "event")
            if not outcome.handled and not outcome.stopped:
                self._log_unmatched(event)
            return outcome
        return _DispatchOutcome()

    def _log_unmatched(self, event: Event) -> None:
        """Interactive events without a handler must not pass silently."""
        if isinstance(event, ActionEvent):
            fields = [f"event_type={event.event_type}", f"action={event.function_name}"]
            message_name = getattr(event.message, "name", None)
            if message_name:
                fields.append(f"message={message_name}")
            _routing_logger.debug(
                "registered actions: %s", self._registered_action_names()
            )
            _routing_logger.warning("unhandled interaction: %s", " ".join(fields))
            if self._strict_interactions:
                raise UnhandledInteractionError(
                    f"Unhandled interactive event: {' '.join(fields)}"
                )
            return
        if isinstance(event, CommandEvent):
            fields = [
                f"event_type={event.event_type}",
                f"command_id={event.command_id}",
            ]
            _routing_logger.warning("unhandled command: %s", " ".join(fields))
            if self._strict_interactions:
                raise UnhandledInteractionError(
                    f"Unhandled interactive event: {' '.join(fields)}"
                )
            return
        _routing_logger.debug("no handler matched: %s", _event_context(event))

    def _registered_action_names(self) -> str:
        names: list[str] = []
        for router, _ in self._walk():
            for name in router.action.registered_names:
                if name not in names:
                    names.append(name)
        return ", ".join(names) if names else "-"

    async def _route_pass(
        self,
        event: Event,
        data: dict[str, object],
        observer_name: str,
    ) -> _DispatchOutcome:
        for router_index, (router, middleware) in enumerate(self._walk()):
            observer = self._observer(router, observer_name)
            handlers = observer.handlers
            if not handlers:
                continue
            _routing_logger.debug(
                "observer selected: router=%s observer=%s router_path=router[%d]",
                router.name,
                observer_name,
                router_index,
            )
            observer_data = dict(data)
            try:
                matches = await evaluate_filters(observer.filters, event, observer_data)
            except SkipHandler:
                continue
            except StopPropagation:
                return _DispatchOutcome(stopped=True)
            except Exception as exc:
                _attach_failure_context(exc, stage="filter")
                raise
            if not matches:
                continue
            for handler in handlers:
                candidate_data = dict(observer_data)
                try:
                    try:
                        matches = await evaluate_filters(
                            handler.filters, event, candidate_data
                        )
                    except Exception as exc:
                        _attach_failure_context(
                            exc,
                            stage="filter",
                            handler=handler_qualified_name(handler.callback),
                        )
                        raise
                    if not matches:
                        continue
                    # `handler_started` must be visible BEFORE the callback
                    # runs: a hung handler must not look like silence.
                    _routing_logger.debug(
                        "handler_selected: router=%s observer=%s handler=%s "
                        "router_path=router[%d]",
                        router.name,
                        observer_name,
                        handler_qualified_name(handler.callback),
                        router_index,
                    )
                    _routing_logger.info(
                        "handler_started: handler=%s",
                        handler_qualified_name(handler.callback),
                    )
                    await _maybe_hook(
                        self._observability_hooks,
                        "before_handler",
                        event,
                        candidate_data,
                        handler_qualified_name(handler.callback),
                    )
                    started_at = time.perf_counter()
                    result = await self._invoke(
                        handler, middleware, event, candidate_data
                    )
                    duration_ms = (time.perf_counter() - started_at) * 1000
                    slow_handler_ms = self._runtime_diagnostics.slow_handler_ms
                    if slow_handler_ms is not None and duration_ms > slow_handler_ms:
                        _runtime_logger.warning(
                            "slow handler: handler=%s duration_ms=%.1f",
                            handler_qualified_name(handler.callback),
                            duration_ms,
                        )
                except SkipHandler:
                    continue
                except StopPropagation:
                    return _DispatchOutcome(stopped=True)
                except Exception as exc:
                    if _failure_stage(exc) == "unknown":
                        # handler/DI exceptions are tagged inside _invoke;
                        # anything untagged escaped a middleware frame.
                        _attach_failure_context(
                            exc,
                            stage="middleware",
                            handler=handler_qualified_name(handler.callback),
                        )
                    raise
                _routing_logger.info(
                    "handler_completed: handler=%s result_type=%s",
                    handler_qualified_name(handler.callback),
                    type(result).__name__,
                )
                await _maybe_hook(
                    self._observability_hooks,
                    "after_handler",
                    event,
                    candidate_data,
                    handler_qualified_name(handler.callback),
                    result,
                )
                return _DispatchOutcome(handled=True, result=result)
        return _DispatchOutcome()

    @staticmethod
    def _observer(router: Router, name: str) -> EventObserver:
        observer = getattr(router, name)
        if not isinstance(observer, EventObserver):
            raise TypeError(f"Router attribute {name!r} is not an EventObserver")
        return observer

    def _specific_observer_names(self, event: Event) -> tuple[str, ...]:
        if isinstance(event, MessageEvent):
            return ("message",)
        if isinstance(event, ActionEvent):
            if event.dialog is not None and (
                event.dialog.type == DialogEventType.SUBMIT_DIALOG
            ):
                return ("dialog_submit",)
            if event.dialog is not None and (
                event.dialog.type == DialogEventType.CANCEL_DIALOG
            ):
                return ("dialog_cancel",)
            return ("action",)
        if isinstance(event, CommandEvent):
            if event.kind is CommandKind.MESSAGE_ACTION:
                return ("message_action", "command")
            if event.kind is CommandKind.SLASH_COMMAND:
                return ("slash_command", "command")
            if event.kind is CommandKind.QUICK_COMMAND:
                return ("quick_command", "command")
            return ()
        if isinstance(event, AddedToSpaceEvent):
            return ("added_to_space",)
        if isinstance(event, RemovedFromSpaceEvent):
            return ("removed_from_space",)
        if isinstance(event, WidgetUpdatedEvent):
            return ("widget_updated",)
        if isinstance(event, AppHomeEvent):
            return ("app_home",)
        if isinstance(event, FormSubmitEvent):
            return ("form_submit",)
        if isinstance(event, UnknownEvent):
            return ("unknown_event",)
        if isinstance(event, ErrorEvent):
            return ("error",)
        return ()

    @staticmethod
    async def _invoke(
        handler: HandlerObject,
        middleware: tuple[MiddlewareLike, ...],
        event: Event,
        data: MutableMapping[str, object],
    ) -> object:
        async def resolved_handler(
            resolved_event: Event, resolved_data: MutableMapping[str, object]
        ) -> object:
            try:
                return await handler.plan.invoke(resolved_event, resolved_data)
            except DependencyResolutionError as exc:
                _attach_failure_context(
                    exc,
                    stage="dependency_resolution",
                    handler=handler_qualified_name(handler.callback),
                )
                raise
            except Exception as exc:
                _attach_failure_context(
                    exc,
                    stage="handler",
                    handler=handler_qualified_name(handler.callback),
                )
                raise

        next_handler: NextHandler = resolved_handler
        return await invoke_with_middleware(next_handler, middleware, event, data)

preview_capabilities property

The immutable Developer Preview enrollment for typed routing.

feed_update(event, **context) async

Route one domain event and return the handler result unchanged.

Source code in src/chattice/dispatcher/dispatcher.py

async def feed_update(self, event: Event, **context: object) -> object:
    """Route one domain event and return the handler result unchanged."""
    if not isinstance(event, Event):
        raise TypeError("feed_update() accepts chattice Event instances only")
    _routing_logger.debug("event received: %s", _event_context(event))
    _routing_logger.debug("routing started: %s", _event_context(event))
    data = dict(context)
    if "bot" not in data and self._bot is not None:
        data["bot"] = self._bot
    contextual_bot = data.get("bot")
    # Configuration, not caller context, is the source of truth: a
    # feed_update() kwarg must not bypass explicit preview enrollment.
    data["preview_capabilities"] = self._preview_capabilities
    if self._fsm_storage is not None:
        data["state"] = FSMContext(
            self._fsm_storage, StorageKey.build(event, self._fsm_strategy)
        )
    result: object = None
    error: BaseException | None = None
    bot_token = _set_current_bot(contextual_bot)
    try:
        await _maybe_hook(self._observability_hooks, "before_event", event, data)
        try:
            outcome = await self._route_event(event, data)
            if outcome.handled:
                result = outcome.result
        except BaseException as exc:
            error = exc
            if not isinstance(exc, Exception):
                # CancelledError and friends bypass error routing but are
                # still reported to the after_event hook, then re-raised.
                raise
            error_event = ErrorEvent(
                source_event=event,
                exception=exc,
                raw=event.raw,
            )
            try:
                error_outcome = await self._route_pass(error_event, data, "error")
            except Exception as error_handler_failure:
                if error_handler_failure is exc:
                    raise
                _runtime_logger.error(
                    "error handler failed: exception_type=%s",
                    type(error_handler_failure).__name__,
                    exc_info=_runtime_logger.isEnabledFor(logging.DEBUG) or None,
                )
                raise error_handler_failure from exc
            if error_outcome.handled:
                result = error_outcome.result
            else:
                # An unhandled failure must never be silent: one safe,
                # structured ERROR per delivery (no payload, no message).
                _log_unhandled_failure(event, exc)
                raise
    finally:
        try:
            await _maybe_hook(
                self._observability_hooks,
                "after_event",
                event,
                data,
                result,
                error,
            )
        finally:
            _reset_current_bot(bot_token)
    return result

lifespan(*resources)

An async context manager starting resources in order and closing them in reverse (partial-start rollback included). Plug it into FastAPI via app.router.lifespan_context = dispatcher.lifespan(...).

Source code in src/chattice/dispatcher/dispatcher.py

def lifespan(self, *resources: LifespanResource) -> Lifespan:
    """An async context manager starting resources in order and closing
    them in reverse (partial-start rollback included). Plug it into
    FastAPI via ``app.router.lifespan_context = dispatcher.lifespan(...)``.
    """
    return Lifespan(*resources)

run_pubsub(subscription, *, bot=None, credentials=None, credentials_provider=None, max_concurrency=10, max_outstanding_messages=100, idempotency_storage=None, max_delivery_attempts=5, stop_event=None) async

Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

Runs every delivery through THIS dispatcher's router/filter/ middleware/DI pipeline. Handler answers go outbound through bot where semantics allow (text -> send_message, Card -> update_message/send_message); Dialog answers are rejected with CapabilityNotSupported (dialogs require the synchronous HTTP transport). Requires the chattice[pubsub] extra.

Blocks until stop_event fires or SIGINT/SIGTERM; drains in-flight handlers before returning.

Source code in src/chattice/dispatcher/dispatcher.py

async def run_pubsub(
    self,
    subscription: str,
    *,
    bot: Bot | None = None,
    credentials: Credentials | None = None,
    credentials_provider: CredentialsProvider | None = None,
    max_concurrency: int = 10,
    max_outstanding_messages: int = 100,
    idempotency_storage: IdempotencyStorage | None = None,
    max_delivery_attempts: int = 5,
    stop_event: asyncio.Event | None = None,
) -> None:
    """Streaming-pull Pub/Sub ingress: the long-lived subscriber mode.

    Runs every delivery through THIS dispatcher's router/filter/
    middleware/DI pipeline. Handler answers go outbound through
    ``bot`` where semantics allow (text -> send_message, Card ->
    update_message/send_message); Dialog answers are rejected with
    ``CapabilityNotSupported`` (dialogs require the synchronous HTTP
    transport). Requires the ``chattice[pubsub]`` extra.

    Blocks until ``stop_event`` fires or SIGINT/SIGTERM; drains
    in-flight handlers before returning.
    """
    from chattice.transports.pubsub_runner import PubSubPullRunner

    runner = PubSubPullRunner(
        self,
        subscription,
        bot=bot,
        credentials=credentials,
        credentials_provider=credentials_provider,
        max_concurrency=max_concurrency,
        max_outstanding_messages=max_outstanding_messages,
        idempotency_storage=idempotency_storage,
        max_delivery_attempts=max_delivery_attempts,
        delayed_event_ms=self._runtime_diagnostics.delayed_event_ms,
        observability_hooks=self._observability_hooks,
    )
    await runner.run(stop_event=stop_event)

EventObserver

An ordered collection of handlers for one event category.

Source code in src/chattice/dispatcher/observer.py

class EventObserver:
    """An ordered collection of handlers for one event category."""

    def __init__(
        self,
        name: str,
        *,
        action_shortcut: bool = False,
        cloud_type_shortcut: bool = False,
    ) -> None:
        self.name = name
        self._action_shortcut = action_shortcut
        self._cloud_type_shortcut = cloud_type_shortcut
        self._filters: list[FilterLike] = []
        self._handlers: list[HandlerObject] = []
        self._shortcut_names: list[str] = []

    @property
    def filters(self) -> tuple[FilterLike, ...]:
        """Return the common filters applied before handler filters."""
        return tuple(self._filters)

    @property
    def handlers(self) -> tuple[HandlerObject, ...]:
        """Return a stable registration snapshot."""
        return tuple(self._handlers)

    @property
    def registered_names(self) -> tuple[str, ...]:
        """Shortcut names registered on this observer (action names etc.)."""
        return tuple(self._shortcut_names)

    def register(
        self, callback: HandlerCallback, *filters: FilterLike | str
    ) -> HandlerCallback:
        """Register and return ``callback`` for programmatic use."""
        normalized = tuple(self._normalize_filter(filter_) for filter_ in filters)
        self._handlers.append(HandlerObject(callback=callback, filters=normalized))
        return callback

    def filter(self, *filters: FilterLike | str) -> None:
        """Append common filters for every handler on this observer."""
        self._filters.extend(self._normalize_filter(filter_) for filter_ in filters)

    def __call__(
        self, *filters: FilterLike | str
    ) -> Callable[[HandlerCallback], HandlerCallback]:
        """Create a handler-registration decorator."""

        def decorator(callback: HandlerCallback) -> HandlerCallback:
            return self.register(callback, *filters)

        return decorator

    def _normalize_filter(self, filter_: FilterLike | str) -> FilterLike:
        if isinstance(filter_, str):
            if self._action_shortcut:
                self._shortcut_names.append(filter_)
                return F.name == filter_
            if self._cloud_type_shortcut:
                return F.cloud_type == filter_
            raise TypeError(
                f"String filter shortcuts are not supported by {self.name!r}"
            )
        if not callable(filter_):
            raise TypeError(
                "Filters must be asynchronous callables or magic expressions"
            )
        return filter_

    def __len__(self) -> int:
        return len(self._handlers)

    def __repr__(self) -> str:
        return f"EventObserver(name={self.name!r}, handlers={len(self)})"

filters property

Return the common filters applied before handler filters.

handlers property

Return a stable registration snapshot.

registered_names property

Shortcut names registered on this observer (action names etc.).

__call__(*filters)

Create a handler-registration decorator.

Source code in src/chattice/dispatcher/observer.py

def __call__(
    self, *filters: FilterLike | str
) -> Callable[[HandlerCallback], HandlerCallback]:
    """Create a handler-registration decorator."""

    def decorator(callback: HandlerCallback) -> HandlerCallback:
        return self.register(callback, *filters)

    return decorator

filter(*filters)

Append common filters for every handler on this observer.

Source code in src/chattice/dispatcher/observer.py

def filter(self, *filters: FilterLike | str) -> None:
    """Append common filters for every handler on this observer."""
    self._filters.extend(self._normalize_filter(filter_) for filter_ in filters)

register(callback, *filters)

Register and return callback for programmatic use.

Source code in src/chattice/dispatcher/observer.py

def register(
    self, callback: HandlerCallback, *filters: FilterLike | str
) -> HandlerCallback:
    """Register and return ``callback`` for programmatic use."""
    normalized = tuple(self._normalize_filter(filter_) for filter_ in filters)
    self._handlers.append(HandlerObject(callback=callback, filters=normalized))
    return callback

Lifespan

Ordered startup / reverse-order shutdown over async resources.

Source code in src/chattice/dispatcher/lifespan.py

class Lifespan:
    """Ordered startup / reverse-order shutdown over async resources."""

    def __init__(self, *resources: LifespanResource) -> None:
        self._resources = tuple(resources)
        self._started: list[LifespanResource] = []
        self._closed = False

    async def __aenter__(self) -> Lifespan:
        if self._closed:
            raise RuntimeError("This lifespan has already been used")
        started = self._started
        try:
            for resource in self._resources:
                await resource.start()
                started.append(resource)
        except BaseException:
            # Partial start: roll back in reverse order, then re-raise.
            for resource in reversed(started):
                try:
                    await resource.close()
                except BaseException as close_error:  # pragma: no cover - logged
                    logger.error(
                        "lifespan rollback failed: resource=%s error=%s",
                        type(resource).__name__,
                        type(close_error).__name__,
                    )
            started.clear()
            raise
        return self

    async def __aexit__(
        self,
        exc_type: object,
        exc_value: object,
        traceback: object,
    ) -> None:
        # Attempt EVERY closer in reverse order; collect the first close
        # failure and surface it only after cleanup completes.
        close_errors: list[BaseException] = []
        for resource in reversed(self._started):
            try:
                await resource.close()
            except BaseException as close_error:
                logger.error(
                    "lifespan shutdown failed: resource=%s error=%s",
                    type(resource).__name__,
                    type(close_error).__name__,
                )
                close_errors.append(close_error)
        self._started.clear()
        self._closed = True
        if close_errors and exc_type is None:
            # A clean exit that hits a close failure must surface.
            raise close_errors[0]

LifespanResource

Bases: Protocol

A resource with ordered async startup and shutdown.

Source code in src/chattice/dispatcher/lifespan.py

class LifespanResource(Protocol):
    """A resource with ordered async startup and shutdown."""

    async def start(self) -> None: ...

    async def close(self) -> None: ...

Router

A named collection of observers, middleware, and child routers.

Source code in src/chattice/dispatcher/router.py

class Router:
    """A named collection of observers, middleware, and child routers."""

    def __init__(self, *, name: str | None = None) -> None:
        self.name = name or "router"
        if not self.name.strip():
            raise ValueError("Router name cannot be empty")
        self.message = EventObserver("message")
        self.action = EventObserver("action", action_shortcut=True)
        self.command = EventObserver("command")
        self.slash_command = EventObserver("slash_command")
        self.quick_command = EventObserver("quick_command")
        self.message_action = EventObserver("message_action")
        self.added_to_space = EventObserver("added_to_space")
        self.removed_from_space = EventObserver("removed_from_space")
        self.widget_updated = EventObserver("widget_updated")
        self.app_home = EventObserver("app_home")
        self.form_submit = EventObserver("form_submit")
        self.dialog_submit = EventObserver("dialog_submit")
        self.dialog_cancel = EventObserver("dialog_cancel")
        self.event = EventObserver("event")
        self.unknown_event = EventObserver("unknown_event")
        self.error = EventObserver("error")
        self.middleware = MiddlewareManager()
        self._parent: Router | None = None
        self._children: list[Router] = []
        self._is_dispatcher = False

    @property
    def parent(self) -> Router | None:
        """The owning parent, or ``None`` for a detached/root router."""
        return self._parent

    @property
    def children(self) -> tuple[Router, ...]:
        """Child routers in deterministic inclusion order."""
        return tuple(self._children)

    def include_router(self, router: Router) -> Router:
        """Attach one detached router as a child."""
        if not isinstance(router, Router):
            raise TypeError("include_router() requires a Router")
        if router is self:
            raise RouterConfigurationError("A router cannot include itself")
        if router._is_dispatcher:
            raise RouterConfigurationError("A Dispatcher cannot be attached as a child")
        if router._parent is not None:
            raise RouterConfigurationError(
                f"Router {router.name!r} is already attached to {router._parent.name!r}"
            )
        if router._contains(self):
            raise RouterConfigurationError(
                f"Including {router.name!r} in {self.name!r} would create a cycle"
            )
        router._parent = self
        if any(child.name == router.name for child in self._children):
            logger.warning("duplicate router name: %s", router.name)
        self._children.append(router)
        return router

    def _contains(self, candidate: Router) -> bool:
        if self is candidate:
            return True
        return any(child._contains(candidate) for child in self._children)

    def _walk(
        self, inherited: tuple[MiddlewareLike, ...] = ()
    ) -> Iterator[tuple[Router, tuple[MiddlewareLike, ...]]]:
        middleware = (*inherited, *tuple(self.middleware))
        yield self, middleware
        for child in tuple(self._children):
            yield from child._walk(middleware)

    def __repr__(self) -> str:
        return (
            f"Router(name={self.name!r}, children={len(self._children)}, "
            f"parent={self._parent.name if self._parent else None!r})"
        )

children property

Child routers in deterministic inclusion order.

parent property

The owning parent, or None for a detached/root router.

include_router(router)

Attach one detached router as a child.

Source code in src/chattice/dispatcher/router.py

def include_router(self, router: Router) -> Router:
    """Attach one detached router as a child."""
    if not isinstance(router, Router):
        raise TypeError("include_router() requires a Router")
    if router is self:
        raise RouterConfigurationError("A router cannot include itself")
    if router._is_dispatcher:
        raise RouterConfigurationError("A Dispatcher cannot be attached as a child")
    if router._parent is not None:
        raise RouterConfigurationError(
            f"Router {router.name!r} is already attached to {router._parent.name!r}"
        )
    if router._contains(self):
        raise RouterConfigurationError(
            f"Including {router.name!r} in {self.name!r} would create a cycle"
        )
    router._parent = self
    if any(child.name == router.name for child in self._children):
        logger.warning("duplicate router name: %s", router.name)
    self._children.append(router)
    return router

Public domain events.

ActionEvent dataclass

Bases: Event

A named action with immutable application parameters.

Source code in src/chattice/events/action.py

@dataclass(frozen=True, slots=True, kw_only=True)
class ActionEvent(Event):
    """A named action with immutable application parameters."""

    event_type: str = field(default="action", init=False)
    name: str = ""
    parameters: Mapping[str, object] = field(default_factory=dict)
    form_inputs: FormInputs = field(default_factory=FormInputs)
    sender_type: str | None = None
    source: ActionSource | None = None
    # Message identity of the clicked card — HTTP responses do not need it
    # (Google knows the target), Pub/Sub answers need it for
    # Bot.update_message.
    message: MessageRef | None = None

    def __post_init__(self) -> None:
        """Take a shallow immutable snapshot of action parameters."""
        object.__setattr__(self, "parameters", MappingProxyType(dict(self.parameters)))

    @property
    def function_name(self) -> str:
        """Google's normalized invoked function with the compatibility ``name``."""
        return self.name

function_name property

Google's normalized invoked function with the compatibility name.

__post_init__()

Take a shallow immutable snapshot of action parameters.

Source code in src/chattice/events/action.py

def __post_init__(self) -> None:
    """Take a shallow immutable snapshot of action parameters."""
    object.__setattr__(self, "parameters", MappingProxyType(dict(self.parameters)))

ActionSource

Bases: StrEnum

The Google Chat surface that produced a card action.

Source code in src/chattice/events/action.py

class ActionSource(StrEnum):
    """The Google Chat surface that produced a card action."""

    MESSAGE = "MESSAGE"
    DIALOG = "DIALOG"
    HOME = "HOME"

AddedToSpaceEvent dataclass

Bases: Event

The Chat app was added to a space.

Source code in src/chattice/events/space.py

@dataclass(frozen=True, slots=True, kw_only=True)
class AddedToSpaceEvent(Event):
    """The Chat app was added to a space."""

    event_type: str = field(default="added_to_space", init=False)

AppHomeEvent dataclass

Bases: Event

A user opened the Chat app's Home tab.

Source code in src/chattice/events/app_home.py

@dataclass(frozen=True, slots=True, kw_only=True)
class AppHomeEvent(Event):
    """A user opened the Chat app's Home tab."""

    event_type: str = field(default="app_home", init=False)

CommandEvent dataclass

Bases: Event

A command identified by configured numeric ID and documented type.

Produced from BOTH wire families: - slash commands arrive as MESSAGE events with message.slashCommand + argumentText (source kind SLASH_COMMAND); - quick commands / message actions arrive as APP_COMMAND events with appCommandMetadata (source kind QUICK_COMMAND; MESSAGE_ACTION for message actions).

Source code in src/chattice/events/command.py

@dataclass(frozen=True, slots=True, kw_only=True)
class CommandEvent(Event):
    """A command identified by configured numeric ID and documented type.

    Produced from BOTH wire families:
    - slash commands arrive as ``MESSAGE`` events with
      ``message.slashCommand`` + ``argumentText`` (source kind
      ``SLASH_COMMAND``);
    - quick commands / message actions arrive as ``APP_COMMAND`` events
      with ``appCommandMetadata`` (source kind ``QUICK_COMMAND``;
      ``MESSAGE_ACTION`` for message actions).
    """

    event_type: str = field(default="command", init=False)
    command_id: int
    command_type: str | None = None
    kind: CommandKind | None = None
    # Compatibility field retained for callers that inspect the source. New code should
    # compare the typed ``kind`` value.
    source_kind: str | None = None
    message_text: str | None = None
    target_message: MessageRef | None = None

    def __post_init__(self) -> None:
        kind = self.kind
        source_kind = self.source_kind
        source_kind_value: CommandKind | None = None
        if source_kind is not None:
            try:
                source_kind_value = CommandKind(source_kind)
            except ValueError:
                source_kind_value = None
        if (
            kind is not None
            and source_kind_value is not None
            and kind != source_kind_value
        ):
            raise ValueError("kind and source_kind describe different command families")
        if source_kind is None and kind is not None:
            source_kind = kind.value
        object.__setattr__(self, "kind", kind)
        object.__setattr__(self, "source_kind", source_kind)

CommandKind

Bases: StrEnum

Google-native command families, independent of their wire envelope.

Source code in src/chattice/events/command.py

class CommandKind(StrEnum):
    """Google-native command families, independent of their wire envelope."""

    SLASH_COMMAND = "SLASH_COMMAND"
    QUICK_COMMAND = "QUICK_COMMAND"
    MESSAGE_ACTION = "MESSAGE_ACTION"

DateInput dataclass

A date represented by Google's lossless epoch-millisecond value.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class DateInput:
    """A date represented by Google's lossless epoch-millisecond value."""

    ms_since_epoch: int

DateTimeInput dataclass

A date/time input with the documented component-presence flags.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class DateTimeInput:
    """A date/time input with the documented component-presence flags."""

    ms_since_epoch: int
    has_date: bool | None = None
    has_time: bool | None = None

DialogEventType

Bases: StrEnum

Stable documented Google Chat dialog interaction types.

Source code in src/chattice/events/common.py

class DialogEventType(StrEnum):
    """Stable documented Google Chat dialog interaction types."""

    REQUEST_DIALOG = "REQUEST_DIALOG"
    SUBMIT_DIALOG = "SUBMIT_DIALOG"
    CANCEL_DIALOG = "CANCEL_DIALOG"

DialogMetadata dataclass

Incoming dialog state without any response-building behavior.

Source code in src/chattice/events/common.py

@dataclass(frozen=True, slots=True, kw_only=True)
class DialogMetadata:
    """Incoming dialog state without any response-building behavior."""

    type: DialogEventType | str
    is_dialog_event: bool = True

ErrorEvent dataclass

Bases: Event

The original event and exception presented to an error observer.

Source code in src/chattice/events/error.py

@dataclass(frozen=True, slots=True, kw_only=True)
class ErrorEvent(Event):
    """The original event and exception presented to an error observer."""

    event_type: str = field(default="error", init=False)
    source_event: Event
    exception: Exception

Event dataclass

Base class for transport-independent domain events.

Source code in src/chattice/events/base.py

@dataclass(frozen=True, slots=True, kw_only=True)
class Event:
    """Base class for transport-independent domain events."""

    event_type: str = "event"
    raw: object = field(default=None, repr=False, compare=False)
    event_time: datetime | None = None
    actor: UserRef | None = None
    space: SpaceRef | None = None
    thread: ThreadRef | None = None
    dialog: DialogMetadata | None = None
    locale: str | None = None
    timezone: TimeZone | None = None

FormInputs dataclass

Bases: Mapping[str, FormValue]

Immutable mapping from widget names to typed submitted values.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class FormInputs(Mapping[str, FormValue]):
    """Immutable mapping from widget names to typed submitted values."""

    data: Mapping[str, FormValue] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "data", MappingProxyType(dict(self.data)))

    def __getitem__(self, key: str) -> FormValue:
        return self.data[key]

    def __iter__(self) -> Iterator[str]:
        return iter(self.data)

    def __len__(self) -> int:
        return len(self.data)

FormSubmitEvent dataclass

Bases: Event

A form submitted from App Home.

Source code in src/chattice/events/app_home.py

@dataclass(frozen=True, slots=True, kw_only=True)
class FormSubmitEvent(Event):
    """A form submitted from App Home."""

    event_type: str = field(default="form_submit", init=False)
    function_name: str = ""
    parameters: Mapping[str, str] = field(default_factory=dict)
    form_inputs: FormInputs = field(default_factory=FormInputs)

    def __post_init__(self) -> None:
        object.__setattr__(self, "parameters", MappingProxyType(dict(self.parameters)))

MessageEvent dataclass

Bases: Event

A normalized Chat message interaction.

text is Google's raw text; argument_text (when present) is the documented mention-stripped body — route on it to handle @MyApp ping as ping without a custom mention parser.

Source code in src/chattice/events/message.py

@dataclass(frozen=True, slots=True, kw_only=True)
class MessageEvent(Event):
    """A normalized Chat message interaction.

    ``text`` is Google's raw text; ``argument_text`` (when present) is the
    documented mention-stripped body — route on it to handle
    ``@MyApp ping`` as ``ping`` without a custom mention parser.
    """

    event_type: str = field(default="message", init=False)
    text: str = ""
    message: MessageRef | None = None
    matched_url: str | None = None
    sender_type: str | None = None
    argument_text: str | None = None

    @property
    def attachments(self) -> tuple[Mapping[str, object], ...]:
        """Lossless snapshots of Google's ``message.attachment`` entries."""
        return _mapping_sequence(_raw_message(self.raw).get("attachment"))

    @property
    def attachment_refs(self) -> tuple[AttachmentRef, ...]:
        """Typed inbound attachment metadata (additive over ``attachments``).

        Distinguishes UPLOADED_CONTENT from DRIVE_FILE and exposes the
        human-facing thumbnail/download links next to the programmatic
        ``attachmentDataRef.resourceName`` download handle.
        """
        return tuple(AttachmentRef.from_mapping(m) for m in self.attachments)

    @property
    def annotations(self) -> tuple[Mapping[str, object], ...]:
        """Lossless snapshots of Google's output-only annotations."""
        return _mapping_sequence(_raw_message(self.raw).get("annotations"))

    @property
    def mentions(self) -> tuple[Mapping[str, object], ...]:
        """User-mention annotations, preserving ranges and mention metadata."""
        return tuple(
            annotation
            for annotation in self.annotations
            if annotation.get("type") == "USER_MENTION"
        )

    @property
    def quote(self) -> Mapping[str, object] | None:
        """Lossless ``quotedMessageMetadata`` snapshot, when present."""
        return _mapping_snapshot(_raw_message(self.raw).get("quotedMessageMetadata"))

    @property
    def reaction_summaries(self) -> tuple[Mapping[str, object], ...]:
        """Lossless snapshots of Google's emoji reaction summaries."""
        return _mapping_sequence(_raw_message(self.raw).get("emojiReactionSummaries"))

    @property
    def is_private(self) -> bool:
        """Whether Google marks the message for a private message viewer."""
        return _raw_message(self.raw).get("privateMessageViewer") is not None

    @property
    def is_silent(self) -> bool:
        """Whether Google suppressed push notifications for the message."""
        return _raw_message(self.raw).get("silent") is True

    async def reply(
        self,
        text: str | None = None,
        *,
        reply_option: object | None = None,
        request_id: str | None = None,
        message_id: str | None = None,
        timeout: float | None = None,
        accessory_widgets: Sequence[AccessoryWidget] | None = None,
        card: Card | None = None,
        notify: str | None = None,
        private_to: UserRef | str | None = None,
        attachments: Sequence[InputFile | UploadedAttachment] | None = None,
        bot: object | None = None,
    ) -> Message:
        """Reply in this message's known thread through the bound Bot."""
        from google.apps.chat_v1.types.message import CreateMessageRequest

        if self.space is None:
            raise RuntimeError("MessageEvent.reply() requires a known space")
        if self.thread is None:
            raise RuntimeError("MessageEvent.reply() requires a known thread")
        if reply_option is None:
            # A reply to a concrete incoming message must never silently
            # fall back to a new thread.
            reply_option = CreateMessageRequest.MessageReplyOption.REPLY_MESSAGE_OR_FAIL
        return await _send_message(
            bot,
            self.space,
            text,
            thread=self.thread,
            reply_option=reply_option,
            request_id=request_id,
            message_id=message_id,
            timeout=timeout,
            accessory_widgets=accessory_widgets,
            card=card,
            notify=notify,
            private_to=private_to,
            attachments=attachments,
        )

annotations property

Lossless snapshots of Google's output-only annotations.

attachment_refs property

Typed inbound attachment metadata (additive over attachments).

Distinguishes UPLOADED_CONTENT from DRIVE_FILE and exposes the human-facing thumbnail/download links next to the programmatic attachmentDataRef.resourceName download handle.

attachments property

Lossless snapshots of Google's message.attachment entries.

is_private property

Whether Google marks the message for a private message viewer.

is_silent property

Whether Google suppressed push notifications for the message.

mentions property

User-mention annotations, preserving ranges and mention metadata.

quote property

Lossless quotedMessageMetadata snapshot, when present.

reaction_summaries property

Lossless snapshots of Google's emoji reaction summaries.

reply(text=None, *, reply_option=None, request_id=None, message_id=None, timeout=None, accessory_widgets=None, card=None, notify=None, private_to=None, attachments=None, bot=None) async

Reply in this message's known thread through the bound Bot.

Source code in src/chattice/events/message.py

async def reply(
    self,
    text: str | None = None,
    *,
    reply_option: object | None = None,
    request_id: str | None = None,
    message_id: str | None = None,
    timeout: float | None = None,
    accessory_widgets: Sequence[AccessoryWidget] | None = None,
    card: Card | None = None,
    notify: str | None = None,
    private_to: UserRef | str | None = None,
    attachments: Sequence[InputFile | UploadedAttachment] | None = None,
    bot: object | None = None,
) -> Message:
    """Reply in this message's known thread through the bound Bot."""
    from google.apps.chat_v1.types.message import CreateMessageRequest

    if self.space is None:
        raise RuntimeError("MessageEvent.reply() requires a known space")
    if self.thread is None:
        raise RuntimeError("MessageEvent.reply() requires a known thread")
    if reply_option is None:
        # A reply to a concrete incoming message must never silently
        # fall back to a new thread.
        reply_option = CreateMessageRequest.MessageReplyOption.REPLY_MESSAGE_OR_FAIL
    return await _send_message(
        bot,
        self.space,
        text,
        thread=self.thread,
        reply_option=reply_option,
        request_id=request_id,
        message_id=message_id,
        timeout=timeout,
        accessory_widgets=accessory_widgets,
        card=card,
        notify=notify,
        private_to=private_to,
        attachments=attachments,
    )

MessageRef dataclass

Minimal message identity exposed to ordinary handlers.

Source code in src/chattice/events/references.py

@dataclass(frozen=True, slots=True, kw_only=True)
class MessageRef:
    """Minimal message identity exposed to ordinary handlers."""

    name: str | None = None

RemovedFromSpaceEvent dataclass

Bases: Event

The Chat app was removed from a space.

Source code in src/chattice/events/space.py

@dataclass(frozen=True, slots=True, kw_only=True)
class RemovedFromSpaceEvent(Event):
    """The Chat app was removed from a space."""

    event_type: str = field(default="removed_from_space", init=False)

SpaceRef dataclass

Minimal Chat space reference.

space_type (DIRECT_MESSAGE / GROUP_CHAT / SPACE) and single_user_bot_dm distinguish the personal bot DM (the Home tab host space) from collaborative spaces — the Home DM space must never be treated as a publish destination.

Source code in src/chattice/events/references.py

@dataclass(frozen=True, slots=True, kw_only=True)
class SpaceRef:
    """Minimal Chat space reference.

    ``space_type`` (DIRECT_MESSAGE / GROUP_CHAT / SPACE) and
    ``single_user_bot_dm`` distinguish the personal bot DM (the Home tab
    host space) from collaborative spaces — the Home DM
    space must never be treated as a publish destination.
    """

    name: str | None = None
    display_name: str | None = None
    type: str | None = None
    space_type: str | None = None
    single_user_bot_dm: bool | None = None

    async def send(
        self,
        text: str | None = None,
        *,
        thread: ThreadRef | None = None,
        reply_option: object | None = None,
        request_id: str | None = None,
        message_id: str | None = None,
        timeout: float | None = None,
        accessory_widgets: Sequence[AccessoryWidget] | None = None,
        card: Card | None = None,
        notify: str | None = None,
        private_to: UserRef | str | None = None,
        attachments: Sequence[InputFile | UploadedAttachment] | None = None,
        bot: object | None = None,
    ) -> Message:
        """Send through the bound Bot without fetching this space."""
        return await _send_message(
            bot,
            self,
            text,
            thread=thread,
            reply_option=reply_option,
            request_id=request_id,
            message_id=message_id,
            timeout=timeout,
            accessory_widgets=accessory_widgets,
            card=card,
            notify=notify,
            private_to=private_to,
            attachments=attachments,
        )

send(text=None, *, thread=None, reply_option=None, request_id=None, message_id=None, timeout=None, accessory_widgets=None, card=None, notify=None, private_to=None, attachments=None, bot=None) async

Send through the bound Bot without fetching this space.

Source code in src/chattice/events/references.py

async def send(
    self,
    text: str | None = None,
    *,
    thread: ThreadRef | None = None,
    reply_option: object | None = None,
    request_id: str | None = None,
    message_id: str | None = None,
    timeout: float | None = None,
    accessory_widgets: Sequence[AccessoryWidget] | None = None,
    card: Card | None = None,
    notify: str | None = None,
    private_to: UserRef | str | None = None,
    attachments: Sequence[InputFile | UploadedAttachment] | None = None,
    bot: object | None = None,
) -> Message:
    """Send through the bound Bot without fetching this space."""
    return await _send_message(
        bot,
        self,
        text,
        thread=thread,
        reply_option=reply_option,
        request_id=request_id,
        message_id=message_id,
        timeout=timeout,
        accessory_widgets=accessory_widgets,
        card=card,
        notify=notify,
        private_to=private_to,
        attachments=attachments,
    )

StringInput dataclass

One or more text or selection values.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class StringInput:
    """One or more text or selection values."""

    values: tuple[str, ...] = ()

ThreadRef dataclass

Minimal Chat thread reference with an optional known parent space.

Source code in src/chattice/events/references.py

@dataclass(frozen=True, slots=True, kw_only=True)
class ThreadRef:
    """Minimal Chat thread reference with an optional known parent space."""

    name: str | None = None
    thread_key: str | None = None
    # Context enrichment must not change reference equality/hash/repr.
    space: SpaceRef | None = field(default=None, repr=False, compare=False)

    def _parent_space(self, explicit: SpaceRef | str | None) -> SpaceRef | str:
        if explicit is not None:
            return explicit
        if self.space is not None:
            return self.space
        if self.name is not None and "/threads/" in self.name:
            parent, _separator, _thread_id = self.name.partition("/threads/")
            if parent.startswith("spaces/"):
                return parent
        raise RuntimeError(
            "ThreadRef.send() requires its parent space. Use a parsed event "
            "thread, construct ThreadRef(..., space=...), or pass space=...."
        )

    async def send(
        self,
        text: str | None = None,
        *,
        space: SpaceRef | str | None = None,
        reply_option: object | None = None,
        request_id: str | None = None,
        message_id: str | None = None,
        timeout: float | None = None,
        accessory_widgets: Sequence[AccessoryWidget] | None = None,
        card: Card | None = None,
        notify: str | None = None,
        private_to: UserRef | str | None = None,
        attachments: Sequence[InputFile | UploadedAttachment] | None = None,
        bot: object | None = None,
    ) -> Message:
        """Send in this thread through the bound Bot with zero fetches."""
        return await _send_message(
            bot,
            self._parent_space(space),
            text,
            thread=self,
            reply_option=reply_option,
            request_id=request_id,
            message_id=message_id,
            timeout=timeout,
            accessory_widgets=accessory_widgets,
            card=card,
            notify=notify,
            private_to=private_to,
            attachments=attachments,
        )

send(text=None, *, space=None, reply_option=None, request_id=None, message_id=None, timeout=None, accessory_widgets=None, card=None, notify=None, private_to=None, attachments=None, bot=None) async

Send in this thread through the bound Bot with zero fetches.

Source code in src/chattice/events/references.py

async def send(
    self,
    text: str | None = None,
    *,
    space: SpaceRef | str | None = None,
    reply_option: object | None = None,
    request_id: str | None = None,
    message_id: str | None = None,
    timeout: float | None = None,
    accessory_widgets: Sequence[AccessoryWidget] | None = None,
    card: Card | None = None,
    notify: str | None = None,
    private_to: UserRef | str | None = None,
    attachments: Sequence[InputFile | UploadedAttachment] | None = None,
    bot: object | None = None,
) -> Message:
    """Send in this thread through the bound Bot with zero fetches."""
    return await _send_message(
        bot,
        self._parent_space(space),
        text,
        thread=self,
        reply_option=reply_option,
        request_id=request_id,
        message_id=message_id,
        timeout=timeout,
        accessory_widgets=accessory_widgets,
        card=card,
        notify=notify,
        private_to=private_to,
        attachments=attachments,
    )

TimeInput dataclass

Wall-clock time components.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class TimeInput:
    """Wall-clock time components."""

    hours: int
    minutes: int

TimeZone dataclass

Locale-independent timezone metadata supplied by Google.

Source code in src/chattice/events/common.py

@dataclass(frozen=True, slots=True, kw_only=True)
class TimeZone:
    """Locale-independent timezone metadata supplied by Google."""

    id: str | None = None
    offset_ms: int | None = None

UnknownEvent dataclass

Bases: Event

An event whose external type is not understood by the framework.

Source code in src/chattice/events/unknown.py

@dataclass(frozen=True, slots=True, kw_only=True)
class UnknownEvent(Event):
    """An event whose external type is not understood by the framework."""

    event_type: str = field(default="unknown", init=False)
    original_type: str = ""

UnknownFormInput dataclass

A future input variant retained without claiming its semantics.

Source code in src/chattice/events/form.py

@dataclass(frozen=True, slots=True, kw_only=True)
class UnknownFormInput:
    """A future input variant retained without claiming its semantics."""

    kind: str
    raw: Mapping[str, object]

    def __post_init__(self) -> None:
        object.__setattr__(self, "raw", MappingProxyType(dict(self.raw)))

UserRef dataclass

Stable user identity and optional presentation metadata.

Source code in src/chattice/events/references.py

@dataclass(frozen=True, slots=True, kw_only=True)
class UserRef:
    """Stable user identity and optional presentation metadata."""

    name: str | None = None
    display_name: str | None = None
    type: str | None = None

WidgetUpdatedEvent dataclass

Bases: Event

A widget with an associated action was updated.

Source code in src/chattice/events/widget.py

@dataclass(frozen=True, slots=True, kw_only=True)
class WidgetUpdatedEvent(Event):
    """A widget with an associated action was updated."""

    event_type: str = field(default="widget_updated", init=False)
    function_name: str = ""
    parameters: Mapping[str, str] = field(default_factory=dict)
    form_inputs: FormInputs = field(default_factory=FormInputs)

    def __post_init__(self) -> None:
        object.__setattr__(self, "parameters", MappingProxyType(dict(self.parameters)))

Public filtering API.

BaseFilter

Convenience base class for asynchronous custom filters.

Source code in src/chattice/filters/base.py

class BaseFilter:
    """Convenience base class for asynchronous custom filters."""

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue:
        raise NotImplementedError

Filter

Bases: Protocol

Structural protocol implemented by asynchronous custom filters.

Source code in src/chattice/filters/base.py

class Filter(Protocol):
    """Structural protocol implemented by asynchronous custom filters."""

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue: ...

MagicExpression

Bases: BaseFilter

Base class for immutable boolean expression nodes.

Source code in src/chattice/filters/magic.py

class MagicExpression(BaseFilter):
    """Base class for immutable boolean expression nodes."""

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue:
        del context
        return self.evaluate(event)

    def evaluate(self, event: Event) -> bool:
        """Evaluate this expression against an event."""
        raise NotImplementedError

    def __and__(self, other: MagicExpression) -> MagicExpression:
        return _BooleanExpression("and", self, other)

    def __or__(self, other: MagicExpression) -> MagicExpression:
        return _BooleanExpression("or", self, other)

    def __invert__(self) -> MagicExpression:
        return _NotExpression(self)

    def __bool__(self) -> bool:
        raise TypeError("Magic-filter expressions cannot be used as Python booleans")

evaluate(event)

Evaluate this expression against an event.

Source code in src/chattice/filters/magic.py

def evaluate(self, event: Event) -> bool:
    """Evaluate this expression against an event."""
    raise NotImplementedError

MagicField dataclass

Bases: MagicExpression

A safely traversed event attribute/item path.

Source code in src/chattice/filters/magic.py

@dataclass(frozen=True, slots=True)
class MagicField(MagicExpression):
    """A safely traversed event attribute/item path."""

    path: tuple[PathStep, ...] = ()

    def __getattr__(self, name: str) -> MagicField:
        if name.startswith("__"):
            raise AttributeError(name)
        return MagicField((*self.path, ("attr", name)))

    def __getitem__(self, key: object) -> MagicField:
        return MagicField((*self.path, ("item", key)))

    def __eq__(self, other: object) -> MagicExpression:  # type: ignore[override]
        return _ComparisonExpression("eq", self, other)

    def __ne__(self, other: object) -> MagicExpression:  # type: ignore[override]
        return _ComparisonExpression("ne", self, other)

    def __lt__(self, other: object) -> MagicExpression:
        return _ComparisonExpression("lt", self, other)

    def __le__(self, other: object) -> MagicExpression:
        return _ComparisonExpression("le", self, other)

    def __gt__(self, other: object) -> MagicExpression:
        return _ComparisonExpression("gt", self, other)

    def __ge__(self, other: object) -> MagicExpression:
        return _ComparisonExpression("ge", self, other)

    def contains(self, value: object) -> MagicExpression:
        """Match when the resolved field contains ``value``."""
        return _MethodExpression("contains", self, value)

    def startswith(self, value: object) -> MagicExpression:
        """Match when the resolved field starts with ``value``."""
        return _MethodExpression("startswith", self, value)

    def endswith(self, value: object) -> MagicExpression:
        """Match when the resolved field ends with ``value``."""
        return _MethodExpression("endswith", self, value)

    def in_(self, value: object) -> MagicExpression:
        """Match when the resolved field is a member of ``value``."""
        return _MethodExpression("in", self, value)

    def is_(self, value: object) -> MagicExpression:
        """Match by object identity."""
        return _MethodExpression("is", self, value)

    def exists(self) -> MagicExpression:
        """Match when the complete path can be resolved."""
        return _MethodExpression("exists", self, None)

    def regexp(self, pattern: str | re.Pattern[str], flags: int = 0) -> MagicExpression:
        """Match a string field with a Python regular expression.

        Uses ``re.match`` semantics — the pattern must match FROM THE
        START of the value. Accepts a pattern string (compiled once, at
        filter construction) or a pre-compiled ``re.Pattern``. Invalid
        patterns raise ``ValueError`` at construction, never at
        evaluation time. ``flags`` cannot be combined with a compiled
        pattern. Missing fields and non-string values never match.

        Do not wrap patterns in ``/.../`` — this is Python regex syntax.
        """
        if isinstance(pattern, re.Pattern):
            if flags:
                raise ValueError("flags cannot be combined with a compiled re.Pattern")
            # cast: re.Pattern[bytes] flows in from untyped callers even
            # though the annotation says str — reject it explicitly.
            if isinstance(cast(Any, pattern).pattern, bytes):
                raise ValueError(
                    "bytes regex patterns cannot match string fields; "
                    "compile a str pattern instead"
                )
            compiled: re.Pattern[str] = pattern
        else:
            try:
                compiled = re.compile(pattern, flags)
            except re.error as error:
                raise ValueError(f"invalid regex pattern: {error}") from error
        return _MethodExpression("regexp", self, compiled)

    def resolve(self, event: Event) -> object:
        """Resolve this path, returning a private missing sentinel on absence."""
        current: object = event
        for operation, value in self.path:
            try:
                if operation == "attr":
                    current = getattr(current, str(value))
                else:
                    current = current[value]  # type: ignore[index]
            except (AttributeError, KeyError, IndexError, TypeError):
                return MISSING
        return current

    def evaluate(self, event: Event) -> bool:
        value = self.resolve(event)
        return value is not MISSING and bool(value)

contains(value)

Match when the resolved field contains value.

Source code in src/chattice/filters/magic.py

def contains(self, value: object) -> MagicExpression:
    """Match when the resolved field contains ``value``."""
    return _MethodExpression("contains", self, value)

endswith(value)

Match when the resolved field ends with value.

Source code in src/chattice/filters/magic.py

def endswith(self, value: object) -> MagicExpression:
    """Match when the resolved field ends with ``value``."""
    return _MethodExpression("endswith", self, value)

exists()

Match when the complete path can be resolved.

Source code in src/chattice/filters/magic.py

def exists(self) -> MagicExpression:
    """Match when the complete path can be resolved."""
    return _MethodExpression("exists", self, None)

in_(value)

Match when the resolved field is a member of value.

Source code in src/chattice/filters/magic.py

def in_(self, value: object) -> MagicExpression:
    """Match when the resolved field is a member of ``value``."""
    return _MethodExpression("in", self, value)

is_(value)

Match by object identity.

Source code in src/chattice/filters/magic.py

def is_(self, value: object) -> MagicExpression:
    """Match by object identity."""
    return _MethodExpression("is", self, value)

regexp(pattern, flags=0)

Match a string field with a Python regular expression.

Uses re.match semantics — the pattern must match FROM THE START of the value. Accepts a pattern string (compiled once, at filter construction) or a pre-compiled re.Pattern. Invalid patterns raise ValueError at construction, never at evaluation time. flags cannot be combined with a compiled pattern. Missing fields and non-string values never match.

Do not wrap patterns in /.../ — this is Python regex syntax.

Source code in src/chattice/filters/magic.py

def regexp(self, pattern: str | re.Pattern[str], flags: int = 0) -> MagicExpression:
    """Match a string field with a Python regular expression.

    Uses ``re.match`` semantics — the pattern must match FROM THE
    START of the value. Accepts a pattern string (compiled once, at
    filter construction) or a pre-compiled ``re.Pattern``. Invalid
    patterns raise ``ValueError`` at construction, never at
    evaluation time. ``flags`` cannot be combined with a compiled
    pattern. Missing fields and non-string values never match.

    Do not wrap patterns in ``/.../`` — this is Python regex syntax.
    """
    if isinstance(pattern, re.Pattern):
        if flags:
            raise ValueError("flags cannot be combined with a compiled re.Pattern")
        # cast: re.Pattern[bytes] flows in from untyped callers even
        # though the annotation says str — reject it explicitly.
        if isinstance(cast(Any, pattern).pattern, bytes):
            raise ValueError(
                "bytes regex patterns cannot match string fields; "
                "compile a str pattern instead"
            )
        compiled: re.Pattern[str] = pattern
    else:
        try:
            compiled = re.compile(pattern, flags)
        except re.error as error:
            raise ValueError(f"invalid regex pattern: {error}") from error
    return _MethodExpression("regexp", self, compiled)

resolve(event)

Resolve this path, returning a private missing sentinel on absence.

Source code in src/chattice/filters/magic.py

def resolve(self, event: Event) -> object:
    """Resolve this path, returning a private missing sentinel on absence."""
    current: object = event
    for operation, value in self.path:
        try:
            if operation == "attr":
                current = getattr(current, str(value))
            else:
                current = current[value]  # type: ignore[index]
        except (AttributeError, KeyError, IndexError, TypeError):
            return MISSING
    return current

startswith(value)

Match when the resolved field starts with value.

Source code in src/chattice/filters/magic.py

def startswith(self, value: object) -> MagicExpression:
    """Match when the resolved field starts with ``value``."""
    return _MethodExpression("startswith", self, value)

Transport-independent dispatch middleware.

BaseMiddleware

Convenience base class for asynchronous dispatch middleware.

Source code in src/chattice/middleware.py

class BaseMiddleware:
    """Convenience base class for asynchronous dispatch middleware."""

    async def __call__(
        self,
        handler: NextHandler,
        event: Event,
        data: MutableMapping[str, object],
    ) -> object:
        raise NotImplementedError

Middleware

Bases: Protocol

Structural protocol for asynchronous dispatch middleware.

Source code in src/chattice/middleware.py

class Middleware(Protocol):
    """Structural protocol for asynchronous dispatch middleware."""

    async def __call__(
        self,
        handler: NextHandler,
        event: Event,
        data: MutableMapping[str, object],
    ) -> object: ...

Actions, Cards, and Forms

Typed action data: a deterministic codec above Google action parameters.

Google card buttons carry action.function (the discriminator — used by @router.action(...)) plus a flat parameters mapping of string key/value pairs. ActionData turns those strings into typed Python fields WITHOUT an aiogram-style packed callback string and WITHOUT a registry: the action function name IS the discriminator.

Codec (deterministic, no eval): - str stays as-is; int/float/bool/Enum encode to canonical strings; - None (optional) fields are OMITTED from parameters and restored from their default on decode; - unknown parameters are ignored (forward compatibility with new Google fields); - malformed values raise ActionDataDecodeError on explicit decode and make the filter NOT match (no partial state); - documented system parameters (e.g. autocomplete_widget_query) are never interpreted by the codec.

ActionData

Base class for typed action parameter models (dataclass subclasses).

Source code in src/chattice/actions.py

class ActionData:
    """Base class for typed action parameter models (dataclass subclasses)."""

    function: ClassVar[str | None] = None

    def __init_subclass__(
        cls, *, function: str | None = None, **kwargs: object
    ) -> None:
        """Optionally bind the model to Google's action function discriminator."""
        super().__init_subclass__(**kwargs)
        if function is not None and not function.strip():
            raise ValueError("ActionData function must be non-empty")
        if function is not None:
            cls.function = function

    @classmethod
    def _fields(cls) -> dict[str, type[object]]:
        if not hasattr(cls, "__dataclass_fields__"):
            raise TypeError(f"{cls.__name__} must be a dataclass ActionData model")
        hints = get_type_hints(cls)
        return {
            field.name: hints[field.name]
            for field in dataclasses.fields(cast(Any, cls))
            if field.name in hints
        }

    def to_parameters(self) -> dict[str, str]:
        """Encode the typed fields into Google action parameters."""
        parameters: dict[str, str] = {}
        for name in self._fields():
            value = getattr(self, name)
            if value is None:
                continue  # optional field: omitted -> default on decode
            parameters[name] = _encode_value(value)
        return parameters

    @classmethod
    def from_parameters(cls, parameters: Mapping[str, str]) -> Self:
        """Decode Google action parameters into an instance.

        Unknown parameters are ignored (forward compatibility). Missing
        optional fields fall back to their dataclass defaults; missing
        REQUIRED fields raise ActionDataDecodeError.
        """
        fields = cls._fields()
        values: dict[str, object] = {}
        for name, target in fields.items():
            if name not in parameters:
                if name in _required_field_names(cls):
                    raise ActionDataDecodeError(
                        f"missing required parameter {name!r} for {cls.__name__}"
                    )
                continue  # optional: dataclass default applies
            values[name] = _decode_value(parameters[name], _unwrap_optional(target))
        return cls(**values)

    @classmethod
    def filter(cls) -> ActionDataFilter:
        """An async filter matching when parameters decode into this model.

        On a match the decoded instance is injected into the handler
        context under the name ``data``:

        @router.action("deploy.confirm", DeployAction.filter())
        async def confirm(event: ActionEvent, data: DeployAction): ...
        """
        return ActionDataFilter(cls)

__init_subclass__(*, function=None, **kwargs)

Optionally bind the model to Google's action function discriminator.

Source code in src/chattice/actions.py

def __init_subclass__(
    cls, *, function: str | None = None, **kwargs: object
) -> None:
    """Optionally bind the model to Google's action function discriminator."""
    super().__init_subclass__(**kwargs)
    if function is not None and not function.strip():
        raise ValueError("ActionData function must be non-empty")
    if function is not None:
        cls.function = function

filter() classmethod

An async filter matching when parameters decode into this model.

On a match the decoded instance is injected into the handler context under the name data:

@router.action("deploy.confirm", DeployAction.filter()) async def confirm(event: ActionEvent, data: DeployAction): ...

Source code in src/chattice/actions.py

@classmethod
def filter(cls) -> ActionDataFilter:
    """An async filter matching when parameters decode into this model.

    On a match the decoded instance is injected into the handler
    context under the name ``data``:

    @router.action("deploy.confirm", DeployAction.filter())
    async def confirm(event: ActionEvent, data: DeployAction): ...
    """
    return ActionDataFilter(cls)

from_parameters(parameters) classmethod

Decode Google action parameters into an instance.

Unknown parameters are ignored (forward compatibility). Missing optional fields fall back to their dataclass defaults; missing REQUIRED fields raise ActionDataDecodeError.

Source code in src/chattice/actions.py

@classmethod
def from_parameters(cls, parameters: Mapping[str, str]) -> Self:
    """Decode Google action parameters into an instance.

    Unknown parameters are ignored (forward compatibility). Missing
    optional fields fall back to their dataclass defaults; missing
    REQUIRED fields raise ActionDataDecodeError.
    """
    fields = cls._fields()
    values: dict[str, object] = {}
    for name, target in fields.items():
        if name not in parameters:
            if name in _required_field_names(cls):
                raise ActionDataDecodeError(
                    f"missing required parameter {name!r} for {cls.__name__}"
                )
            continue  # optional: dataclass default applies
        values[name] = _decode_value(parameters[name], _unwrap_optional(target))
    return cls(**values)

to_parameters()

Encode the typed fields into Google action parameters.

Source code in src/chattice/actions.py

def to_parameters(self) -> dict[str, str]:
    """Encode the typed fields into Google action parameters."""
    parameters: dict[str, str] = {}
    for name in self._fields():
        value = getattr(self, name)
        if value is None:
            continue  # optional field: omitted -> default on decode
        parameters[name] = _encode_value(value)
    return parameters

ActionDataDecodeError

Bases: ValueError

Action parameters cannot be decoded into the typed model.

Source code in src/chattice/actions.py

class ActionDataDecodeError(ValueError):
    """Action parameters cannot be decoded into the typed model."""

ActionDataFilter

Decode-based filter: returns {"data": instance} or False.

Source code in src/chattice/actions.py

class ActionDataFilter:
    """Decode-based filter: returns ``{"data": instance}`` or False."""

    def __init__(self, model: type[ActionData]) -> None:
        self.model = model

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue:
        if not isinstance(event, ActionEvent):
            return False
        parameters = {
            key: value
            for key, value in event.parameters.items()
            if isinstance(value, str)
        }
        try:
            instance = self.model.from_parameters(parameters)
        except ActionDataDecodeError as error:
            _logger.debug(
                "action data decode failed for %s: %s",
                self.model.__name__,
                error,
            )
            return False
        return {"data": instance}

Typed facade builders for Google Chat Cards v2.

AccessoryWidget dataclass

A message accessory widget (button list).

Source code in src/chattice/cards/accessory.py

@dataclass(frozen=True, slots=True)
class AccessoryWidget:
    """A message accessory widget (button list)."""

    button_list: ButtonList

    def to_proto(self) -> ProtoAccessoryWidget:
        return ProtoAccessoryWidget(button_list=self.button_list.to_proto())

    def to_dict(self) -> dict[str, object]:
        """The documented camelCase JSON shape (message.accessoryWidgets entry)."""
        return jsonlib.loads(  # type: ignore[no-any-return]
            MessageToJson(self.to_proto()._pb)
        )

to_dict()

The documented camelCase JSON shape (message.accessoryWidgets entry).

Source code in src/chattice/cards/accessory.py

def to_dict(self) -> dict[str, object]:
    """The documented camelCase JSON shape (message.accessoryWidgets entry)."""
    return jsonlib.loads(  # type: ignore[no-any-return]
        MessageToJson(self.to_proto()._pb)
    )

Action dataclass

An action invoked by a card widget (button click, form submit...).

Source code in src/chattice/cards/actions.py

@dataclass(frozen=True, slots=True)
class Action:
    """An action invoked by a card widget (button click, form submit...)."""

    function: str | None = None
    parameters: Mapping[str, str] = field(default_factory=dict)
    # Documented wire value, e.g. ButtonInteraction.OPEN_DIALOG ("OPEN_DIALOG").
    interaction: str | None = None

    def to_proto(self) -> ProtoAction:
        """Build the SDK Action proto (parameters stay strings)."""
        kwargs: dict[str, Any] = {}
        if self.function is not None:
            kwargs["function"] = self.function
        kwargs["parameters"] = [
            {"key": key, "value": value} for key, value in self.parameters.items()
        ]
        if self.interaction is not None:
            kwargs["interaction"] = self.interaction
        return ProtoAction(**kwargs)

    @classmethod
    def from_proto(cls, proto: ProtoAction) -> Action:
        """Rebuild the facade from an SDK proto."""
        # proto.interaction is an SDK enum member; normalize to the wire string
        # name so the facade field stays str | None (round-trip friendly).
        # The zero value INTERACTION_UNSPECIFIED is falsy, so "if proto.interaction"
        # correctly yields None when unset.
        return cls(
            function=proto.function or None,
            parameters={p.key: p.value for p in proto.parameters},
            interaction=proto.interaction.name if proto.interaction else None,
        )

from_proto(proto) classmethod

Rebuild the facade from an SDK proto.

Source code in src/chattice/cards/actions.py

@classmethod
def from_proto(cls, proto: ProtoAction) -> Action:
    """Rebuild the facade from an SDK proto."""
    # proto.interaction is an SDK enum member; normalize to the wire string
    # name so the facade field stays str | None (round-trip friendly).
    # The zero value INTERACTION_UNSPECIFIED is falsy, so "if proto.interaction"
    # correctly yields None when unset.
    return cls(
        function=proto.function or None,
        parameters={p.key: p.value for p in proto.parameters},
        interaction=proto.interaction.name if proto.interaction else None,
    )

to_proto()

Build the SDK Action proto (parameters stay strings).

Source code in src/chattice/cards/actions.py

def to_proto(self) -> ProtoAction:
    """Build the SDK Action proto (parameters stay strings)."""
    kwargs: dict[str, Any] = {}
    if self.function is not None:
        kwargs["function"] = self.function
    kwargs["parameters"] = [
        {"key": key, "value": value} for key, value in self.parameters.items()
    ]
    if self.interaction is not None:
        kwargs["interaction"] = self.interaction
    return ProtoAction(**kwargs)

ActionStatus dataclass

The outcome of a dialog submit, shown to the user.

Source code in src/chattice/cards/status.py

@dataclass(frozen=True, slots=True)
class ActionStatus:
    """The outcome of a dialog submit, shown to the user."""

    status_code: ActionStatusCode
    user_facing_message: str | None = None

    @classmethod
    def ok(cls, message: str | None = None) -> ActionStatus:
        """A successful submit (optional success message)."""
        return cls(ActionStatusCode.OK, message)

    @classmethod
    def invalid(cls, message: str) -> ActionStatus:
        """A validation failure shown to the user."""
        return cls(ActionStatusCode.INVALID_ARGUMENT, message)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to the documented actionStatus JSON."""
        data: dict[str, Any] = {"statusCode": self.status_code.value}
        if self.user_facing_message is not None:
            data["userFacingMessage"] = self.user_facing_message
        return data

invalid(message) classmethod

A validation failure shown to the user.

Source code in src/chattice/cards/status.py

@classmethod
def invalid(cls, message: str) -> ActionStatus:
    """A validation failure shown to the user."""
    return cls(ActionStatusCode.INVALID_ARGUMENT, message)

ok(message=None) classmethod

A successful submit (optional success message).

Source code in src/chattice/cards/status.py

@classmethod
def ok(cls, message: str | None = None) -> ActionStatus:
    """A successful submit (optional success message)."""
    return cls(ActionStatusCode.OK, message)

to_dict()

Serialize to the documented actionStatus JSON.

Source code in src/chattice/cards/status.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to the documented actionStatus JSON."""
    data: dict[str, Any] = {"statusCode": self.status_code.value}
    if self.user_facing_message is not None:
        data["userFacingMessage"] = self.user_facing_message
    return data

ActionStatusCode

Bases: Enum

Documented ActionStatus.StatusCode values.

Source code in src/chattice/cards/status.py

class ActionStatusCode(Enum):
    """Documented ActionStatus.StatusCode values."""

    OK = "OK"
    INVALID_ARGUMENT = "INVALID_ARGUMENT"

Button dataclass

A clickable button: either an action or a link.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class Button:
    """A clickable button: either an action or a link."""

    text: str
    action: str | ActionData | None = None
    interaction: str | None = None
    parameters: Mapping[str, str] = field(default_factory=dict)
    open_link: str | None = None
    # SDK accepts a google.type.Color instance or an RGB mapping
    # (e.g. {"red": 1.0, "green": 0.0, "blue": 0.0}).
    color: Any = None
    # Documented Button.type: OUTLINED/FILLED/FILLED_TONAL/BORDERLESS
    # (Chat apps only). Unset -> Google defaults to OUTLINED; when
    # ``color`` is set Google forces FILLED and ignores this value.
    type: str | None = None
    disabled: bool = False
    alt_text: str | None = None
    required_widgets: tuple[str, ...] = ()
    persist_values: bool = False
    load_indicator: bool = False

    def __post_init__(self) -> None:
        if isinstance(self.action, ActionData):
            if self.action.function is None:
                raise ValueError(
                    "Button ActionData requires a Google action function; "
                    "declare the model as class Deploy(ActionData, "
                    "function='deploy')"
                )
            if self.parameters:
                raise ValueError(
                    "Button parameters must be omitted when action is ActionData"
                )
            object.__setattr__(self, "parameters", self.action.to_parameters())
            object.__setattr__(self, "action", self.action.function)
        # exactly one of action/open_link — a button with both used
        # to silently prefer the action; neither only failed at
        # serialization. Fail at construction instead.
        if self.action is not None and self.open_link is not None:
            raise ValueError(
                "Button accepts either an action or an open link, not both"
            )
        object.__setattr__(self, "required_widgets", tuple(self.required_widgets))
        # Snapshot mutable mappings so the frozen facade cannot change
        # after validation.
        object.__setattr__(self, "parameters", dict(self.parameters))

    def to_proto(self) -> ProtoButton:
        if self.action is not None:
            action: dict[str, Any] = {
                "function": self.action,
                "parameters": [
                    {"key": key, "value": value}
                    for key, value in self.parameters.items()
                ],
            }
            if self.interaction is not None:
                action["interaction"] = self.interaction
            if self.required_widgets:
                action["required_widgets"] = list(self.required_widgets)
            if self.persist_values:
                action["persist_values"] = True
            # the SDK LoadIndicator enum has NO unspecified value
            # (SPINNER == 0 == unset), so write NONE explicitly — the
            # round-trip through the wire otherwise cannot tell "unset"
            # apart from "SPINNER".
            action["load_indicator"] = "SPINNER" if self.load_indicator else "NONE"
            on_click = {"action": action}
        elif self.open_link is not None:
            on_click = {"open_link": {"url": self.open_link}}
        else:
            raise ValueError("Button requires 'action' or 'open_link'")
        kwargs: dict[str, Any] = {
            "text": self.text,
            "on_click": on_click,
            "disabled": self.disabled,
        }
        if self.color is not None:
            kwargs["color"] = self.color
        if self.type is not None:
            kwargs["type"] = self.type
        if self.alt_text is not None:
            kwargs["alt_text"] = self.alt_text
        return ProtoButton(**kwargs)

ButtonInteraction

Documented action.interaction values.

Source code in src/chattice/cards/widgets.py

class ButtonInteraction:
    """Documented action.interaction values."""

    OPEN_DIALOG = "OPEN_DIALOG"

ButtonList dataclass

A horizontal row of buttons.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class ButtonList:
    """A horizontal row of buttons."""

    buttons: Sequence[Button] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        # Canonicalize the container so round-trips compare equal
        # regardless of whether the caller passed a list or a tuple.
        object.__setattr__(self, "buttons", tuple(self.buttons))

    def to_proto(self) -> ProtoButtonList:
        return ProtoButtonList(buttons=[b.to_proto() for b in self.buttons])

ButtonType

Documented Button.type values (Google Chat apps only).

https://developers.google.com/workspace/chat/api/reference/rest/v1/cards#button

Source code in src/chattice/cards/widgets.py

class ButtonType:
    """Documented Button.type values (Google Chat apps only).

    https://developers.google.com/workspace/chat/api/reference/rest/v1/cards#button
    """

    OUTLINED = "OUTLINED"  # default when unset
    FILLED = "FILLED"  # primary action, most visual impact
    FILLED_TONAL = "FILLED_TONAL"  # middle ground between filled and outlined
    BORDERLESS = "BORDERLESS"  # lowest priority

Card dataclass

A Google Chat Cards v2 card.

Source code in src/chattice/cards/card.py

@dataclass(frozen=True, slots=True)
class Card:
    """A Google Chat Cards v2 card."""

    header: CardHeader | None = None
    sections: Sequence[Section] = field(default_factory=tuple)
    name: str | None = None
    _raw: Mapping[str, object] | None = field(
        default=None, init=False, repr=False, compare=False
    )

    def __post_init__(self) -> None:
        object.__setattr__(self, "sections", tuple(self.sections))

    def to_proto(self) -> ProtoCard:
        """Build the SDK Card proto."""
        if self._raw is not None:
            parsed = ProtoCard()
            json_format.ParseDict(
                dict(self._raw), parsed._pb, ignore_unknown_fields=True
            )
            return parsed
        kwargs: dict[str, Any] = {}
        if self.header is not None:
            kwargs["header"] = self.header.to_proto()
        if self.name is not None:
            kwargs["name"] = self.name
        card = ProtoCard(**kwargs)
        for section in self.sections:
            card.sections.append(section.to_proto_dict())  # type: ignore[arg-type]
        return card

    def to_dict(self) -> dict[str, Any]:
        """Serialize to the documented camelCase Cards v2 JSON."""
        if self._raw is not None:
            return cast(
                dict[str, Any],
                deep_snapshot(self._raw, where="Card._raw"),
            )
        return to_dict(self.to_proto())

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Card:
        """Rebuild from Cards v2 JSON while preserving unknown fields."""
        snapshot = cast(
            dict[str, object],
            deep_snapshot(data, where="Card.from_dict"),
        )
        parsed = from_dict(snapshot, ignore_unknown_fields=True)
        card = cls.from_proto(parsed)

        # The protobuf parser cannot retain fields absent from its schema.
        # Replace unsupported widget facades with their original documented
        # JSON while the card-level raw snapshot preserves every field.
        raw_sections = snapshot.get("sections")
        if isinstance(raw_sections, list):
            rebuilt_sections: list[Section] = []
            for index, section in enumerate(card.sections):
                raw_section = raw_sections[index] if index < len(raw_sections) else None
                widgets = list(section.widgets)
                if isinstance(raw_section, Mapping):
                    raw_widgets = raw_section.get("widgets")
                    if isinstance(raw_widgets, list):
                        for widget_index, raw_widget in enumerate(raw_widgets):
                            if not isinstance(raw_widget, Mapping):
                                continue
                            if _SUPPORTED_WIDGET_KEYS & set(raw_widget):
                                continue
                            replacement = RawWidget(raw_widget)
                            if widget_index < len(widgets):
                                widgets[widget_index] = replacement
                            else:
                                widgets.append(replacement)
                rebuilt_sections.append(Section(header=section.header, widgets=widgets))
            object.__setattr__(card, "sections", tuple(rebuilt_sections))
        object.__setattr__(card, "_raw", snapshot)
        return card

    @classmethod
    def from_proto(cls, proto: ProtoCard) -> Card:
        """Rebuild the facade from an SDK proto."""
        header = None
        if proto.header.title or proto.header.subtitle or proto.header.image_url:
            header = CardHeader(
                title=proto.header.title,
                subtitle=proto.header.subtitle or None,
                image_url=proto.header.image_url or None,
            )
        sections = [Section.from_proto(s) for s in proto.sections]
        return cls(header=header, sections=sections, name=proto.name or None)

from_dict(data) classmethod

Rebuild from Cards v2 JSON while preserving unknown fields.

Source code in src/chattice/cards/card.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Card:
    """Rebuild from Cards v2 JSON while preserving unknown fields."""
    snapshot = cast(
        dict[str, object],
        deep_snapshot(data, where="Card.from_dict"),
    )
    parsed = from_dict(snapshot, ignore_unknown_fields=True)
    card = cls.from_proto(parsed)

    # The protobuf parser cannot retain fields absent from its schema.
    # Replace unsupported widget facades with their original documented
    # JSON while the card-level raw snapshot preserves every field.
    raw_sections = snapshot.get("sections")
    if isinstance(raw_sections, list):
        rebuilt_sections: list[Section] = []
        for index, section in enumerate(card.sections):
            raw_section = raw_sections[index] if index < len(raw_sections) else None
            widgets = list(section.widgets)
            if isinstance(raw_section, Mapping):
                raw_widgets = raw_section.get("widgets")
                if isinstance(raw_widgets, list):
                    for widget_index, raw_widget in enumerate(raw_widgets):
                        if not isinstance(raw_widget, Mapping):
                            continue
                        if _SUPPORTED_WIDGET_KEYS & set(raw_widget):
                            continue
                        replacement = RawWidget(raw_widget)
                        if widget_index < len(widgets):
                            widgets[widget_index] = replacement
                        else:
                            widgets.append(replacement)
            rebuilt_sections.append(Section(header=section.header, widgets=widgets))
        object.__setattr__(card, "sections", tuple(rebuilt_sections))
    object.__setattr__(card, "_raw", snapshot)
    return card

from_proto(proto) classmethod

Rebuild the facade from an SDK proto.

Source code in src/chattice/cards/card.py

@classmethod
def from_proto(cls, proto: ProtoCard) -> Card:
    """Rebuild the facade from an SDK proto."""
    header = None
    if proto.header.title or proto.header.subtitle or proto.header.image_url:
        header = CardHeader(
            title=proto.header.title,
            subtitle=proto.header.subtitle or None,
            image_url=proto.header.image_url or None,
        )
    sections = [Section.from_proto(s) for s in proto.sections]
    return cls(header=header, sections=sections, name=proto.name or None)

to_dict()

Serialize to the documented camelCase Cards v2 JSON.

Source code in src/chattice/cards/card.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to the documented camelCase Cards v2 JSON."""
    if self._raw is not None:
        return cast(
            dict[str, Any],
            deep_snapshot(self._raw, where="Card._raw"),
        )
    return to_dict(self.to_proto())

to_proto()

Build the SDK Card proto.

Source code in src/chattice/cards/card.py

def to_proto(self) -> ProtoCard:
    """Build the SDK Card proto."""
    if self._raw is not None:
        parsed = ProtoCard()
        json_format.ParseDict(
            dict(self._raw), parsed._pb, ignore_unknown_fields=True
        )
        return parsed
    kwargs: dict[str, Any] = {}
    if self.header is not None:
        kwargs["header"] = self.header.to_proto()
    if self.name is not None:
        kwargs["name"] = self.name
    card = ProtoCard(**kwargs)
    for section in self.sections:
        card.sections.append(section.to_proto_dict())  # type: ignore[arg-type]
    return card

CardHeader dataclass

The card header.

Source code in src/chattice/cards/card.py

@dataclass(frozen=True, slots=True)
class CardHeader:
    """The card header."""

    title: str
    subtitle: str | None = None
    image_url: str | None = None

    def to_proto(self) -> dict[str, Any]:
        """Build the header as a proto-plus dict (the SDK class is not exported)."""
        data: dict[str, Any] = {"title": self.title}
        if self.subtitle is not None:
            data["subtitle"] = self.subtitle
        if self.image_url is not None:
            data["image_url"] = self.image_url
        return data

to_proto()

Build the header as a proto-plus dict (the SDK class is not exported).

Source code in src/chattice/cards/card.py

def to_proto(self) -> dict[str, Any]:
    """Build the header as a proto-plus dict (the SDK class is not exported)."""
    data: dict[str, Any] = {"title": self.title}
    if self.subtitle is not None:
        data["subtitle"] = self.subtitle
    if self.image_url is not None:
        data["image_url"] = self.image_url
    return data

DateTimePicker dataclass

A date/time picker.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class DateTimePicker:
    """A date/time picker."""

    name: str
    label: str
    value_ms_epoch: int | None = None
    timezone_offset_date: int | None = None

    def to_proto(self) -> ProtoDateTimePicker:
        kwargs: dict[str, Any] = {"name": self.name, "label": self.label}
        if self.value_ms_epoch is not None:
            kwargs["value_ms_epoch"] = self.value_ms_epoch
        if self.timezone_offset_date is not None:
            kwargs["timezone_offset_date"] = self.timezone_offset_date
        return ProtoDateTimePicker(**kwargs)

Dialog dataclass

A dialog body displayed to the user who triggered the interaction.

Source code in src/chattice/cards/dialog.py

@dataclass(frozen=True, slots=True)
class Dialog:
    """A dialog body displayed to the user who triggered the interaction."""

    body: Card

    def __post_init__(self) -> None:
        for section in self.body.sections:
            for widget in section.widgets:
                _reject_datetime_picker(widget)

    def to_proto(self) -> ProtoDialog:
        """Build the chat SDK Dialog proto."""
        return ProtoDialog(body=self.body.to_proto())

    def to_dict(self) -> dict[str, Any]:
        """Serialize to the JSON shape carried under dialogAction.dialog.

        The Chat REST API nests the dialog under ``dialogAction.dialog``,
        so the payload is returned as ``{"dialog": {"body": ...}}``.
        """
        return {"dialog": json.loads(json_format.MessageToJson(self.to_proto()._pb))}

to_dict()

Serialize to the JSON shape carried under dialogAction.dialog.

The Chat REST API nests the dialog under dialogAction.dialog, so the payload is returned as {"dialog": {"body": ...}}.

Source code in src/chattice/cards/dialog.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to the JSON shape carried under dialogAction.dialog.

    The Chat REST API nests the dialog under ``dialogAction.dialog``,
    so the payload is returned as ``{"dialog": {"body": ...}}``.
    """
    return {"dialog": json.loads(json_format.MessageToJson(self.to_proto()._pb))}

to_proto()

Build the chat SDK Dialog proto.

Source code in src/chattice/cards/dialog.py

def to_proto(self) -> ProtoDialog:
    """Build the chat SDK Dialog proto."""
    return ProtoDialog(body=self.body.to_proto())

Divider dataclass

A horizontal divider between widgets.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class Divider:
    """A horizontal divider between widgets."""

    def to_proto(self) -> ProtoDivider:
        return ProtoDivider()

Image dataclass

A URL or lazily published local picture rendered inside a Card.

Card Image is a URL-based UI widget — the other Google media surface (a local file uploaded as a Chat attachment) is chattice.media.InputFile. from_path and from_bytes require an AssetPublisher on the Bot that sends or updates the Card.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class Image:
    """A URL or lazily published local picture rendered inside a Card.

    Card Image is a URL-based UI widget — the other Google media surface
    (a local file uploaded as a Chat attachment) is
    ``chattice.media.InputFile``. ``from_path`` and ``from_bytes`` require
    an AssetPublisher on the Bot that sends or updates the Card.
    """

    image_url: str | None
    alt_text: str | None = None
    on_click: Action | OpenLink | None = None
    _local_source: _LocalImageSource | None = field(
        default=None, init=False, repr=False, compare=True
    )

    def __post_init__(self) -> None:
        if self.image_url is None:
            raise ValueError("Image requires an absolute HTTPS URL")
        self._validate_url(self.image_url)
        self._snapshot_action()

    @staticmethod
    def _validate_url(image_url: str) -> None:
        parsed = urlparse(image_url)
        if parsed.scheme != "https" or not parsed.netloc:
            raise ValueError(
                "Image.image_url must be an absolute HTTPS URL; local "
                "paths, bytes and data: URLs require Image.from_path() "
                "or Image.from_bytes()"
            )

    def _snapshot_action(self) -> None:
        if isinstance(self.on_click, Action):
            # Snapshot mutable parameters so the frozen facade cannot
            # change after validation.
            object.__setattr__(
                self,
                "on_click",
                Action(
                    function=self.on_click.function,
                    parameters=dict(self.on_click.parameters),
                    interaction=self.on_click.interaction,
                ),
            )

    @classmethod
    def from_url(
        cls,
        image_url: str,
        *,
        alt_text: str | None = None,
        on_click: Action | OpenLink | None = None,
    ) -> Image:
        """Build an Image from an already published HTTPS URL."""
        return cls(image_url=image_url, alt_text=alt_text, on_click=on_click)

    @classmethod
    def from_path(
        cls,
        path: str | PathLike[str],
        *,
        filename: str | None = None,
        content_type: str | None = None,
        namespace: str | None = None,
        alt_text: str | None = None,
        on_click: Action | OpenLink | None = None,
    ) -> Image:
        """Build a lazy local Image without reading the file."""
        source_path = Path(path).expanduser().absolute()
        resolved_filename = _filename(filename or source_path.name)
        return cls._from_local_source(
            _PathImageSource(
                path=source_path,
                filename=resolved_filename,
                content_type=_content_type(resolved_filename, content_type),
                namespace=_namespace(namespace),
            ),
            alt_text=alt_text,
            on_click=on_click,
        )

    @classmethod
    def from_bytes(
        cls,
        data: bytes | bytearray | memoryview,
        *,
        filename: str,
        content_type: str | None = None,
        namespace: str | None = None,
        alt_text: str | None = None,
        on_click: Action | OpenLink | None = None,
    ) -> Image:
        """Build an Image from an immutable snapshot of generated bytes."""
        snapshot = bytes(data)
        if not snapshot:
            raise ValueError("Local Card image bytes cannot be empty")
        resolved_filename = _filename(filename)
        return cls._from_local_source(
            _BytesImageSource(
                data=snapshot,
                filename=resolved_filename,
                content_type=_content_type(resolved_filename, content_type),
                namespace=_namespace(namespace),
            ),
            alt_text=alt_text,
            on_click=on_click,
        )

    @classmethod
    def _from_local_source(
        cls,
        source: _LocalImageSource,
        *,
        alt_text: str | None,
        on_click: Action | OpenLink | None,
    ) -> Image:
        image = cls.__new__(cls)
        object.__setattr__(image, "image_url", None)
        object.__setattr__(image, "alt_text", alt_text)
        object.__setattr__(image, "on_click", on_click)
        object.__setattr__(image, "_local_source", source)
        image._snapshot_action()
        return image

    def to_proto(self) -> ProtoImage:
        """Build the SDK Image proto."""
        if self.image_url is None:
            from chattice.exceptions import AssetPublishError

            raise AssetPublishError(
                "Local Card images must be resolved through the message resource "
                "clients before serialization"
            )
        kwargs: dict[str, Any] = {"image_url": self.image_url}
        if self.alt_text is not None:
            kwargs["alt_text"] = self.alt_text
        if self.on_click is not None:
            if isinstance(self.on_click, Action):
                kwargs["on_click"] = {"action": self.on_click.to_proto()}
            else:
                kwargs["on_click"] = {"open_link": self.on_click.to_proto()}
        return ProtoImage(**kwargs)

from_bytes(data, *, filename, content_type=None, namespace=None, alt_text=None, on_click=None) classmethod

Build an Image from an immutable snapshot of generated bytes.

Source code in src/chattice/cards/widgets.py

@classmethod
def from_bytes(
    cls,
    data: bytes | bytearray | memoryview,
    *,
    filename: str,
    content_type: str | None = None,
    namespace: str | None = None,
    alt_text: str | None = None,
    on_click: Action | OpenLink | None = None,
) -> Image:
    """Build an Image from an immutable snapshot of generated bytes."""
    snapshot = bytes(data)
    if not snapshot:
        raise ValueError("Local Card image bytes cannot be empty")
    resolved_filename = _filename(filename)
    return cls._from_local_source(
        _BytesImageSource(
            data=snapshot,
            filename=resolved_filename,
            content_type=_content_type(resolved_filename, content_type),
            namespace=_namespace(namespace),
        ),
        alt_text=alt_text,
        on_click=on_click,
    )

from_path(path, *, filename=None, content_type=None, namespace=None, alt_text=None, on_click=None) classmethod

Build a lazy local Image without reading the file.

Source code in src/chattice/cards/widgets.py

@classmethod
def from_path(
    cls,
    path: str | PathLike[str],
    *,
    filename: str | None = None,
    content_type: str | None = None,
    namespace: str | None = None,
    alt_text: str | None = None,
    on_click: Action | OpenLink | None = None,
) -> Image:
    """Build a lazy local Image without reading the file."""
    source_path = Path(path).expanduser().absolute()
    resolved_filename = _filename(filename or source_path.name)
    return cls._from_local_source(
        _PathImageSource(
            path=source_path,
            filename=resolved_filename,
            content_type=_content_type(resolved_filename, content_type),
            namespace=_namespace(namespace),
        ),
        alt_text=alt_text,
        on_click=on_click,
    )

from_url(image_url, *, alt_text=None, on_click=None) classmethod

Build an Image from an already published HTTPS URL.

Source code in src/chattice/cards/widgets.py

@classmethod
def from_url(
    cls,
    image_url: str,
    *,
    alt_text: str | None = None,
    on_click: Action | OpenLink | None = None,
) -> Image:
    """Build an Image from an already published HTTPS URL."""
    return cls(image_url=image_url, alt_text=alt_text, on_click=on_click)

to_proto()

Build the SDK Image proto.

Source code in src/chattice/cards/widgets.py

def to_proto(self) -> ProtoImage:
    """Build the SDK Image proto."""
    if self.image_url is None:
        from chattice.exceptions import AssetPublishError

        raise AssetPublishError(
            "Local Card images must be resolved through the message resource "
            "clients before serialization"
        )
    kwargs: dict[str, Any] = {"image_url": self.image_url}
    if self.alt_text is not None:
        kwargs["alt_text"] = self.alt_text
    if self.on_click is not None:
        if isinstance(self.on_click, Action):
            kwargs["on_click"] = {"action": self.on_click.to_proto()}
        else:
            kwargs["on_click"] = {"open_link": self.on_click.to_proto()}
    return ProtoImage(**kwargs)

OpenLink dataclass

Opens a URL in a browser.

Source code in src/chattice/cards/actions.py

@dataclass(frozen=True, slots=True)
class OpenLink:
    """Opens a URL in a browser."""

    url: str
    open_as: Any = None

    def to_proto(self) -> ProtoOpenLink:
        """Build the SDK OpenLink proto."""
        kwargs: dict[str, Any] = {"url": self.url}
        if self.open_as is not None:
            kwargs["open_as"] = self.open_as
        return ProtoOpenLink(**kwargs)

    @classmethod
    def from_proto(cls, proto: ProtoOpenLink) -> OpenLink:
        """Rebuild the facade from an SDK proto."""
        return cls(url=proto.url, open_as=proto.open_as or None)

from_proto(proto) classmethod

Rebuild the facade from an SDK proto.

Source code in src/chattice/cards/actions.py

@classmethod
def from_proto(cls, proto: ProtoOpenLink) -> OpenLink:
    """Rebuild the facade from an SDK proto."""
    return cls(url=proto.url, open_as=proto.open_as or None)

to_proto()

Build the SDK OpenLink proto.

Source code in src/chattice/cards/actions.py

def to_proto(self) -> ProtoOpenLink:
    """Build the SDK OpenLink proto."""
    kwargs: dict[str, Any] = {"url": self.url}
    if self.open_as is not None:
        kwargs["open_as"] = self.open_as
    return ProtoOpenLink(**kwargs)

RawWidget dataclass

An arbitrary Cards v2 widget as documented camelCase JSON.

The payload is deep-snapshotted at construction — mutating the caller's mapping (including nested values) afterwards cannot change the widget, and the snapshot is what to_dict returns.

Source code in src/chattice/cards/raw.py

@dataclass(frozen=True, slots=True)
class RawWidget:
    """An arbitrary Cards v2 widget as documented camelCase JSON.

    The payload is deep-snapshotted at construction — mutating the
    caller's mapping (including nested values) afterwards cannot change
    the widget, and the snapshot is what ``to_dict`` returns.
    """

    payload: Mapping[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        snapshot = cast(
            dict[str, object],
            deep_snapshot(self.payload, where="RawWidget.payload"),
        )
        try:
            json_format.ParseDict(
                snapshot,
                ProtoWidget()._pb,
                ignore_unknown_fields=True,
            )
        except Exception as error:
            raise ValueError(
                f"RawWidget payload is not a valid Cards v2 widget: {error}"
            ) from error
        object.__setattr__(self, "payload", snapshot)

    def to_dict(self) -> dict[str, object]:
        """The documented camelCase widget JSON (lossless snapshot)."""
        return dict(self.payload)

to_dict()

The documented camelCase widget JSON (lossless snapshot).

Source code in src/chattice/cards/raw.py

def to_dict(self) -> dict[str, object]:
    """The documented camelCase widget JSON (lossless snapshot)."""
    return dict(self.payload)

Section dataclass

A card section: optional header plus a widget list.

Source code in src/chattice/cards/card.py

@dataclass(frozen=True, slots=True)
class Section:
    """A card section: optional header plus a widget list."""

    header: str | None = None
    widgets: Sequence[Widget] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        # Canonicalize the container so round-trips compare equal
        # regardless of whether the caller passed a list or a tuple.
        object.__setattr__(self, "widgets", tuple(self.widgets))

    def to_proto_dict(self) -> dict[str, Any]:
        """Build the section as a proto-plus dict (oneof dispatch)."""
        data: dict[str, Any] = {}
        if self.header is not None:
            data["header"] = self.header
        data["widgets"] = [self._widget_dict(w) for w in self.widgets]
        return data

    @staticmethod
    def _widget_dict(widget: Widget) -> dict[str, Any]:
        if isinstance(widget, TextParagraph):
            return {"text_paragraph": _proto_dict(widget.to_proto())}
        if isinstance(widget, Divider):
            return {"divider": _proto_dict(widget.to_proto())}
        if isinstance(widget, ButtonList):
            return {"button_list": _proto_dict(widget.to_proto())}
        if isinstance(widget, TextInput):
            return {"text_input": _proto_dict(widget.to_proto())}
        if isinstance(widget, SelectionInput):
            return {"selection_input": _proto_dict(widget.to_proto())}
        if isinstance(widget, DateTimePicker):
            return {"date_time_picker": _proto_dict(widget.to_proto())}
        if isinstance(widget, Image):
            return {"image": _proto_dict(widget.to_proto())}
        if isinstance(widget, RawWidget):
            # Escape hatch: parse the documented camelCase payload into a
            # proto and emit the proto-plus dict (unknown fields are
            # ignored on the SDK path — documented in RawWidget).
            parsed = ProtoWidget()
            json_format.ParseDict(
                widget.to_dict(), parsed._pb, ignore_unknown_fields=True
            )
            return _proto_dict(parsed)
        raise TypeError(f"Unsupported widget type {type(widget).__name__}")

    @classmethod
    def from_proto(cls, proto: Any) -> Section:
        """Rebuild the facade from an SDK Section proto (oneof dispatch).

        Unsupported SDK widget kinds become ``RawWidget`` instead of raising
        or being silently dropped.
        """
        widgets: list[Widget] = []
        for widget in proto.widgets:
            which = widget._pb.WhichOneof("data")
            if which == "text_paragraph":
                tp = widget.text_paragraph
                widgets.append(TextParagraph(tp.text, max_lines=tp.max_lines or None))
            elif which == "divider":
                widgets.append(Divider())
            elif which == "button_list":
                bl = widget.button_list
                rebuilt: list[Button] = []
                for b in bl.buttons:
                    click_kind = b.on_click._pb.WhichOneof("data")
                    if click_kind == "action":
                        action_proto = b.on_click.action
                        rebuilt.append(
                            Button(
                                b.text,
                                action=action_proto.function or None,
                                parameters={
                                    p.key: p.value for p in action_proto.parameters
                                },
                                interaction=(
                                    action_proto.interaction.name
                                    if action_proto.interaction
                                    else None
                                ),
                                disabled=b.disabled or False,
                                alt_text=b.alt_text or None,
                                # full round-trip — every facade-
                                # supported Button field is preserved.
                                required_widgets=tuple(action_proto.required_widgets),
                                persist_values=bool(action_proto.persist_values),
                                load_indicator=(
                                    action_proto.load_indicator.name == "SPINNER"
                                ),
                                type=_button_type_name(b.type),
                                color=(
                                    _color_to_mapping(b.color)
                                    if b._pb.HasField("color")
                                    else None
                                ),
                            )
                        )
                    elif click_kind == "open_link":
                        rebuilt.append(
                            Button(
                                b.text,
                                open_link=b.on_click.open_link.url,
                                disabled=b.disabled or False,
                                alt_text=b.alt_text or None,
                                type=_button_type_name(b.type),
                            )
                        )
                    else:
                        raise NotImplementedError(
                            f"Button onClick kind {click_kind!r} is not rebuildable"
                        )
                widgets.append(ButtonList(buttons=rebuilt))
            elif which == "text_input":
                ti = widget.text_input
                validation = None
                if ti.validation.character_limit or ti.validation.input_type:
                    validation = Validation(
                        character_limit=ti.validation.character_limit or None,
                        input_type=(
                            TextInputType(ti.validation.input_type.name)
                            if ti.validation.input_type
                            else None
                        ),
                    )
                widgets.append(
                    TextInput(
                        name=ti.name,
                        label=ti.label,
                        hint_text=ti.hint_text or None,
                        value=ti.value or None,
                        validation=validation,
                    )
                )
            elif which == "selection_input":
                si = widget.selection_input
                widgets.append(
                    SelectionInput(
                        name=si.name,
                        label=si.label,
                        items=tuple(
                            {"text": i.text, "value": i.value} for i in si.items
                        ),
                        # preserve the dynamic-selection surface.
                        external_data_source=(
                            Action(
                                function=si.external_data_source.function or None,
                                parameters={
                                    p.key: p.value
                                    for p in si.external_data_source.parameters
                                },
                            )
                            if si._pb.HasField("external_data_source")
                            else None
                        ),
                        multi_select_max_selected_items=(
                            si.multi_select_max_selected_items or None
                        ),
                        multi_select_min_query_length=(
                            si.multi_select_min_query_length or None
                        ),
                    )
                )
            elif which == "date_time_picker":
                dp = widget.date_time_picker
                widgets.append(
                    DateTimePicker(
                        name=dp.name,
                        label=dp.label,
                        value_ms_epoch=dp.value_ms_epoch or None,
                        timezone_offset_date=dp.timezone_offset_date or None,
                    )
                )
            elif which == "image":
                image = widget.image
                on_click: Action | OpenLink | None = None
                click_kind = image.on_click._pb.WhichOneof("data")
                if click_kind == "action":
                    on_click = Action.from_proto(image.on_click.action)
                elif click_kind == "open_link":
                    on_click = OpenLink.from_proto(image.on_click.open_link)
                widgets.append(
                    Image(
                        image_url=image.image_url,
                        alt_text=image.alt_text or None,
                        on_click=on_click,
                    )
                )
            else:
                widgets.append(RawWidget(_proto_json(widget)))
        return cls(header=proto.header or None, widgets=widgets)

from_proto(proto) classmethod

Rebuild the facade from an SDK Section proto (oneof dispatch).

Unsupported SDK widget kinds become RawWidget instead of raising or being silently dropped.

Source code in src/chattice/cards/card.py

@classmethod
def from_proto(cls, proto: Any) -> Section:
    """Rebuild the facade from an SDK Section proto (oneof dispatch).

    Unsupported SDK widget kinds become ``RawWidget`` instead of raising
    or being silently dropped.
    """
    widgets: list[Widget] = []
    for widget in proto.widgets:
        which = widget._pb.WhichOneof("data")
        if which == "text_paragraph":
            tp = widget.text_paragraph
            widgets.append(TextParagraph(tp.text, max_lines=tp.max_lines or None))
        elif which == "divider":
            widgets.append(Divider())
        elif which == "button_list":
            bl = widget.button_list
            rebuilt: list[Button] = []
            for b in bl.buttons:
                click_kind = b.on_click._pb.WhichOneof("data")
                if click_kind == "action":
                    action_proto = b.on_click.action
                    rebuilt.append(
                        Button(
                            b.text,
                            action=action_proto.function or None,
                            parameters={
                                p.key: p.value for p in action_proto.parameters
                            },
                            interaction=(
                                action_proto.interaction.name
                                if action_proto.interaction
                                else None
                            ),
                            disabled=b.disabled or False,
                            alt_text=b.alt_text or None,
                            # full round-trip — every facade-
                            # supported Button field is preserved.
                            required_widgets=tuple(action_proto.required_widgets),
                            persist_values=bool(action_proto.persist_values),
                            load_indicator=(
                                action_proto.load_indicator.name == "SPINNER"
                            ),
                            type=_button_type_name(b.type),
                            color=(
                                _color_to_mapping(b.color)
                                if b._pb.HasField("color")
                                else None
                            ),
                        )
                    )
                elif click_kind == "open_link":
                    rebuilt.append(
                        Button(
                            b.text,
                            open_link=b.on_click.open_link.url,
                            disabled=b.disabled or False,
                            alt_text=b.alt_text or None,
                            type=_button_type_name(b.type),
                        )
                    )
                else:
                    raise NotImplementedError(
                        f"Button onClick kind {click_kind!r} is not rebuildable"
                    )
            widgets.append(ButtonList(buttons=rebuilt))
        elif which == "text_input":
            ti = widget.text_input
            validation = None
            if ti.validation.character_limit or ti.validation.input_type:
                validation = Validation(
                    character_limit=ti.validation.character_limit or None,
                    input_type=(
                        TextInputType(ti.validation.input_type.name)
                        if ti.validation.input_type
                        else None
                    ),
                )
            widgets.append(
                TextInput(
                    name=ti.name,
                    label=ti.label,
                    hint_text=ti.hint_text or None,
                    value=ti.value or None,
                    validation=validation,
                )
            )
        elif which == "selection_input":
            si = widget.selection_input
            widgets.append(
                SelectionInput(
                    name=si.name,
                    label=si.label,
                    items=tuple(
                        {"text": i.text, "value": i.value} for i in si.items
                    ),
                    # preserve the dynamic-selection surface.
                    external_data_source=(
                        Action(
                            function=si.external_data_source.function or None,
                            parameters={
                                p.key: p.value
                                for p in si.external_data_source.parameters
                            },
                        )
                        if si._pb.HasField("external_data_source")
                        else None
                    ),
                    multi_select_max_selected_items=(
                        si.multi_select_max_selected_items or None
                    ),
                    multi_select_min_query_length=(
                        si.multi_select_min_query_length or None
                    ),
                )
            )
        elif which == "date_time_picker":
            dp = widget.date_time_picker
            widgets.append(
                DateTimePicker(
                    name=dp.name,
                    label=dp.label,
                    value_ms_epoch=dp.value_ms_epoch or None,
                    timezone_offset_date=dp.timezone_offset_date or None,
                )
            )
        elif which == "image":
            image = widget.image
            on_click: Action | OpenLink | None = None
            click_kind = image.on_click._pb.WhichOneof("data")
            if click_kind == "action":
                on_click = Action.from_proto(image.on_click.action)
            elif click_kind == "open_link":
                on_click = OpenLink.from_proto(image.on_click.open_link)
            widgets.append(
                Image(
                    image_url=image.image_url,
                    alt_text=image.alt_text or None,
                    on_click=on_click,
                )
            )
        else:
            widgets.append(RawWidget(_proto_json(widget)))
    return cls(header=proto.header or None, widgets=widgets)

to_proto_dict()

Build the section as a proto-plus dict (oneof dispatch).

Source code in src/chattice/cards/card.py

def to_proto_dict(self) -> dict[str, Any]:
    """Build the section as a proto-plus dict (oneof dispatch)."""
    data: dict[str, Any] = {}
    if self.header is not None:
        data["header"] = self.header
    data["widgets"] = [self._widget_dict(w) for w in self.widgets]
    return data

SelectionInput dataclass

A selection field.

NOTE: the installed SDK proto (google-apps-card 0.7.0) has no default- selection field, so the facade models exactly what the SDK supports.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class SelectionInput:
    """A selection field.

    NOTE: the installed SDK proto (google-apps-card 0.7.0) has no default-
    selection field, so the facade models exactly what the SDK supports.
    """

    name: str
    label: str
    items: Sequence[Mapping[str, str]] = field(default_factory=tuple)
    # Dynamic selections (enterprise autocomplete): Google's
    # external-data-source is an ACTION (a function the app serves for
    # suggestions) — the datasource logic itself is application code.
    external_data_source: Action | None = None
    multi_select_max_selected_items: int | None = None
    multi_select_min_query_length: int | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "items", tuple(self.items))

    def to_proto(self) -> ProtoSelectionInput:
        kwargs: dict[str, Any] = {
            "name": self.name,
            "label": self.label,
            "items": [dict(item) for item in self.items],
        }
        if self.external_data_source is not None:
            kwargs["external_data_source"] = self.external_data_source.to_proto()
        if self.multi_select_max_selected_items is not None:
            kwargs["multi_select_max_selected_items"] = (
                self.multi_select_max_selected_items
            )
        if self.multi_select_min_query_length is not None:
            kwargs["multi_select_min_query_length"] = self.multi_select_min_query_length
        return ProtoSelectionInput(**kwargs)

TextInput dataclass

A text input field.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class TextInput:
    """A text input field."""

    name: str
    label: str
    hint_text: str | None = None
    value: str | None = None
    validation: Validation | None = None

    def to_proto(self) -> ProtoTextInput:
        kwargs: dict[str, Any] = {"name": self.name, "label": self.label}
        if self.hint_text is not None:
            kwargs["hint_text"] = self.hint_text
        if self.value is not None:
            kwargs["value"] = self.value
        if self.validation is not None:
            kwargs["validation"] = self.validation.to_proto()
        return ProtoTextInput(**kwargs)

TextInputType

Bases: Enum

Documented Validation inputType values (SDK enum mirror).

Source code in src/chattice/cards/validation.py

class TextInputType(Enum):
    """Documented Validation inputType values (SDK enum mirror)."""

    TEXT = "TEXT"
    INTEGER = "INTEGER"
    FLOAT = "FLOAT"
    EMAIL = "EMAIL"
    EMOJI_PICKER = "EMOJI_PICKER"

    def to_proto(self) -> Any:
        """Map to the SDK Validation.InputType enum member."""
        return getattr(ProtoValidation.InputType, self.value)

to_proto()

Map to the SDK Validation.InputType enum member.

Source code in src/chattice/cards/validation.py

def to_proto(self) -> Any:
    """Map to the SDK Validation.InputType enum member."""
    return getattr(ProtoValidation.InputType, self.value)

TextParagraph dataclass

A paragraph of text.

Source code in src/chattice/cards/widgets.py

@dataclass(frozen=True, slots=True)
class TextParagraph:
    """A paragraph of text."""

    text: str
    max_lines: int | None = None

    def to_proto(self) -> ProtoTextParagraph:
        kwargs: dict[str, Any] = {"text": self.text}
        if self.max_lines is not None:
            kwargs["max_lines"] = self.max_lines
        return ProtoTextParagraph(**kwargs)

Validation dataclass

Validation rules for a form input.

Source code in src/chattice/cards/validation.py

@dataclass(frozen=True, slots=True)
class Validation:
    """Validation rules for a form input."""

    character_limit: int | None = None
    input_type: TextInputType | None = None

    def to_proto(self) -> ProtoValidation:
        """Build the SDK Validation proto."""
        kwargs: dict[str, Any] = {}
        if self.character_limit is not None:
            kwargs["character_limit"] = self.character_limit
        if self.input_type is not None:
            kwargs["input_type"] = self.input_type.to_proto()
        return ProtoValidation(**kwargs)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to the documented camelCase JSON."""
        return json.loads(json_format.MessageToJson(self.to_proto()._pb))  # type: ignore[no-any-return]

to_dict()

Serialize to the documented camelCase JSON.

Source code in src/chattice/cards/validation.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to the documented camelCase JSON."""
    return json.loads(json_format.MessageToJson(self.to_proto()._pb))  # type: ignore[no-any-return]

to_proto()

Build the SDK Validation proto.

Source code in src/chattice/cards/validation.py

def to_proto(self) -> ProtoValidation:
    """Build the SDK Validation proto."""
    kwargs: dict[str, Any] = {}
    if self.character_limit is not None:
        kwargs["character_limit"] = self.character_limit
    if self.input_type is not None:
        kwargs["input_type"] = self.input_type.to_proto()
    return ProtoValidation(**kwargs)

Typed form decoding: opt-in schemas above the existing typed FormInputs.

Google's common.formInputs values are ALREADY typed by the adapter (StringInput / DateInput / DateTimeInput / TimeInput / UnknownFormInput). FormModel adds an opt-in dataclass schema that maps widget names onto those typed values — no string flattening, no mandatory pydantic domain model, and no automatic translation of decode errors into dialog errors (Google's error response rules depend on the surface; the application decides how to answer).

Usage:

@dataclass
class ContactForm(FormModel):
    name: StringInput
    birthday: DateInput | None = None

@router.dialog_submit(ContactForm.filter())
async def submit(event: ActionEvent, form: ContactForm): ...

On a match the decoded instance is injected under the name form.

FormDecodeError

Bases: ValueError

Form inputs cannot be decoded into the typed model.

Source code in src/chattice/forms.py

class FormDecodeError(ValueError):
    """Form inputs cannot be decoded into the typed model."""

FormFilter

Decode-based filter: returns {"form": instance} or False.

Accepts the full event union that owns typed form_inputs — dialog ActionEvent AND App Home FormSubmitEvent — so the advertised form logic is reusable on App Home submits.

Source code in src/chattice/forms.py

class FormFilter:
    """Decode-based filter: returns ``{"form": instance}`` or False.

    Accepts the full event union that owns typed ``form_inputs`` —
    dialog ``ActionEvent`` AND App Home ``FormSubmitEvent`` — so the
    advertised form logic is reusable on App Home submits.
    """

    def __init__(self, model: type[FormModel]) -> None:
        self.model = model

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue:
        if not isinstance(event, (ActionEvent, FormSubmitEvent)):
            return False
        try:
            instance = self.model.from_form_inputs(event.form_inputs)
        except FormDecodeError as error:
            # A filter mismatch is silent by contract, but a decode failure
            # of an explicitly registered typed form is worth a debug line:
            # new users otherwise search for hours why a button "does nothing".
            _logger.debug(
                "form decode failed for %s: %s",
                self.model.__name__,
                error,
            )
            return False
        return {"form": instance}

FormModel

Base class for typed form-input models (dataclass subclasses).

Source code in src/chattice/forms.py

class FormModel:
    """Base class for typed form-input models (dataclass subclasses)."""

    @classmethod
    def _fields(cls) -> dict[str, type[object]]:
        if not hasattr(cls, "__dataclass_fields__"):
            raise TypeError(f"{cls.__name__} must be a dataclass FormModel")
        hints = get_type_hints(cls)
        return {
            field.name: hints[field.name]
            for field in dataclasses.fields(cast(Any, cls))
            if field.name in hints
        }

    @classmethod
    def from_form_inputs(cls, inputs: FormInputs) -> Self:
        """Decode typed form inputs into an instance.

        Missing optional fields fall back to their dataclass defaults;
        missing required fields raise FormDecodeError. A present value of
        the wrong input kind (e.g. DateInput where StringInput was
        declared) raises FormDecodeError.
        """
        fields = cls._fields()
        values: dict[str, object] = {}
        for name, target in fields.items():
            target = _unwrap_optional(target)
            if target not in _SUPPORTED:
                raise TypeError(
                    f"FormModel field {name!r} must be one of "
                    f"{[t.__name__ for t in _SUPPORTED]}"
                )
            if name not in inputs:
                if name in _required_field_names(cls):
                    raise FormDecodeError(
                        f"missing required form input {name!r} for {cls.__name__}"
                    )
                continue  # optional: dataclass default applies
            value = inputs[name]
            if not isinstance(value, target):
                raise FormDecodeError(
                    f"form input {name!r} is {type(value).__name__}, "
                    f"expected {target.__name__}"
                )
            values[name] = value
        return cls(**values)

    @classmethod
    def filter(cls) -> FormFilter:
        """An async filter matching when form inputs decode into this model.

        On a match the decoded instance is injected into the handler
        context under the name ``form``.
        """
        return FormFilter(cls)

filter() classmethod

An async filter matching when form inputs decode into this model.

On a match the decoded instance is injected into the handler context under the name form.

Source code in src/chattice/forms.py

@classmethod
def filter(cls) -> FormFilter:
    """An async filter matching when form inputs decode into this model.

    On a match the decoded instance is injected into the handler
    context under the name ``form``.
    """
    return FormFilter(cls)

from_form_inputs(inputs) classmethod

Decode typed form inputs into an instance.

Missing optional fields fall back to their dataclass defaults; missing required fields raise FormDecodeError. A present value of the wrong input kind (e.g. DateInput where StringInput was declared) raises FormDecodeError.

Source code in src/chattice/forms.py

@classmethod
def from_form_inputs(cls, inputs: FormInputs) -> Self:
    """Decode typed form inputs into an instance.

    Missing optional fields fall back to their dataclass defaults;
    missing required fields raise FormDecodeError. A present value of
    the wrong input kind (e.g. DateInput where StringInput was
    declared) raises FormDecodeError.
    """
    fields = cls._fields()
    values: dict[str, object] = {}
    for name, target in fields.items():
        target = _unwrap_optional(target)
        if target not in _SUPPORTED:
            raise TypeError(
                f"FormModel field {name!r} must be one of "
                f"{[t.__name__ for t in _SUPPORTED]}"
            )
        if name not in inputs:
            if name in _required_field_names(cls):
                raise FormDecodeError(
                    f"missing required form input {name!r} for {cls.__name__}"
                )
            continue  # optional: dataclass default applies
        value = inputs[name]
        if not isinstance(value, target):
            raise FormDecodeError(
                f"form input {name!r} is {type(value).__name__}, "
                f"expected {target.__name__}"
            )
        values[name] = value
    return cls(**values)

Authentication, capabilities, and outbound client

Storage-agnostic publication contract for local Card assets.

AssetPublisher

Bases: Protocol

Publish bytes and return an absolute HTTPS URL for a Card Image.

Source code in src/chattice/assets.py

class AssetPublisher(Protocol):
    """Publish bytes and return an absolute HTTPS URL for a Card Image."""

    async def publish(
        self,
        data: bytes,
        *,
        filename: str,
        content_type: str,
        namespace: str | None = None,
    ) -> str:
        """Publish one immutable byte snapshot."""
        ...

publish(data, *, filename, content_type, namespace=None) async

Publish one immutable byte snapshot.

Source code in src/chattice/assets.py

async def publish(
    self,
    data: bytes,
    *,
    filename: str,
    content_type: str,
    namespace: str | None = None,
) -> str:
    """Publish one immutable byte snapshot."""
    ...

Outgoing authentication providers (app / user modes).

AuthMode

Bases: Enum

The outgoing authentication identity class.

Source code in src/chattice/auth/providers.py

class AuthMode(Enum):
    """The outgoing authentication identity class."""

    APP = "app"
    USER = "user"
    NONE = "none"

CredentialsProvider

Bases: Protocol

Callable returning Google credentials valid at call time.

Source code in src/chattice/auth/providers.py

class CredentialsProvider(Protocol):
    """Callable returning Google credentials valid at call time."""

    def __call__(self) -> Credentials:
        """Return credentials valid at call time."""
        ...

__call__()

Return credentials valid at call time.

Source code in src/chattice/auth/providers.py

def __call__(self) -> Credentials:
    """Return credentials valid at call time."""
    ...

DelegatedUserCredentialsProvider dataclass

User-auth provider via Google Workspace Domain-Wide Delegation.

A service account granted domain-wide delegation impersonates a Workspace user through with_subject; Google treats the resulting credentials as USER authentication (media.upload, user-scoped operations) without per-user consent flows. ONE service-account JSON therefore serves both identities of a dual-identity Bot: the same file supplies the app provider directly and this provider with a subject. Requires the Workspace administrator to configure the delegation and OAuth scopes.

Source code in src/chattice/auth/providers.py

@dataclass(frozen=True, slots=True)
class DelegatedUserCredentialsProvider:
    """User-auth provider via Google Workspace Domain-Wide Delegation.

    A service account granted domain-wide delegation impersonates a
    Workspace user through ``with_subject``; Google treats the resulting
    credentials as USER authentication (media.upload, user-scoped
    operations) without per-user consent flows. ONE service-account JSON
    therefore serves both identities of a dual-identity Bot: the same
    file supplies the app provider directly and this provider with a
    ``subject``. Requires the Workspace administrator to configure the
    delegation and OAuth scopes.
    """

    provider: CredentialsProvider
    subject: str

    @classmethod
    def from_service_account_file(
        cls,
        path: str | Path,
        subject: str,
        *,
        scopes: list[str] | None = None,
    ) -> DelegatedUserCredentialsProvider:
        """Build from a service-account JSON file and a delegated subject."""
        return cls(
            provider=ServiceAccountCredentialsProvider.from_service_account_file(
                path, scopes=scopes if scopes is not None else [_USER_SCOPE]
            ),
            subject=subject,
        )

    def __call__(self) -> Credentials:
        credentials = self.provider()
        with_subject = getattr(credentials, "with_subject", None)
        if not callable(with_subject):
            raise ValueError(
                "these credentials do not support with_subject delegation; "
                "use service-account credentials with Domain-Wide Delegation "
                "configured in the Workspace admin console"
            )
        delegated = cast(Credentials, with_subject(self.subject))
        if delegated is None:
            raise ValueError(
                "these credentials do not support with_subject delegation; "
                "use service-account credentials with Domain-Wide Delegation "
                "configured in the Workspace admin console"
            )
        return delegated

from_service_account_file(path, subject, *, scopes=None) classmethod

Build from a service-account JSON file and a delegated subject.

Source code in src/chattice/auth/providers.py

@classmethod
def from_service_account_file(
    cls,
    path: str | Path,
    subject: str,
    *,
    scopes: list[str] | None = None,
) -> DelegatedUserCredentialsProvider:
    """Build from a service-account JSON file and a delegated subject."""
    return cls(
        provider=ServiceAccountCredentialsProvider.from_service_account_file(
            path, scopes=scopes if scopes is not None else [_USER_SCOPE]
        ),
        subject=subject,
    )

ServiceAccountCredentialsProvider dataclass

App-auth provider: lazy service-account credentials.

The JSON file/info is not read at construction — only when the provider is called (each call re-reads; the Bot calls it once).

Source code in src/chattice/auth/providers.py

@dataclass(frozen=True, slots=True)
class ServiceAccountCredentialsProvider:
    """App-auth provider: lazy service-account credentials.

    The JSON file/info is not read at construction — only when the
    provider is called (each call re-reads; the Bot calls it once).
    """

    credentials: Credentials | None = None
    file_path: str | None = None
    info: Mapping[str, Any] | None = None
    scopes: tuple[str, ...] = (CHAT_BOT_SCOPE,)

    @classmethod
    def from_service_account_file(
        cls, path: str | Path, scopes: list[str] | None = None
    ) -> ServiceAccountCredentialsProvider:
        """Build from a service-account JSON file path (lazily read)."""
        return cls(
            file_path=str(path),
            scopes=tuple(scopes) if scopes is not None else (CHAT_BOT_SCOPE,),
        )

    @classmethod
    def from_service_account_info(
        cls, info: Mapping[str, Any], scopes: list[str] | None = None
    ) -> ServiceAccountCredentialsProvider:
        """Build from an in-memory service-account info mapping."""
        return cls(
            info=dict(info),
            scopes=tuple(scopes) if scopes is not None else (CHAT_BOT_SCOPE,),
        )

    def __call__(self) -> Credentials:
        if self.credentials is not None:
            return self.credentials
        if self.file_path is not None:
            return _service_account_file(self.file_path, list(self.scopes))
        if self.info is not None:
            return _service_account_info(self.info, list(self.scopes))
        raise ValueError("ServiceAccountCredentialsProvider has no credentials source")

from_service_account_file(path, scopes=None) classmethod

Build from a service-account JSON file path (lazily read).

Source code in src/chattice/auth/providers.py

@classmethod
def from_service_account_file(
    cls, path: str | Path, scopes: list[str] | None = None
) -> ServiceAccountCredentialsProvider:
    """Build from a service-account JSON file path (lazily read)."""
    return cls(
        file_path=str(path),
        scopes=tuple(scopes) if scopes is not None else (CHAT_BOT_SCOPE,),
    )

from_service_account_info(info, scopes=None) classmethod

Build from an in-memory service-account info mapping.

Source code in src/chattice/auth/providers.py

@classmethod
def from_service_account_info(
    cls, info: Mapping[str, Any], scopes: list[str] | None = None
) -> ServiceAccountCredentialsProvider:
    """Build from an in-memory service-account info mapping."""
    return cls(
        info=dict(info),
        scopes=tuple(scopes) if scopes is not None else (CHAT_BOT_SCOPE,),
    )

UserCredentialsProvider dataclass

User-auth provider: authorized-user credentials with lazy refresh.

Token storage and OAuth code acquisition belong to the application. The refresh happens synchronously at call time (once, at lazy client creation in Bot); subsequent refreshes are handled inside the SDK.

Source code in src/chattice/auth/providers.py

@dataclass(frozen=True, slots=True)
class UserCredentialsProvider:
    """User-auth provider: authorized-user credentials with lazy refresh.

    Token storage and OAuth code acquisition belong to the application.
    The refresh happens synchronously at call time (once, at lazy client
    creation in Bot); subsequent refreshes are handled inside the SDK.
    """

    credentials: Credentials | Mapping[str, Any]
    refresh_before_call: bool = True

    def __call__(self) -> Credentials:
        credentials = self._credentials()
        if (
            self.refresh_before_call
            and credentials.expired
            and hasattr(credentials, "refresh")
        ):
            credentials.refresh(Request())  # type: ignore[no-untyped-call]
        return credentials

    def _credentials(self) -> Credentials:
        if isinstance(self.credentials, Mapping):
            return _user_credentials_info(self.credentials)
        return self.credentials

Capability model: response channel, outbound operations, preview features.

AuthPath dataclass

One identity-specific, any-of OAuth scope rule.

identity: the credential identity class (APP or USER). any_scope: any one of these scopes makes the path admissible. variant: execution variant; ADMIN adds the request_flag, IMPORT selects import scopes. request_flag: optional wire flag sent with the request (e.g. "use_admin_access").

Source code in src/chattice/capabilities/operations.py

@dataclass(frozen=True, slots=True)
class AuthPath:
    """One identity-specific, any-of OAuth scope rule.

    identity: the credential identity class (APP or USER).
    any_scope: any one of these scopes makes the path admissible.
    variant: execution variant; ADMIN adds the request_flag, IMPORT
        selects import scopes.
    request_flag: optional wire flag sent with the request (e.g.
        "use_admin_access").
    """

    identity: AuthMode
    any_scope: frozenset[str]
    variant: ExecutionVariant = ExecutionVariant.NORMAL
    request_flag: str | None = None

CapabilityNotSupported

Bases: RuntimeError

An operation was attempted without its required capability.

Source code in src/chattice/capabilities/matrix.py

class CapabilityNotSupported(RuntimeError):
    """An operation was attempted without its required capability."""

ExecutionVariant

Bases: StrEnum

Execution variant of an identity for an operation.

Source code in src/chattice/capabilities/operations.py

class ExecutionVariant(StrEnum):
    """Execution variant of an identity for an operation."""

    NORMAL = "normal"
    ADMIN = "admin"
    IMPORT = "import"

Operation

Bases: StrEnum

A Google Chat outbound operation.

Source code in src/chattice/capabilities/operations.py

class Operation(StrEnum):
    """A Google Chat outbound operation."""

    MESSAGES_CREATE = "messages.create"
    MESSAGES_GET = "messages.get"
    MESSAGES_LIST = "messages.list"
    MESSAGES_SEARCH = "messages.search"
    MESSAGES_UPDATE = "messages.update"
    MESSAGES_DELETE = "messages.delete"
    MEDIA_UPLOAD = "media.upload"
    MEDIA_DOWNLOAD = "media.download"
    ATTACHMENT_METADATA_GET = "spaces.messages.attachments.get"
    SPACES_GET = "spaces.get"
    SPACES_LIST = "spaces.list"
    SPACES_SEARCH = "spaces.search"
    SPACES_FIND_DIRECT_MESSAGE = "spaces.find_direct_message"
    SPACES_FIND_GROUP_CHATS = "spaces.find_group_chats"
    SPACES_SETUP = "spaces.setup"
    SPACES_CREATE = "spaces.create"
    SPACES_UPDATE = "spaces.update"
    SPACES_DELETE = "spaces.delete"
    MEMBERSHIPS_CREATE = "memberships.create"
    MEMBERSHIPS_GET = "memberships.get"
    MEMBERSHIPS_LIST = "memberships.list"
    MEMBERSHIPS_UPDATE = "memberships.update"
    MEMBERSHIPS_DELETE = "memberships.delete"
    REACTIONS_CREATE = "reactions.create"
    REACTIONS_LIST = "reactions.list"
    REACTIONS_DELETE = "reactions.delete"
    SPACE_READ_STATE_GET = "space_read_state.get"
    SPACE_READ_STATE_UPDATE = "space_read_state.update"
    THREAD_READ_STATE_GET = "thread_read_state.get"
    NOTIFICATION_SETTING_GET = "notification_setting.get"
    NOTIFICATION_SETTING_UPDATE = "notification_setting.update"

OperationRegistry

Lookup and local preflight over OperationSpecs.

Source code in src/chattice/capabilities/registry.py

class OperationRegistry:
    """Lookup and local preflight over OperationSpecs."""

    def __init__(self, specs: Iterable[OperationSpec]) -> None:
        self._specs: dict[Operation, OperationSpec] = {}
        for spec in specs:
            if spec.operation in self._specs:
                raise ValueError(f"Duplicate spec for {spec.operation}")
            self._specs[spec.operation] = spec

    def __contains__(self, operation: Operation) -> bool:
        return operation in self._specs

    def __iter__(self) -> Iterator[OperationSpec]:
        return iter(self._specs.values())

    def spec(self, operation: Operation) -> OperationSpec:
        try:
            return self._specs[operation]
        except KeyError:
            raise UnknownOperation(operation.value) from None

    def preflight(
        self,
        operation: Operation,
        *,
        identity: AuthMode | None,
        scopes: Iterable[str] | None = None,
        variant: ExecutionVariant = ExecutionVariant.NORMAL,
    ) -> bool:
        """Return whether the local configuration allows an attempt."""
        paths = [
            path
            for path in self.spec(operation).auth_paths
            if path.identity is identity and path.variant is variant
        ]
        if not paths:
            return False
        if scopes is None:
            return True
        known = frozenset(scopes)
        return any(path.any_scope & known for path in paths)

    def require(
        self,
        operation: Operation,
        *,
        identity: AuthMode | None,
        scopes: Iterable[str] | None = None,
        variant: ExecutionVariant = ExecutionVariant.NORMAL,
    ) -> None:
        """Raise CapabilityNotSupported when preflight fails."""
        if self.preflight(operation, identity=identity, scopes=scopes, variant=variant):
            return
        # Deferred import: matrix.py imports the registry, so the error
        # class resolves at call time to avoid a package cycle.
        from chattice.capabilities.matrix import CapabilityNotSupported

        spec = self.spec(operation)
        raise CapabilityNotSupported(
            f"{operation.name} is not supported in this configuration. "
            f"{spec.description}".rstrip()
        )

preflight(operation, *, identity, scopes=None, variant=ExecutionVariant.NORMAL)

Return whether the local configuration allows an attempt.

Source code in src/chattice/capabilities/registry.py

def preflight(
    self,
    operation: Operation,
    *,
    identity: AuthMode | None,
    scopes: Iterable[str] | None = None,
    variant: ExecutionVariant = ExecutionVariant.NORMAL,
) -> bool:
    """Return whether the local configuration allows an attempt."""
    paths = [
        path
        for path in self.spec(operation).auth_paths
        if path.identity is identity and path.variant is variant
    ]
    if not paths:
        return False
    if scopes is None:
        return True
    known = frozenset(scopes)
    return any(path.any_scope & known for path in paths)

require(operation, *, identity, scopes=None, variant=ExecutionVariant.NORMAL)

Raise CapabilityNotSupported when preflight fails.

Source code in src/chattice/capabilities/registry.py

def require(
    self,
    operation: Operation,
    *,
    identity: AuthMode | None,
    scopes: Iterable[str] | None = None,
    variant: ExecutionVariant = ExecutionVariant.NORMAL,
) -> None:
    """Raise CapabilityNotSupported when preflight fails."""
    if self.preflight(operation, identity=identity, scopes=scopes, variant=variant):
        return
    # Deferred import: matrix.py imports the registry, so the error
    # class resolves at call time to avoid a package cycle.
    from chattice.capabilities.matrix import CapabilityNotSupported

    spec = self.spec(operation)
    raise CapabilityNotSupported(
        f"{operation.name} is not supported in this configuration. "
        f"{spec.description}".rstrip()
    )

OperationSpec dataclass

Declarative local-preflight description of one operation.

google_method is the discovery-document method id used by the CI registry verifier. paginated marks SDK pager-backed methods; preview marks Developer Preview surface (explicit opt-in via Bot(enable_preview=True)); retry_policy is a human-readable label — the executor never runs its own retry loop, it passes the per-call RequestConfig.retry through to GAPIC.

Source code in src/chattice/capabilities/operations.py

@dataclass(frozen=True, slots=True)
class OperationSpec:
    """Declarative local-preflight description of one operation.

    google_method is the discovery-document method id used by the CI
    registry verifier. ``paginated`` marks SDK pager-backed methods;
    ``preview`` marks Developer Preview surface (explicit opt-in via
    ``Bot(enable_preview=True)``); ``retry_policy`` is a human-readable
    label — the executor never runs its own retry loop, it passes the
    per-call ``RequestConfig.retry`` through to GAPIC.
    """

    operation: Operation
    google_method: str
    auth_paths: tuple[AuthPath, ...]
    description: str = ""
    paginated: bool = False
    preview: bool = False
    retry_policy: str | None = None

PreviewCapabilities

Explicit enrollment set for typed Developer Preview routing.

Source code in src/chattice/capabilities/matrix.py

class PreviewCapabilities:
    """Explicit enrollment set for typed Developer Preview routing."""

    def __init__(self, features: Iterable[PreviewFeature] = ()) -> None:
        self._features = frozenset(features)

    def __contains__(self, feature: PreviewFeature) -> bool:
        return feature in self._features

    def __iter__(self) -> Iterator[PreviewFeature]:
        return iter(self._features)

    def require(self, feature: PreviewFeature) -> None:
        if feature not in self._features:
            raise CapabilityNotSupported(
                f"{feature.name} is a Developer Preview feature; enable it "
                "explicitly on Dispatcher(preview_features=...)."
            )

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, PreviewCapabilities):
            return NotImplemented
        return self._features == other._features

PreviewFeature

Bases: Enum

Developer Preview Google features (stability flags, not auth).

Sources: Google Chat release notes. These exist so documentation and code reference ONE list of preview surfaces.

Source code in src/chattice/capabilities/matrix.py

class PreviewFeature(Enum):
    """Developer Preview Google features (stability flags, not auth).

    Sources: Google Chat release notes. These exist
    so documentation and code reference ONE list of preview surfaces.
    """

    MESSAGE_ACTION = auto()  # Legacy opt-in retained for compatibility; now GA
    REPLACE_CARDS = auto()  # messages.replaceCards
    USER_AUTH_CARDS = auto()  # card creation with user auth
    CUSTOMER_LEVEL_SUBSCRIPTIONS = auto()  # Workspace Events customer-level
    PINNED_MESSAGES = auto()  # chat.spaces.pins
    RELEVANCE_SEARCH_ORDERING = auto()  # messages/spaces search relevance

ResponseCapabilities

An immutable response-channel capability set with require() guards.

Source code in src/chattice/capabilities/matrix.py

class ResponseCapabilities:
    """An immutable response-channel capability set with require() guards."""

    def __init__(
        self, capabilities: frozenset[ResponseCapability] | set[ResponseCapability]
    ) -> None:
        self._capabilities = frozenset(capabilities)

    def __contains__(self, capability: ResponseCapability) -> bool:
        return capability in self._capabilities

    def __iter__(self) -> Iterator[ResponseCapability]:
        return iter(self._capabilities)

    def require(self, capability: ResponseCapability) -> None:
        """Raise CapabilityNotSupported when the capability is missing."""
        if capability in self._capabilities:
            return
        detail = _RESPONSE_DESCRIPTIONS.get(capability, "")
        message = f"{capability.name} is not supported in this configuration. {detail}"
        raise CapabilityNotSupported(message.rstrip())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ResponseCapabilities):
            return NotImplemented
        return self._capabilities == other._capabilities

    @classmethod
    def resolve(
        cls, *, transport: str = "http", event: Event | None = None
    ) -> ResponseCapabilities:
        """Resolve the response-channel capabilities for a transport + event.

        HTTP provides the sync response channel and dialogs; Pub/Sub push
        provides NO response channel (ack-only). Event-specific response
        rules are derived from the concrete event, never guessed.
        """
        if transport not in _TRANSPORTS:
            raise ValueError(
                f"Unknown transport {transport!r}; supported: {', '.join(_TRANSPORTS)}"
            )
        if transport == "pubsub":
            return cls(set())
        capabilities: set[ResponseCapability] = {ResponseCapability.SYNC_RESPONSE}
        # DIALOGS only for events that can actually open a dialog
        # (command / REQUEST_DIALOG action) — a plain Message advertising
        # DIALOGS produced 500s at serialization.
        if can_open_dialog(event):
            capabilities.add(ResponseCapability.DIALOGS)
        if isinstance(event, (AppHomeEvent, FormSubmitEvent)):
            capabilities.add(ResponseCapability.APP_HOME)
        if isinstance(event, WidgetUpdatedEvent):
            capabilities.add(ResponseCapability.UPDATE_WIDGET)
        if isinstance(event, ActionEvent):
            if event.sender_type == "BOT":
                capabilities.add(ResponseCapability.CARD_UPDATE_BOT)
            elif event.sender_type == "HUMAN":
                capabilities.add(ResponseCapability.CARD_UPDATE_USER)
        if isinstance(event, MessageEvent) and event.matched_url is not None:
            capabilities.add(ResponseCapability.CARD_UPDATE_USER)
        return cls(capabilities)

require(capability)

Raise CapabilityNotSupported when the capability is missing.

Source code in src/chattice/capabilities/matrix.py

def require(self, capability: ResponseCapability) -> None:
    """Raise CapabilityNotSupported when the capability is missing."""
    if capability in self._capabilities:
        return
    detail = _RESPONSE_DESCRIPTIONS.get(capability, "")
    message = f"{capability.name} is not supported in this configuration. {detail}"
    raise CapabilityNotSupported(message.rstrip())

resolve(*, transport='http', event=None) classmethod

Resolve the response-channel capabilities for a transport + event.

HTTP provides the sync response channel and dialogs; Pub/Sub push provides NO response channel (ack-only). Event-specific response rules are derived from the concrete event, never guessed.

Source code in src/chattice/capabilities/matrix.py

@classmethod
def resolve(
    cls, *, transport: str = "http", event: Event | None = None
) -> ResponseCapabilities:
    """Resolve the response-channel capabilities for a transport + event.

    HTTP provides the sync response channel and dialogs; Pub/Sub push
    provides NO response channel (ack-only). Event-specific response
    rules are derived from the concrete event, never guessed.
    """
    if transport not in _TRANSPORTS:
        raise ValueError(
            f"Unknown transport {transport!r}; supported: {', '.join(_TRANSPORTS)}"
        )
    if transport == "pubsub":
        return cls(set())
    capabilities: set[ResponseCapability] = {ResponseCapability.SYNC_RESPONSE}
    # DIALOGS only for events that can actually open a dialog
    # (command / REQUEST_DIALOG action) — a plain Message advertising
    # DIALOGS produced 500s at serialization.
    if can_open_dialog(event):
        capabilities.add(ResponseCapability.DIALOGS)
    if isinstance(event, (AppHomeEvent, FormSubmitEvent)):
        capabilities.add(ResponseCapability.APP_HOME)
    if isinstance(event, WidgetUpdatedEvent):
        capabilities.add(ResponseCapability.UPDATE_WIDGET)
    if isinstance(event, ActionEvent):
        if event.sender_type == "BOT":
            capabilities.add(ResponseCapability.CARD_UPDATE_BOT)
        elif event.sender_type == "HUMAN":
            capabilities.add(ResponseCapability.CARD_UPDATE_USER)
    if isinstance(event, MessageEvent) and event.matched_url is not None:
        capabilities.add(ResponseCapability.CARD_UPDATE_USER)
    return cls(capabilities)

ResponseCapability

Bases: Enum

What the synchronous ingress response channel can do.

Source code in src/chattice/capabilities/matrix.py

class ResponseCapability(Enum):
    """What the synchronous ingress response channel can do."""

    SYNC_RESPONSE = auto()
    DIALOGS = auto()
    APP_HOME = auto()
    CARD_UPDATE_BOT = auto()
    CARD_UPDATE_USER = auto()
    UPDATE_WIDGET = auto()

can_open_dialog(event)

Return whether an event may open a dialog.

Commands always can; actions only when Google delivered them WITH REQUEST_DIALOG metadata. SUBMIT/CANCEL actions cannot return a new dialog, and a plain Message cannot open one either.

Source code in src/chattice/capabilities/matrix.py

def can_open_dialog(event: Event | None) -> bool:
    """Return whether an event may open a dialog.

    Commands always can; actions only when Google delivered them WITH
    REQUEST_DIALOG metadata. SUBMIT/CANCEL actions cannot return a new
    dialog, and a plain Message cannot open one either.
    """
    if isinstance(event, CommandEvent):
        return True
    if isinstance(event, ActionEvent) and event.dialog is not None:
        return event.dialog.type == DialogEventType.REQUEST_DIALOG
    return False

scopes(*names)

Build fully qualified Google OAuth scopes from short names.

Source code in src/chattice/capabilities/operations.py

def scopes(*names: str) -> frozenset[str]:
    """Build fully qualified Google OAuth scopes from short names."""
    return frozenset(GOOGLE_AUTH_SCOPE_PREFIX + name for name in names)

High-level async Chat API client.

Bot

Authenticated outgoing Google Chat operations.

The SDK client is created lazily on the first call so that Bot() can be constructed before credentials are available (e.g. in app factories).

Source code in src/chattice/client/bot.py

class Bot:
    """Authenticated outgoing Google Chat operations.

    The SDK client is created lazily on the first call so that Bot() can be
    constructed before credentials are available (e.g. in app factories).
    """

    def __init__(
        self,
        credentials: Credentials | None = None,
        *,
        credentials_provider: CredentialsProvider | None = None,
        app_credentials_provider: CredentialsProvider | None = None,
        user_credentials_provider: CredentialsProvider | None = None,
        asset_publisher: AssetPublisher | None = None,
        auth_mode: AuthMode | None = None,
        transport: ChatServiceTransport | None = None,
        enable_preview: bool = False,
    ) -> None:
        has_dual_providers = (
            app_credentials_provider is not None
            or user_credentials_provider is not None
        )
        if has_dual_providers:
            if credentials is not None or credentials_provider is not None:
                raise ValueError(
                    "app_credentials_provider/user_credentials_provider cannot "
                    "be combined with credentials/credentials_provider"
                )
            if auth_mode is not None:
                raise ValueError(
                    "auth_mode is implied by the app/user credential "
                    "providers; pass it only for a single-identity Bot"
                )
        self._credentials = credentials
        # Explicit app/user providers take precedence; credentials_provider
        # covers single-identity construction.
        self._credentials_provider = credentials_provider or app_credentials_provider
        self._user_credentials_provider = user_credentials_provider
        self._auth_mode = auth_mode
        self._transport = transport
        # Preview gating: registry is the source of truth for WHICH
        # operations are preview; this flag is the user's explicit
        # opt-in (per Bot instance).
        self._preview_enabled = enable_preview
        if asset_publisher is not None and not callable(
            getattr(asset_publisher, "publish", None)
        ):
            raise TypeError("asset_publisher must define an async publish() method")
        self._asset_publisher = asset_publisher
        self._client: ChatServiceAsyncClient | None = None
        self._resolved_credentials: Credentials | None = None
        self._resolved_set = False
        self._resolved_user_credentials: Credentials | None = None
        self._resolved_user_set = False
        self._closed = False
        # single-flight tasks — concurrent first calls share ONE
        # credential resolution and ONE client construction (a shared
        # Task, not a lock held across provider code).
        self._credential_task: asyncio.Task[Credentials | None] | None = None
        self._user_credential_task: asyncio.Task[Credentials | None] | None = None
        self._init_task: asyncio.Task[ChatServiceAsyncClient] | None = None
        # The USER identity has its own cached client: attachment messages
        # must be created with the SAME USER credentials that uploaded
        # them (an APP-authenticated create cannot consume
        # a USER-uploaded attachment — Google rejects the handoff).
        self._user_client: ChatServiceAsyncClient | None = None
        self._user_init_task: asyncio.Task[ChatServiceAsyncClient] | None = None
        # Explicit identity facades (ADR-012). Roots are cached: the
        # facade must stay stable across accesses and cost no I/O.
        self._app_root = IdentityNamespace(self, AuthMode.APP)
        self._user_root = IdentityNamespace(self, AuthMode.USER, allow_setup=True)
        self._raw_namespace = RawClients(self)
        self._executor: OperationExecutor[object] = OperationExecutor(self)

    @property
    def app(self) -> IdentityNamespace:
        """Resource clients bound to the APP identity."""
        return self._app_root

    @property
    def user(self) -> IdentityNamespace:
        """Resource clients bound to the USER identity."""
        return self._user_root

    @property
    def raw(self) -> RawClients:
        """Async raw SDK access split by identity."""
        return self._raw_namespace

    async def warmup(self, *, app: bool = True, user: bool = False) -> None:
        """Resolve credentials and build clients before the first event.

        May resolve credentials and prepare clients/transports; never
        performs business API calls. Useful to move the slow first
        outbound request (credential resolution, IAM, gRPC channel)
        into startup.
        """
        if app:
            await self._get_client_async()
        if user:
            await self._get_user_client_async()

    def _classify(self, credentials: Credentials | None) -> AuthMode | None:
        if credentials is None:
            return None
        if getattr(credentials, "_subject", None):
            # Domain-wide delegation: a service account impersonating a
            # user (with_subject) acts as USER authentication.
            return AuthMode.USER
        if hasattr(credentials, "signer"):  # service account
            return AuthMode.APP
        if getattr(credentials, "refresh_token", None):
            return AuthMode.USER
        return None

    @property
    def auth_mode(self) -> AuthMode | None:
        """The outgoing auth mode: explicit, or classified from credentials.

        Synchronous classification; the async Bot methods use
        ``_auth_mode_async`` so blocking providers never run on the loop.
        """
        if self._auth_mode is not None:
            return self._auth_mode
        return self._classify(self._resolve_credentials())

    async def close(self) -> None:
        """Close the underlying SDK transport (idempotent, awaitable).

        Close is linearizable with initialization. Once close
        begins, NO client may be published; if construction already
        completed, its transport is closed exactly once before close
        returns. The async gRPC transport's closer is itself awaitable —
        it is AWAITED here (a plain sync call would leak the channel).
        Safe to call multiple times; after close the client must not be
        used. Also available as ``async with Bot(...)``.
        """
        if self._closed:
            return
        self._closed = True
        for task in (
            self._credential_task,
            self._user_credential_task,
            self._init_task,
            self._user_init_task,
        ):
            if task is None:
                continue
            # In-flight resolution/construction either completes and
            # publishes (then we close it below) or observes _closed and
            # raises — both are deterministic; shield keeps OUR
            # cancellation from breaking the shared task for other
            # waiters.
            try:
                await asyncio.shield(task)
            except asyncio.CancelledError:
                raise
            except ChatAPIError:
                pass  # aborted by our own close — nothing to close
        # Close every initialized client exactly once. The injected test
        # transport may back both clients; each transport is closed once.
        closed_transports: set[object] = set()
        for client in (self._client, self._user_client):
            if client is None:
                continue
            transport = client.transport
            if transport in closed_transports:
                continue
            closed_transports.add(transport)
            closer = getattr(transport, "close", None)
            if closer is None:
                continue
            result = closer()
            if inspect.isawaitable(result):
                await result

    async def __aenter__(self) -> Bot:
        return self

    async def __aexit__(
        self,
        exc_type: object,
        exc_value: object,
        traceback: object,
    ) -> None:
        await self.close()

    async def _auth_mode_async(self) -> AuthMode | None:
        """Off-loop variant of :attr:`auth_mode` for async call paths."""
        if self._auth_mode is not None:
            return self._auth_mode
        return self._classify(await self._resolve_credentials_async())

    def _get_client(self) -> ChatServiceAsyncClient:
        if self._closed:
            raise ChatAPIError("Bot is closed; create a new instance")
        client = self._client
        if client is not None:
            return client
        return self._build_client(self._resolve_credentials())

    async def _get_client_async(self) -> ChatServiceAsyncClient:
        """Single-flight async client initialization.

        Credential providers may perform blocking I/O (file reads, token
        refresh); the async path runs them in a worker thread. Concurrent
        first calls share ONE construction task. A failed construction is
        not cached — the next call retries (provider errors stay
        retryable, the pinned contract).
        """
        client = self._client
        if client is not None:
            return client
        task = self._init_task
        if task is None:
            task = asyncio.create_task(self._initialize_client())
            self._init_task = task
        try:
            # Shield: cancellation of a WAITER must not kill the shared
            # construction for everyone else.
            return await asyncio.shield(task)
        except asyncio.CancelledError:
            raise
        except Exception:
            if self._init_task is task:
                self._init_task = None
            raise

    async def _initialize_client(self) -> ChatServiceAsyncClient:
        """Resolve credentials and build the client; never publish after close."""
        if self._closed:
            raise ChatAPIError("Bot is closed; create a new instance")
        credentials = await self._resolve_credentials_async()
        if self._closed:
            # close() began while the provider ran off-loop: the client
            # must NOT be published after terminal close.
            raise ChatAPIError("Bot is closed; create a new instance")
        return self._build_client(credentials)

    def _build_client(self, credentials: Credentials | None) -> ChatServiceAsyncClient:
        if self._transport is not None:
            # SDK rule: a transport instance carries its own credentials;
            # passing credentials alongside raises ValueError.
            self._client = ChatServiceAsyncClient(transport=self._transport)
            return self._client
        if credentials is None:
            raise ChatAPIError(
                "Bot has no credentials; pass google.auth credentials "
                "or a credentials_provider to Bot(...)"
            )
        self._client = ChatServiceAsyncClient(
            credentials=credentials,
            transport=_GRPC_ASYNCIO,
        )
        return self._client

    def _build_user_client(self, credentials: Credentials) -> ChatServiceAsyncClient:
        """Build the cached USER Chat client.

        Mirrors the APP client construction: an injected transport (used
        by the testing toolkit) carries no identity, so the same fake
        transport may back both clients in tests. In production the USER
        client gets its own real gRPC-asyncio transport.
        """
        if self._transport is not None:
            self._user_client = ChatServiceAsyncClient(transport=self._transport)
            return self._user_client
        self._user_client = ChatServiceAsyncClient(
            credentials=credentials,
            transport=_GRPC_ASYNCIO,
        )
        return self._user_client

    async def _get_user_client_async(self) -> ChatServiceAsyncClient:
        """Single-flight async USER client initialization.

        Same contract as the APP path: concurrent first attachment sends
        share ONE construction task, a waiter's cancellation never kills
        the shared construction, and a failed construction is not cached
        (provider errors stay retryable).
        """
        client = self._user_client
        if client is not None:
            return client
        task = self._user_init_task
        if task is None:
            task = asyncio.create_task(self._initialize_user_client())
            self._user_init_task = task
        try:
            return await asyncio.shield(task)
        except asyncio.CancelledError:
            raise
        except Exception:
            if self._user_init_task is task:
                self._user_init_task = None
            raise

    async def _initialize_user_client(self) -> ChatServiceAsyncClient:
        """Resolve USER credentials and build the client; never publish after close."""
        if self._closed:
            raise ChatAPIError("Bot is closed; create a new instance")
        credentials = await self._resolve_user_credentials_async()
        if credentials is None:
            raise CapabilityNotSupported(
                "Sending message attachments requires USER authentication for "
                "both media.upload and messages.create. Configure "
                "user_credentials_provider=... — UserCredentialsProvider, or "
                "DelegatedUserCredentialsProvider for domain-wide delegation."
            )
        if self._closed:
            # close() began while the provider ran off-loop: the client
            # must NOT be published after terminal close.
            raise ChatAPIError("Bot is closed; create a new instance")
        return self._build_user_client(credentials)

    def _resolve_credentials(self) -> Credentials | None:
        """Resolve credentials once (the provider is called a single time).

        A provider failure is NOT cached: the flag is set only on success,
        so the error re-raises on every attempt instead of silently
        degrading to None (which would disable the capability guards).
        """
        if not self._resolved_set:
            if self._closed:
                raise ChatAPIError("Bot is closed; create a new instance")
            if self._credentials_provider is not None:
                self._resolved_credentials = self._credentials_provider()
            else:
                self._resolved_credentials = self._credentials
            self._resolved_set = True
        return self._resolved_credentials

    async def _resolve_credentials_async(self) -> Credentials | None:
        """Async-safe single-flight credential resolution.

        The provider is called exactly ONCE even under concurrent first
        sends (the previous lock-free path raced on the resolve flag). A
        provider failure is not cached: the shared task is dropped so the
        next attempt re-invokes the provider.
        """
        task = self._credential_task
        if task is None:
            task = asyncio.create_task(self._resolve_credentials_once())
            self._credential_task = task
        try:
            return await asyncio.shield(task)
        except asyncio.CancelledError:
            raise
        except Exception:
            if self._credential_task is task:
                self._credential_task = None
            raise

    async def _resolve_credentials_once(self) -> Credentials | None:
        if self._closed:
            raise ChatAPIError("Bot is closed; create a new instance")
        if self._resolved_set:
            return self._resolved_credentials
        if self._credentials_provider is not None:
            self._resolved_credentials = await asyncio.to_thread(
                self._credentials_provider
            )
        else:
            self._resolved_credentials = self._credentials
        self._resolved_set = True
        return self._resolved_credentials

    async def _resolve_user_credentials_async(self) -> Credentials | None:
        """Async-safe single-flight USER identity resolution."""
        task = self._user_credential_task
        if task is None:
            task = asyncio.create_task(self._resolve_user_credentials_once())
            self._user_credential_task = task
        try:
            return await asyncio.shield(task)
        except asyncio.CancelledError:
            raise
        except Exception:
            if self._user_credential_task is task:
                self._user_credential_task = None
            raise

    async def _resolve_user_credentials_once(self) -> Credentials | None:
        if self._closed:
            raise ChatAPIError("Bot is closed; create a new instance")
        if self._resolved_user_set:
            return self._resolved_user_credentials
        if self._user_credentials_provider is not None:
            self._resolved_user_credentials = await asyncio.to_thread(
                self._user_credentials_provider
            )
        else:
            single = await self._resolve_credentials_async()
            self._resolved_user_credentials = (
                single if self._classify(single) is AuthMode.USER else None
            )
        self._resolved_user_set = True
        return self._resolved_user_credentials

    async def _preflight_operation(
        self,
        operation: Operation,
        *,
        identity: AuthMode,
        variant: ExecutionVariant = ExecutionVariant.NORMAL,
        registry: OperationRegistry | None = None,
    ) -> None:
        """Registry preflight for an explicit identity (ADR-012).

        Resolves the identity's credentials, classifies the mode and
        known scopes, then delegates to the registry. A mismatch
        between requested and resolved identity fails, never silently
        switches.
        """
        credentials = (
            await self._resolve_user_credentials_async()
            if identity is AuthMode.USER
            else await self._resolve_credentials_async()
        )
        if credentials is None:
            # Fail closed for BOTH identities: a call with no credentials
            # is deterministically invalid and must never reach the
            # transport (with a custom transport the SDK client would
            # happily issue an unauthenticated request).
            if identity is AuthMode.USER:
                raise CapabilityNotSupported(
                    "USER identity is not configured on this Bot"
                )
            raise CapabilityNotSupported(
                "The APP identity has no credentials; pass google.auth "
                "credentials or a credentials_provider to Bot(...)"
            )
        mode = self._classify(credentials)
        if mode is None:
            # Unclassifiable credentials: local preflight cannot decide,
            # the attempt proceeds and the server remains the authority.
            return
        if mode is not identity:
            raise CapabilityNotSupported(
                f"Requested {identity.name} identity but resolved "
                f"credentials are {mode.name}"
            )
        (registry or REGISTRY).require(
            operation,
            identity=mode,
            scopes=_credential_scopes(mode, credentials),
            variant=variant,
        )

app property

Resource clients bound to the APP identity.

auth_mode property

The outgoing auth mode: explicit, or classified from credentials.

Synchronous classification; the async Bot methods use _auth_mode_async so blocking providers never run on the loop.

raw property

Async raw SDK access split by identity.

user property

Resource clients bound to the USER identity.

close() async

Close the underlying SDK transport (idempotent, awaitable).

Close is linearizable with initialization. Once close begins, NO client may be published; if construction already completed, its transport is closed exactly once before close returns. The async gRPC transport's closer is itself awaitable — it is AWAITED here (a plain sync call would leak the channel). Safe to call multiple times; after close the client must not be used. Also available as async with Bot(...).

Source code in src/chattice/client/bot.py

async def close(self) -> None:
    """Close the underlying SDK transport (idempotent, awaitable).

    Close is linearizable with initialization. Once close
    begins, NO client may be published; if construction already
    completed, its transport is closed exactly once before close
    returns. The async gRPC transport's closer is itself awaitable —
    it is AWAITED here (a plain sync call would leak the channel).
    Safe to call multiple times; after close the client must not be
    used. Also available as ``async with Bot(...)``.
    """
    if self._closed:
        return
    self._closed = True
    for task in (
        self._credential_task,
        self._user_credential_task,
        self._init_task,
        self._user_init_task,
    ):
        if task is None:
            continue
        # In-flight resolution/construction either completes and
        # publishes (then we close it below) or observes _closed and
        # raises — both are deterministic; shield keeps OUR
        # cancellation from breaking the shared task for other
        # waiters.
        try:
            await asyncio.shield(task)
        except asyncio.CancelledError:
            raise
        except ChatAPIError:
            pass  # aborted by our own close — nothing to close
    # Close every initialized client exactly once. The injected test
    # transport may back both clients; each transport is closed once.
    closed_transports: set[object] = set()
    for client in (self._client, self._user_client):
        if client is None:
            continue
        transport = client.transport
        if transport in closed_transports:
            continue
        closed_transports.add(transport)
        closer = getattr(transport, "close", None)
        if closer is None:
            continue
        result = closer()
        if inspect.isawaitable(result):
            await result

warmup(*, app=True, user=False) async

Resolve credentials and build clients before the first event.

May resolve credentials and prepare clients/transports; never performs business API calls. Useful to move the slow first outbound request (credential resolution, IAM, gRPC channel) into startup.

Source code in src/chattice/client/bot.py

async def warmup(self, *, app: bool = True, user: bool = False) -> None:
    """Resolve credentials and build clients before the first event.

    May resolve credentials and prepare clients/transports; never
    performs business API calls. Useful to move the slow first
    outbound request (credential resolution, IAM, gRPC channel)
    into startup.
    """
    if app:
        await self._get_client_async()
    if user:
        await self._get_user_client_async()

ChatAPIError

Bases: Exception

An outgoing Chat API call failed.

SDK failures preserve the original error as cause (raise with from error); framework-raised errors (e.g. missing credentials) have no SDK cause, and the properties below return None for them.

Source code in src/chattice/client/errors.py

class ChatAPIError(Exception):
    """An outgoing Chat API call failed.

    SDK failures preserve the original error as __cause__ (raise with
    ``from error``); framework-raised errors (e.g. missing credentials)
    have no SDK cause, and the properties below return None for them.
    """

    @property
    def cause(self) -> api_core_exceptions.GoogleAPICallError | None:
        """The original SDK error, or None for framework-raised errors."""
        cause = self.__cause__
        if isinstance(cause, api_core_exceptions.GoogleAPICallError):
            return cause
        return None

    @property
    def code(self) -> int | None:
        """HTTP status code carried by the SDK error, or None."""
        cause = self.cause
        return cause.code if cause is not None else None

    @property
    def details(self) -> object:
        """Error details carried by the SDK error, or None."""
        cause = self.cause
        return cause.details if cause is not None else None

cause property

The original SDK error, or None for framework-raised errors.

code property

HTTP status code carried by the SDK error, or None.

details property

Error details carried by the SDK error, or None.

ChatAlreadyExistsError

Bases: ChatAPIError

The requested resource already exists (409 / already-exists).

Source code in src/chattice/client/errors.py

class ChatAlreadyExistsError(ChatAPIError):
    """The requested resource already exists (409 / already-exists)."""

ChatInvalidArgumentError

Bases: ChatAPIError

The request was rejected by validation.

Source code in src/chattice/client/errors.py

class ChatInvalidArgumentError(ChatAPIError):
    """The request was rejected by validation."""

ChatNotFoundError

Bases: ChatAPIError

The target resource does not exist.

Source code in src/chattice/client/errors.py

class ChatNotFoundError(ChatAPIError):
    """The target resource does not exist."""

ChatPermissionDeniedError

Bases: ChatAPIError

The app lacks permission (e.g. not a member of the space).

Source code in src/chattice/client/errors.py

class ChatPermissionDeniedError(ChatAPIError):
    """The app lacks permission (e.g. not a member of the space)."""

ChatRateLimitError

Bases: ChatAPIError

Quota exhausted or 429 response; retry only per the app's policy.

Source code in src/chattice/client/errors.py

class ChatRateLimitError(ChatAPIError):
    """Quota exhausted or 429 response; retry only per the app's policy."""

ChatServiceUnavailableError

Bases: ChatAPIError

Transient Chat API unavailability (5xx).

DeadlineExceeded (a GatewayTimeout subclass) also maps here.

Source code in src/chattice/client/errors.py

class ChatServiceUnavailableError(ChatAPIError):
    """Transient Chat API unavailability (5xx).

    ``DeadlineExceeded`` (a ``GatewayTimeout`` subclass) also maps here.
    """

ChatUnauthenticatedError

Bases: ChatAPIError

The credentials were rejected.

Source code in src/chattice/client/errors.py

class ChatUnauthenticatedError(ChatAPIError):
    """The credentials were rejected."""

CredentialsProvider

Bases: Protocol

Callable returning Google credentials valid at call time.

Source code in src/chattice/auth/providers.py

class CredentialsProvider(Protocol):
    """Callable returning Google credentials valid at call time."""

    def __call__(self) -> Credentials:
        """Return credentials valid at call time."""
        ...

__call__()

Return credentials valid at call time.

Source code in src/chattice/auth/providers.py

def __call__(self) -> Credentials:
    """Return credentials valid at call time."""
    ...

RawClients

Explicit raw client access: APP or USER identity.

Each accessor builds and serves exactly the identity it names, so dual-auth-sensitive operations never depend on an implicit primary identity.

Source code in src/chattice/client/bot.py

class RawClients:
    """Explicit raw client access: APP or USER identity.

    Each accessor builds and serves exactly the identity it names, so
    dual-auth-sensitive operations never depend on an implicit primary
    identity.
    """

    def __init__(self, bot: Bot) -> None:
        self._bot = bot

    async def app(self) -> ChatServiceAsyncClient:
        """The raw APP-authenticated Chat client."""
        return await self._bot._get_client_async()

    async def user(self) -> ChatServiceAsyncClient:
        """The raw USER-authenticated Chat client.

        Raises ``CapabilityNotSupported`` when no USER credentials are
        configured on the Bot.
        """
        return await self._bot._get_user_client_async()

app() async

The raw APP-authenticated Chat client.

Source code in src/chattice/client/bot.py

async def app(self) -> ChatServiceAsyncClient:
    """The raw APP-authenticated Chat client."""
    return await self._bot._get_client_async()

user() async

The raw USER-authenticated Chat client.

Raises CapabilityNotSupported when no USER credentials are configured on the Bot.

Source code in src/chattice/client/bot.py

async def user(self) -> ChatServiceAsyncClient:
    """The raw USER-authenticated Chat client.

    Raises ``CapabilityNotSupported`` when no USER credentials are
    configured on the Bot.
    """
    return await self._bot._get_user_client_async()

wrap_api_error(error)

Map an SDK error to its framework subtype (raise the result with 'from').

Source code in src/chattice/client/errors.py

def wrap_api_error(error: api_core_exceptions.GoogleAPICallError) -> ChatAPIError:
    """Map an SDK error to its framework subtype (raise the result with 'from')."""
    already_exists = getattr(api_core_exceptions, "AlreadyExists", ())
    conflict = getattr(api_core_exceptions, "Conflict", ())
    # ``Aborted`` inherits from ``Conflict`` but represents a different
    # retryable condition, so only map the concrete HTTP 409 error classes.
    if type(error) is already_exists or type(error) is conflict:
        return ChatAlreadyExistsError(str(error.message))
    for error_types, wrapper in _WRAPPERS:
        if isinstance(error, error_types):
            return wrapper(str(error.message))
    return ChatAPIError(str(error.message))

State and reliability

Finite-state machine primitives.

BaseStorage

Bases: Protocol

Storage contract implemented by MemoryStorage and RedisStorage.

Source code in src/chattice/fsm/storage.py

class BaseStorage(Protocol):
    """Storage contract implemented by MemoryStorage and RedisStorage."""

    async def get_state(self, key: StorageKey) -> str | None: ...
    async def set_state(self, key: StorageKey, state: str | None) -> None: ...
    async def get_data(self, key: StorageKey) -> dict[str, Any]: ...
    async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None: ...
    async def update_data(
        self, key: StorageKey, partial: Mapping[str, Any]
    ) -> dict[str, Any]: ...
    async def finish(self, key: StorageKey) -> None: ...

BaseStorageFromRecord

Bases: BaseStorage

Serve the six-method BaseStorage contract over a record store.

Transitions use compare-and-set; a concurrent modification raises FSMRecordConflict instead of silently losing data.

Source code in src/chattice/fsm/record.py

class BaseStorageFromRecord(BaseStorage):
    """Serve the six-method BaseStorage contract over a record store.

    Transitions use compare-and-set; a concurrent modification raises
    FSMRecordConflict instead of silently losing data.
    """

    def __init__(self, record_storage: FSMRecordStorage) -> None:
        self._records = record_storage

    async def get_state(self, key: StorageKey) -> str | None:
        record = await self._records.get_record(key)
        return record.state if record else None

    async def set_state(self, key: StorageKey, state: str | None) -> None:
        await self._mutate(key, state=state)

    async def get_data(self, key: StorageKey) -> dict[str, Any]:
        record = await self._records.get_record(key)
        return dict(record.data) if record else {}

    async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
        await self._mutate(key, data=dict(data))

    async def update_data(
        self, key: StorageKey, partial: Mapping[str, Any]
    ) -> dict[str, Any]:
        """Read/merge/compare-and-set retry loop: concurrent updates never
        silently overwrite each other (a conflict retries on the new
        revision instead of losing fields)."""
        while True:
            current = await self._records.get_record(key)
            revision = 0 if current is None else current.revision
            merged = dict(current.data) if current else {}
            merged.update(partial)
            try:
                replacement = FSMRecord(
                    state=current.state if current else None,
                    data=merged,
                    expires_at=current.expires_at if current else None,
                    schema_version=current.schema_version if current else 0,
                )
                await self._records.compare_and_set(key, revision, replacement)
                return merged
            except FSMRecordConflict:
                continue  # someone else wrote: re-read and retry

    async def finish(self, key: StorageKey) -> None:
        current = await self._records.get_record(key)
        if current is None:
            return
        await self._mutate(key, state=None, data={})

    async def _mutate(
        self,
        key: StorageKey,
        *,
        state: str | object | None = _NOT_PROVIDED,
        data: Mapping[str, Any] | None = None,
    ) -> None:
        current = await self._records.get_record(key)
        revision = 0 if current is None else current.revision
        if state is _NOT_PROVIDED:
            state_value = current.state if current else None
        else:
            state_value = cast("str | None", state)
        replacement = FSMRecord(
            state=state_value,
            data=data if data is not None else (dict(current.data) if current else {}),
            expires_at=current.expires_at if current else None,
            schema_version=current.schema_version if current else 0,
        )
        await self._records.compare_and_set(key, revision, replacement)

update_data(key, partial) async

Read/merge/compare-and-set retry loop: concurrent updates never silently overwrite each other (a conflict retries on the new revision instead of losing fields).

Source code in src/chattice/fsm/record.py

async def update_data(
    self, key: StorageKey, partial: Mapping[str, Any]
) -> dict[str, Any]:
    """Read/merge/compare-and-set retry loop: concurrent updates never
    silently overwrite each other (a conflict retries on the new
    revision instead of losing fields)."""
    while True:
        current = await self._records.get_record(key)
        revision = 0 if current is None else current.revision
        merged = dict(current.data) if current else {}
        merged.update(partial)
        try:
            replacement = FSMRecord(
                state=current.state if current else None,
                data=merged,
                expires_at=current.expires_at if current else None,
                schema_version=current.schema_version if current else 0,
            )
            await self._records.compare_and_set(key, revision, replacement)
            return merged
        except FSMRecordConflict:
            continue  # someone else wrote: re-read and retry

FSMContext

Bound to (storage, key) for one event; injected by the dispatcher.

Source code in src/chattice/fsm/context.py

class FSMContext:
    """Bound to (storage, key) for one event; injected by the dispatcher."""

    def __init__(self, storage: BaseStorage, key: StorageKey | None) -> None:
        self._storage = storage
        self._key = key

    def _require_key(self) -> StorageKey:
        if self._key is None:
            raise FSMError(
                "Cannot derive an FSM storage key for this event "
                "(missing user/space refs for the configured strategy)"
            )
        return self._key

    async def get_state(self) -> str | None:
        if self._key is None:
            return None
        return await self._storage.get_state(self._key)

    async def set_state(self, state: str | State | None) -> None:
        if isinstance(state, State):
            state = state.state
        await self._storage.set_state(self._require_key(), state)

    async def get_data(self) -> dict[str, Any]:
        if self._key is None:
            return {}
        return await self._storage.get_data(self._key)

    async def set_data(self, data: Mapping[str, Any]) -> None:
        await self._storage.set_data(self._require_key(), data)

    async def update_data(self, **partial: Any) -> dict[str, Any]:
        return await self._storage.update_data(self._require_key(), partial)

    async def finish(self) -> None:
        await self._storage.finish(self._require_key())

FSMError

Bases: RuntimeError

FSM operation attempted without a derivable storage key.

Source code in src/chattice/fsm/context.py

class FSMError(RuntimeError):
    """FSM operation attempted without a derivable storage key."""

FSMRecord dataclass

One FSM state+data snapshot under a StorageKey.

data is validated against the recursive JSONValue contract and defensively copied (MappingProxyType) so callers cannot mutate stored state without a compare-and-set.

Source code in src/chattice/fsm/record.py

@dataclass(frozen=True, slots=True)
class FSMRecord:
    """One FSM state+data snapshot under a StorageKey.

    ``data`` is validated against the recursive JSONValue contract and
    defensively copied (MappingProxyType) so callers cannot mutate
    stored state without a compare-and-set.
    """

    state: str | None = None
    data: Mapping[str, object] = field(default_factory=dict)
    revision: int = 0
    updated_at: float | None = None
    expires_at: float | None = None
    schema_version: int = 0

    def __post_init__(self) -> None:
        data = cast(
            dict[str, object],
            _require_json_value(dict(self.data), where="FSMRecord.data"),
        )
        object.__setattr__(self, "data", MappingProxyType(data))

FSMRecordConflict

Bases: RuntimeError

A compare-and-set failed: the stored revision differs from expected.

Source code in src/chattice/fsm/record.py

class FSMRecordConflict(RuntimeError):
    """A compare-and-set failed: the stored revision differs from expected."""

FSMRecordStorage

Bases: Protocol

Atomic record contract (optional storage).

Source code in src/chattice/fsm/record.py

class FSMRecordStorage(Protocol):
    """Atomic record contract (optional storage)."""

    async def get_record(self, key: StorageKey) -> FSMRecord | None:
        """Read the record; an expired record reads as None (lazy TTL)."""
        ...

    async def compare_and_set(
        self,
        key: StorageKey,
        expected_revision: int,
        replacement: FSMRecord,
    ) -> FSMRecord:
        """Atomically store ``replacement`` iff the current record has
        ``expected_revision`` (0 = no record). Returns the stored record
        with its revision bumped; raises FSMRecordConflict otherwise."""
        ...

compare_and_set(key, expected_revision, replacement) async

Atomically store replacement iff the current record has expected_revision (0 = no record). Returns the stored record with its revision bumped; raises FSMRecordConflict otherwise.

Source code in src/chattice/fsm/record.py

async def compare_and_set(
    self,
    key: StorageKey,
    expected_revision: int,
    replacement: FSMRecord,
) -> FSMRecord:
    """Atomically store ``replacement`` iff the current record has
    ``expected_revision`` (0 = no record). Returns the stored record
    with its revision bumped; raises FSMRecordConflict otherwise."""
    ...

get_record(key) async

Read the record; an expired record reads as None (lazy TTL).

Source code in src/chattice/fsm/record.py

async def get_record(self, key: StorageKey) -> FSMRecord | None:
    """Read the record; an expired record reads as None (lazy TTL)."""
    ...

FSMStrategy

Bases: Enum

How the storage key is derived from an event.

Source code in src/chattice/fsm/storage.py

class FSMStrategy(Enum):
    """How the storage key is derived from an event."""

    USER_IN_SPACE = "user_in_space"
    USER = "user"
    SPACE = "space"

MemoryFSMRecordStorage

In-process record storage: CAS under a per-key asyncio lock.

Source code in src/chattice/fsm/record.py

class MemoryFSMRecordStorage:
    """In-process record storage: CAS under a per-key asyncio lock."""

    def __init__(self, *, clock: Clock = time.time) -> None:
        self._records: dict[StorageKey, FSMRecord] = {}
        self._locks: dict[StorageKey, asyncio.Lock] = {}
        self._clock = clock

    def _lock_for(self, key: StorageKey) -> asyncio.Lock:
        lock = self._locks.get(key)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[key] = lock
        return lock

    async def get_record(self, key: StorageKey) -> FSMRecord | None:
        async with self._lock_for(key):
            record = self._records.get(key)
            if record is None:
                return None
            if _expired(record, self._clock()):
                del self._records[key]
                return None
            return record

    async def compare_and_set(
        self,
        key: StorageKey,
        expected_revision: int,
        replacement: FSMRecord,
    ) -> FSMRecord:
        async with self._lock_for(key):
            current = self._records.get(key)
            if current is not None and _expired(current, self._clock()):
                del self._records[key]
                current = None
            current_revision = 0 if current is None else current.revision
            if current_revision != expected_revision:
                raise FSMRecordConflict(
                    f"expected revision {expected_revision}, found {current_revision}"
                )
            stored = replace(
                replacement,
                revision=current_revision + 1,
                updated_at=self._clock(),
            )
            self._records[key] = stored
            return stored

MemoryStorage

In-process storage.

Concurrency guarantees are process-local: per-key asyncio locks serialize writes within one event loop; nothing here survives across processes.

Source code in src/chattice/fsm/storage.py

class MemoryStorage:
    """In-process storage.

    Concurrency guarantees are process-local: per-key asyncio locks serialize
    writes within one event loop; nothing here survives across processes.
    """

    def __init__(self) -> None:
        self._records: dict[StorageKey, _Record] = {}
        self._locks: dict[StorageKey, asyncio.Lock] = {}

    def _lock_for(self, key: StorageKey) -> asyncio.Lock:
        lock = self._locks.get(key)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[key] = lock
        return lock

    def _record(self, key: StorageKey) -> _Record:
        record = self._records.get(key)
        if record is None:
            record = _Record()
            self._records[key] = record
        return record

    async def get_state(self, key: StorageKey) -> str | None:
        async with self._lock_for(key):
            record = self._records.get(key)
            return record.state if record is not None else None

    async def set_state(self, key: StorageKey, state: str | None) -> None:
        async with self._lock_for(key):
            if state is None and key not in self._records:
                return
            self._record(key).state = state

    async def get_data(self, key: StorageKey) -> dict[str, Any]:
        async with self._lock_for(key):
            record = self._records.get(key)
            return dict(record.data) if record is not None else {}

    async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
        async with self._lock_for(key):
            self._record(key).data = dict(data)

    async def update_data(
        self, key: StorageKey, partial: Mapping[str, Any]
    ) -> dict[str, Any]:
        async with self._lock_for(key):
            record = self._record(key)
            record.data.update(partial)
            return dict(record.data)

    async def finish(self, key: StorageKey) -> None:
        async with self._lock_for(key):
            self._records.pop(key, None)

RedisFSMRecordStorage

Redis record storage: CAS via WATCH/MULTI, whole-record TTL (PX).

One JSON document per StorageKey; compare-and-set is optimistic with retry on watch conflicts (no distributed lock held across I/O).

Source code in src/chattice/fsm/record.py

class RedisFSMRecordStorage:
    """Redis record storage: CAS via WATCH/MULTI, whole-record TTL (PX).

    One JSON document per StorageKey; compare-and-set is optimistic with
    retry on watch conflicts (no distributed lock held across I/O).
    """

    def __init__(
        self,
        redis: aioredis.Redis | None = None,
        *,
        url: str = _DEFAULT_URL,
        prefix: str = "chattice:fsmrecord",
        clock: Clock = time.time,
    ) -> None:
        try:
            from redis import asyncio as aioredis
        except ImportError as error:
            raise ImportError(
                "Redis-backed storage requires the `chattice[redis]` extra "
                "(pip install 'chattice[redis]')."
            ) from error

        if redis is not None:
            self._redis = redis
        else:
            self._redis = aioredis.from_url(  # type: ignore[no-untyped-call]
                url, decode_responses=True
            )
        self._prefix = prefix
        self._clock = clock
        self._owns_redis = redis is None
        self._redis_closed = False

    async def aclose(self) -> None:
        """Close the internally created client (idempotent).

        An injected client is never closed by the storage.
        """
        if not self._owns_redis or self._redis_closed:
            return
        self._redis_closed = True
        closer = getattr(self._redis, "aclose", None) or getattr(
            self._redis, "close", None
        )
        if closer is None:
            return
        result = closer()
        if inspect.isawaitable(result):
            await result

    def _redis_key(self, key: StorageKey) -> str:
        parts = (key.user or "*", key.space or "*", key.thread or "*")
        return f"{self._prefix}:{':'.join(parts)}"

    @staticmethod
    def _encode(record: FSMRecord) -> str:
        return json.dumps(
            {
                "state": record.state,
                "data": dict(record.data),
                "revision": record.revision,
                "updated_at": record.updated_at,
                "expires_at": record.expires_at,
                "schema_version": record.schema_version,
            },
            separators=(",", ":"),
        )

    @staticmethod
    def _decode(raw: str) -> FSMRecord:
        payload = json.loads(raw)
        return FSMRecord(
            state=payload.get("state"),
            data=payload.get("data") or {},
            revision=int(payload.get("revision") or 0),
            updated_at=payload.get("updated_at"),
            expires_at=payload.get("expires_at"),
            schema_version=int(payload.get("schema_version") or 0),
        )

    async def get_record(self, key: StorageKey) -> FSMRecord | None:
        from redis.exceptions import WatchError

        redis_key = self._redis_key(key)
        raw = await self._redis.get(redis_key)
        if raw is None:
            return None
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8")
        record = self._decode(raw)
        if not _expired(record, self._clock()):
            return record
        # Lazy expiry: delete ONLY the exact value we read — a concurrent
        # writer may have replaced the expired record meanwhile, and a
        # blind delete would destroy the NEW record.
        while True:
            async with self._redis.pipeline() as pipe:
                try:
                    await pipe.watch(redis_key)
                    current_raw = await pipe.get(redis_key)
                    if isinstance(current_raw, bytes):
                        current_raw = current_raw.decode("utf-8")
                    if current_raw != raw:
                        # replaced concurrently: the new value wins
                        if current_raw is None:
                            return None
                        current = self._decode(current_raw)
                        return None if _expired(current, self._clock()) else current
                    pipe.multi()  # type: ignore[no-untyped-call]
                    pipe.delete(redis_key)
                    await pipe.execute()
                    return None
                except WatchError:
                    continue  # changed during watch: re-read and re-decide

    async def compare_and_set(
        self,
        key: StorageKey,
        expected_revision: int,
        replacement: FSMRecord,
    ) -> FSMRecord:
        from redis.exceptions import WatchError

        redis_key = self._redis_key(key)
        while True:
            async with self._redis.pipeline() as pipe:
                try:
                    await pipe.watch(redis_key)
                    raw = await pipe.get(redis_key)
                    current: FSMRecord | None = None
                    if raw is not None:
                        if isinstance(raw, bytes):
                            raw = raw.decode("utf-8")
                        current = self._decode(raw)
                    if current is not None and _expired(current, self._clock()):
                        current = None  # lazy TTL: treat as absent
                    current_revision = 0 if current is None else current.revision
                    if current_revision != expected_revision:
                        raise FSMRecordConflict(
                            f"expected revision {expected_revision}, "
                            f"found {current_revision}"
                        )
                    stored = replace(
                        replacement,
                        revision=current_revision + 1,
                        updated_at=self._clock(),
                    )
                    pipe.multi()  # type: ignore[no-untyped-call]
                    if stored.expires_at is not None:
                        ttl_ms = max(1, int((stored.expires_at - self._clock()) * 1000))
                        pipe.set(redis_key, self._encode(stored), px=ttl_ms)
                    else:
                        pipe.set(redis_key, self._encode(stored))
                    await pipe.execute()
                    return stored
                except WatchError:
                    continue  # another writer changed the record; retry

aclose() async

Close the internally created client (idempotent).

An injected client is never closed by the storage.

Source code in src/chattice/fsm/record.py

async def aclose(self) -> None:
    """Close the internally created client (idempotent).

    An injected client is never closed by the storage.
    """
    if not self._owns_redis or self._redis_closed:
        return
    self._redis_closed = True
    closer = getattr(self._redis, "aclose", None) or getattr(
        self._redis, "close", None
    )
    if closer is None:
        return
    result = closer()
    if inspect.isawaitable(result):
        await result

RedisStorage

FSM storage over redis.asyncio with a namespaced key layout.

Source code in src/chattice/fsm/redis.py

class RedisStorage:
    """FSM storage over redis.asyncio with a namespaced key layout."""

    def __init__(
        self,
        redis: aioredis.Redis | None = None,
        *,
        url: str = _DEFAULT_URL,
        prefix: str = "chattice:fsm",
    ) -> None:
        if redis is not None:
            self._redis = redis
        else:
            # redis.asyncio.from_url is untyped in redis 6.4; it returns a client.
            # decode_responses=True is required so get() returns str, not bytes.
            self._redis = aioredis.from_url(  # type: ignore[no-untyped-call]
                url, decode_responses=True
            )
        self._prefix = prefix
        self._locks: dict[str, asyncio.Lock] = {}
        self._owns_redis = redis is None
        self._redis_closed = False

    async def aclose(self) -> None:
        """Close the internally created client (idempotent).

        An injected client is never closed by the storage.
        """
        if not self._owns_redis or self._redis_closed:
            return
        self._redis_closed = True
        closer = getattr(self._redis, "aclose", None) or getattr(
            self._redis, "close", None
        )
        if closer is None:
            return
        result = closer()
        if inspect.isawaitable(result):
            await result

    def _lock_for(self, redis_key: str) -> asyncio.Lock:
        lock = self._locks.get(redis_key)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[redis_key] = lock
        return lock

    def _base(self, key: StorageKey) -> str:
        parts = (key.user or "*", key.space or "*", key.thread or "*")
        return f"{self._prefix}:{':'.join(parts)}"

    def _state_key(self, key: StorageKey) -> str:
        return f"{self._base(key)}:state"

    def _data_key(self, key: StorageKey) -> str:
        return f"{self._base(key)}:data"

    async def get_state(self, key: StorageKey) -> str | None:
        raw = await self._redis.get(self._state_key(key))
        if isinstance(raw, str):
            return raw
        if isinstance(raw, bytes):
            return raw.decode("utf-8")
        return None

    async def set_state(self, key: StorageKey, state: str | None) -> None:
        state_key = self._state_key(key)
        if state is None:
            await self._redis.delete(state_key)
        else:
            await self._redis.set(state_key, state)

    async def get_data(self, key: StorageKey) -> dict[str, Any]:
        raw = await self._redis.get(self._data_key(key))
        if raw is None:
            return {}
        try:
            decoded = json.loads(raw)
        except json.JSONDecodeError:
            return {}
        return decoded if isinstance(decoded, dict) else {}

    async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
        await self._redis.set(self._data_key(key), json.dumps(dict(data)))

    async def update_data(
        self, key: StorageKey, partial: Mapping[str, Any]
    ) -> dict[str, Any]:
        data_key = self._data_key(key)
        async with self._lock_for(data_key):
            merged = await self.get_data(key)
            merged.update(partial)
            await self._redis.set(data_key, json.dumps(merged))
            return merged

    async def finish(self, key: StorageKey) -> None:
        await self._redis.delete(self._state_key(key), self._data_key(key))

aclose() async

Close the internally created client (idempotent).

An injected client is never closed by the storage.

Source code in src/chattice/fsm/redis.py

async def aclose(self) -> None:
    """Close the internally created client (idempotent).

    An injected client is never closed by the storage.
    """
    if not self._owns_redis or self._redis_closed:
        return
    self._redis_closed = True
    closer = getattr(self._redis, "aclose", None) or getattr(
        self._redis, "close", None
    )
    if closer is None:
        return
    result = closer()
    if inspect.isawaitable(result):
        await result

State

A named workflow state.

Members of a StatesGroup get <GroupName>:<attr_name> string keys; standalone states may pass an explicit key.

Source code in src/chattice/fsm/states.py

class State:
    """A named workflow state.

    Members of a StatesGroup get ``<GroupName>:<attr_name>`` string keys;
    standalone states may pass an explicit key.
    """

    def __init__(self, *, state: str | None = None) -> None:
        self._state = state
        self._name: str | None = None
        self._group: str | None = None

    @property
    def state(self) -> str:
        """The string key used by storages and StateFilter."""
        if self._state is not None:
            return self._state
        if self._group is None or self._name is None:
            raise RuntimeError("Unbound State: it must be a StatesGroup member")
        return f"{self._group}:{self._name}"

state property

The string key used by storages and StateFilter.

StateFilter

Bases: BaseFilter

Matches when the event's current FSM state is one of the given states.

An empty filter matches ANY non-None state. Without an FSM context in the filter context (dispatcher configured without fsm_storage) it never matches.

Source code in src/chattice/fsm/filter.py

class StateFilter(BaseFilter):
    """Matches when the event's current FSM state is one of the given states.

    An empty filter matches ANY non-None state. Without an FSM context in
    the filter context (dispatcher configured without fsm_storage) it never
    matches.
    """

    def __init__(self, *states: State) -> None:
        self._states = {state.state for state in states}

    async def __call__(
        self, event: Event, context: Mapping[str, object]
    ) -> FilterValue:
        del event
        state = context.get("state")
        if not isinstance(state, FSMContext):
            return False
        current = await state.get_state()
        if current is None:
            return False
        if not self._states:
            return True
        return current in self._states

StatesGroup

Base class for workflow state groups.

Source code in src/chattice/fsm/states.py

class StatesGroup(metaclass=StatesGroupMeta):
    """Base class for workflow state groups."""

    __all_states__: ClassVar[dict[str, State]] = {}

StorageKey dataclass

The composite identity an FSM record is stored under.

Source code in src/chattice/fsm/storage.py

@dataclass(frozen=True, slots=True)
class StorageKey:
    """The composite identity an FSM record is stored under."""

    user: str | None
    space: str | None
    thread: str | None

    @classmethod
    def build(cls, event: Event, strategy: FSMStrategy) -> StorageKey | None:
        """Derive the key from the event refs; None when refs are missing."""
        user = event.actor.name if event.actor is not None else None
        space = event.space.name if event.space is not None else None
        thread = event.thread.name if event.thread is not None else None
        if strategy is FSMStrategy.USER_IN_SPACE:
            if user is None or space is None:
                return None
            return cls(user=user, space=space, thread=thread)
        if strategy is FSMStrategy.USER:
            if user is None:
                return None
            return cls(user=user, space=None, thread=None)
        if strategy is FSMStrategy.SPACE:
            if space is None:
                return None
            return cls(user=None, space=space, thread=None)
        raise ValueError(f"Unknown FSMStrategy {strategy!r}")

build(event, strategy) classmethod

Derive the key from the event refs; None when refs are missing.

Source code in src/chattice/fsm/storage.py

@classmethod
def build(cls, event: Event, strategy: FSMStrategy) -> StorageKey | None:
    """Derive the key from the event refs; None when refs are missing."""
    user = event.actor.name if event.actor is not None else None
    space = event.space.name if event.space is not None else None
    thread = event.thread.name if event.thread is not None else None
    if strategy is FSMStrategy.USER_IN_SPACE:
        if user is None or space is None:
            return None
        return cls(user=user, space=space, thread=thread)
    if strategy is FSMStrategy.USER:
        if user is None:
            return None
        return cls(user=user, space=None, thread=None)
    if strategy is FSMStrategy.SPACE:
        if space is None:
            return None
        return cls(user=None, space=space, thread=None)
    raise ValueError(f"Unknown FSMStrategy {strategy!r}")

Owner-safe push idempotency: claimed(owner, lease) -> completed.

A TTL presence bit can acknowledge work that never completed. The contract is a small state machine:

claim(key, owner, lease)  -> FIRST | COMPLETED | ACTIVE
complete(key, owner)      -> mark done (keeps absorbing duplicates)
release(key, owner)       -> drop the claim (only the OWNER may)
renew(key, owner, lease)  -> extend a long handler's lease

A second delivery that observes another owner's ACTIVE claim answers 429 ("still processing") so Pub/Sub redelivers later — it is never acknowledged as a completed duplicate. Keys must be namespaced by the caller (subscription/topic + messageId): Google message IDs are unique per topic only.

Memory is process-local; Redis stores one JSON document per key with claim via SET NX and owner-checked release via WATCH/MULTI.

ClaimResult

Bases: Enum

Outcome of claim(): who owns the delivery now.

Source code in src/chattice/idempotency.py

class ClaimResult(enum.Enum):
    """Outcome of claim(): who owns the delivery now."""

    FIRST = "first"  # this owner claimed it: dispatch
    COMPLETED = "completed"  # a previous owner finished: absorb as duplicate
    ACTIVE = "active"  # another owner is still processing: retry later

IdempotencyStorage

Bases: Protocol

Owner-safe claim/complete/release contract for push dedupe.

Source code in src/chattice/idempotency.py

class IdempotencyStorage(Protocol):
    """Owner-safe claim/complete/release contract for push dedupe."""

    async def claim(
        self, key: str, *, owner: str, lease_seconds: float
    ) -> ClaimResult: ...

    async def complete(self, key: str, *, owner: str) -> None: ...

    async def release(self, key: str, *, owner: str) -> None: ...

    async def renew(self, key: str, *, owner: str, lease_seconds: float) -> bool: ...

MemoryIdempotencyStorage

In-process state machine (per-key asyncio locks).

Source code in src/chattice/idempotency.py

class MemoryIdempotencyStorage:
    """In-process state machine (per-key asyncio locks)."""

    def __init__(self, *, clock: Clock = time.time) -> None:
        self._claims: dict[str, dict[str, Any]] = {}
        self._locks: dict[str, asyncio.Lock] = {}
        self._clock = clock

    def _lock_for(self, key: str) -> asyncio.Lock:
        lock = self._locks.get(key)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[key] = lock
        return lock

    def _prune_expired(self, key: str) -> None:
        claim = self._claims.get(key)
        if claim is not None and not claim["completed"]:
            if claim["lease_until"] <= self._clock():
                del self._claims[key]

    async def claim(self, key: str, *, owner: str, lease_seconds: float) -> ClaimResult:
        async with self._lock_for(key):
            self._prune_expired(key)
            claim = self._claims.get(key)
            if claim is None:
                self._claims[key] = {
                    "owner": owner,
                    "lease_until": self._clock() + lease_seconds,
                    "completed": False,
                }
                return ClaimResult.FIRST
            if claim["completed"]:
                return ClaimResult.COMPLETED
            return ClaimResult.ACTIVE

    async def complete(self, key: str, *, owner: str) -> None:
        async with self._lock_for(key):
            claim = self._claims.get(key)
            if claim is not None and claim["owner"] == owner:
                claim["completed"] = True

    async def release(self, key: str, *, owner: str) -> None:
        async with self._lock_for(key):
            claim = self._claims.get(key)
            if claim is not None and claim["owner"] == owner:
                del self._claims[key]

    async def renew(self, key: str, *, owner: str, lease_seconds: float) -> bool:
        async with self._lock_for(key):
            claim = self._claims.get(key)
            if claim is None or claim["owner"] != owner or claim["completed"]:
                return False
            claim["lease_until"] = self._clock() + lease_seconds
            return True

RedisIdempotencyStorage

Redis state machine: SET NX claim, owner-checked WATCH/MULTI ops.

Source code in src/chattice/idempotency.py

class RedisIdempotencyStorage:
    """Redis state machine: SET NX claim, owner-checked WATCH/MULTI ops."""

    def __init__(
        self,
        redis: aioredis.Redis | None = None,
        *,
        url: str = _DEFAULT_URL,
        prefix: str = "chattice:idem",
        clock: Clock = time.time,
        completed_retention_seconds: float = 86400.0,
    ) -> None:
        try:
            from redis import asyncio as aioredis
        except ImportError as error:
            raise ImportError(
                "Redis-backed storage requires the `chattice[redis]` extra "
                "(pip install 'chattice[redis]')."
            ) from error

        if redis is not None:
            self._redis = redis
        else:
            self._redis = aioredis.from_url(  # type: ignore[no-untyped-call]
                url, decode_responses=True
            )
        self._prefix = prefix
        self._clock = clock
        self._completed_ttl_ms = max(1, int(completed_retention_seconds * 1000))
        self._owns_redis = redis is None
        self._redis_closed = False

    async def aclose(self) -> None:
        """Close the internally created client (idempotent).

        An injected client is never closed by the storage.
        """
        if not self._owns_redis or self._redis_closed:
            return
        self._redis_closed = True
        closer = getattr(self._redis, "aclose", None) or getattr(
            self._redis, "close", None
        )
        if closer is None:
            return
        result = closer()
        if inspect.isawaitable(result):
            await result

    def _key(self, key: str) -> str:
        return f"{self._prefix}:{key}"

    @staticmethod
    def _encode(owner: str, lease_until: float, completed: bool) -> str:
        return json.dumps(
            {"o": owner, "l": lease_until, "c": completed}, separators=(",", ":")
        )

    @staticmethod
    def _decode(raw: str) -> dict[str, Any]:
        payload = json.loads(raw)
        return {
            "owner": payload["o"],
            "lease_until": payload["l"],
            "completed": payload["c"],
        }

    async def claim(self, key: str, *, owner: str, lease_seconds: float) -> ClaimResult:
        from redis.exceptions import WatchError

        redis_key = self._key(key)
        value = self._encode(owner, self._clock() + lease_seconds, False)
        px = max(1, int(lease_seconds * 1000))
        first = await self._redis.set(redis_key, value, nx=True, px=px)
        if first:
            return ClaimResult.FIRST
        while True:
            # Expired-takeover MUST be a single conditional decision:
            # WATCH the key, verify the EXACT claim we read is still the
            # expired one, then replace inside MULTI/EXEC. An
            # unconditional SET would let two reclaimers both win.
            async with self._redis.pipeline() as pipe:
                try:
                    await pipe.watch(redis_key)
                    raw = await pipe.get(redis_key)
                    if raw is None:
                        pipe.multi()  # type: ignore[no-untyped-call]
                        pipe.set(redis_key, value, nx=True, px=px)
                        executed = await pipe.execute()
                        return (
                            ClaimResult.FIRST
                            if executed and executed[0]
                            else ClaimResult.ACTIVE
                        )
                    if isinstance(raw, bytes):
                        raw = raw.decode("utf-8")
                    claim = self._decode(raw)
                    if claim["completed"]:
                        return ClaimResult.COMPLETED
                    if claim["lease_until"] > self._clock():
                        return ClaimResult.ACTIVE
                    pipe.multi()  # type: ignore[no-untyped-call]
                    pipe.set(redis_key, value, px=px)
                    await pipe.execute()
                    return ClaimResult.FIRST
                except WatchError:
                    continue  # changed under us: re-read and re-decide

    async def complete(self, key: str, *, owner: str) -> None:
        await self._owner_op(key, owner, completed=True, px=self._completed_ttl_ms)

    async def release(self, key: str, *, owner: str) -> None:
        await self._owner_op(key, owner, delete=True)

    async def renew(self, key: str, *, owner: str, lease_seconds: float) -> bool:
        from redis.exceptions import WatchError

        redis_key = self._key(key)
        while True:
            async with self._redis.pipeline() as pipe:
                try:
                    await pipe.watch(redis_key)
                    raw = await pipe.get(redis_key)
                    if raw is None:
                        return False
                    if isinstance(raw, bytes):
                        raw = raw.decode("utf-8")
                    claim = self._decode(raw)
                    if claim["owner"] != owner or claim["completed"]:
                        return False
                    pipe.multi()  # type: ignore[no-untyped-call]
                    # The renewed lease MUST carry its TTL — a plain
                    # SET discards the previous expiry and the key would
                    # never expire again (TTL -1 after renew).
                    pipe.set(
                        redis_key,
                        self._encode(owner, self._clock() + lease_seconds, False),
                        px=max(1, int(lease_seconds * 1000)),
                    )
                    await pipe.execute()
                    return True
                except WatchError:
                    continue

    async def _owner_op(
        self,
        key: str,
        owner: str,
        *,
        completed: bool = False,
        delete: bool = False,
        px: int | None = None,
    ) -> None:
        from redis.exceptions import WatchError

        redis_key = self._key(key)
        while True:
            async with self._redis.pipeline() as pipe:
                try:
                    await pipe.watch(redis_key)
                    raw = await pipe.get(redis_key)
                    if raw is None:
                        return
                    if isinstance(raw, bytes):
                        raw = raw.decode("utf-8")
                    claim = self._decode(raw)
                    if claim["owner"] != owner:
                        return  # never touch another owner's claim
                    pipe.multi()  # type: ignore[no-untyped-call]
                    if delete:
                        pipe.delete(redis_key)
                    elif px is not None:
                        pipe.set(
                            redis_key,
                            self._encode(owner, claim["lease_until"], completed),
                            px=px,
                        )
                    else:
                        pipe.set(
                            redis_key,
                            self._encode(owner, claim["lease_until"], completed),
                        )
                    await pipe.execute()
                    return
                except WatchError:
                    continue

aclose() async

Close the internally created client (idempotent).

An injected client is never closed by the storage.

Source code in src/chattice/idempotency.py

async def aclose(self) -> None:
    """Close the internally created client (idempotent).

    An injected client is never closed by the storage.
    """
    if not self._owns_redis or self._redis_closed:
        return
    self._redis_closed = True
    closer = getattr(self._redis, "aclose", None) or getattr(
        self._redis, "close", None
    )
    if closer is None:
        return
    result = closer()
    if inspect.isawaitable(result):
        await result

new_owner()

A fresh owner token for one delivery attempt.

Source code in src/chattice/idempotency.py

def new_owner() -> str:
    """A fresh owner token for one delivery attempt."""
    return uuid.uuid4().hex

Extension hooks for observability (application-owned integrations).

The framework ships NO OTel dependency; applications implement these hooks and bridge to their tracer of choice (see docs/architecture/observability.md).

The optional hooks (everything after after_event) are no-ops in the protocol: an implementation may provide any subset without breaking the structural contract. Hooks receive the original event and dispatch context, including access to raw payloads. Applications choose which fields to export.

ObservabilityHooks

Bases: Protocol

Called around each feed_update routing pass.

Source code in src/chattice/observability.py

class ObservabilityHooks(Protocol):
    """Called around each feed_update routing pass."""

    async def before_event(self, event: object, data: dict[str, object]) -> None: ...

    async def after_event(
        self,
        event: object,
        data: dict[str, object],
        result: object,
        error: BaseException | None,
    ) -> None: ...

    # ---- optional hooks (no-ops in the protocol) ----

    async def before_handler(
        self, event: object, data: dict[str, object], handler: str
    ) -> None:
        """Called before the selected handler invokes; ``handler`` is the
        qualified callback name."""

    async def after_handler(
        self, event: object, data: dict[str, object], handler: str, result: object
    ) -> None:
        """Called after the selected handler returns."""

    async def before_outbound(self, event: object, operation: str) -> None:
        """Called before an outbound Google answer (send/update)."""

    async def after_outbound(self, event: object, operation: str) -> None:
        """Called after an outbound Google answer completes."""

    async def delivery_acked(self, message_id: str, event: object) -> None:
        """Called after a Pub/Sub delivery is ACKed."""

    async def delivery_nacked(self, message_id: str, event: object) -> None:
        """Called after a Pub/Sub delivery is NACKed."""

after_handler(event, data, handler, result) async

Called after the selected handler returns.

Source code in src/chattice/observability.py

async def after_handler(
    self, event: object, data: dict[str, object], handler: str, result: object
) -> None:
    """Called after the selected handler returns."""

after_outbound(event, operation) async

Called after an outbound Google answer completes.

Source code in src/chattice/observability.py

async def after_outbound(self, event: object, operation: str) -> None:
    """Called after an outbound Google answer completes."""

before_handler(event, data, handler) async

Called before the selected handler invokes; handler is the qualified callback name.

Source code in src/chattice/observability.py

async def before_handler(
    self, event: object, data: dict[str, object], handler: str
) -> None:
    """Called before the selected handler invokes; ``handler`` is the
    qualified callback name."""

before_outbound(event, operation) async

Called before an outbound Google answer (send/update).

Source code in src/chattice/observability.py

async def before_outbound(self, event: object, operation: str) -> None:
    """Called before an outbound Google answer (send/update)."""

delivery_acked(message_id, event) async

Called after a Pub/Sub delivery is ACKed.

Source code in src/chattice/observability.py

async def delivery_acked(self, message_id: str, event: object) -> None:
    """Called after a Pub/Sub delivery is ACKed."""

delivery_nacked(message_id, event) async

Called after a Pub/Sub delivery is NACKed.

Source code in src/chattice/observability.py

async def delivery_nacked(self, message_id: str, event: object) -> None:
    """Called after a Pub/Sub delivery is NACKed."""

RuntimeDiagnostics dataclass

Configurable thresholds for runtime diagnostics.

None disables the corresponding warning.

Source code in src/chattice/observability.py

@dataclass(frozen=True, slots=True)
class RuntimeDiagnostics:
    """Configurable thresholds for runtime diagnostics.

    ``None`` disables the corresponding warning.
    """

    slow_handler_ms: float | None = 1000.0
    delayed_event_ms: float | None = 5000.0

Transports and integrations

HTTP interaction transport core.

DoubleResponseError

Bases: HTTPInteractionError

The synchronous response was already set for this interaction.

Source code in src/chattice/transports/http/errors.py

class DoubleResponseError(HTTPInteractionError):
    """The synchronous response was already set for this interaction."""

GoogleTokenVerifier

Verify Chat bearer tokens using google-auth and the documented flows.

One audience string supports both documented Authentication Audience strategies: the HTTP endpoint URL (OIDC ID token via verify_oauth2_token) or the project number (self-signed JWT via the Chat service-account certificates). Signature, exp, aud, and kid-based certificate selection are handled by google-auth; the Google Chat identity (email) and issuer are checked explicitly per strategy (the official samples do the same). Fail-closed: any verification failure answers VerificationError, never a bypass.

Source code in src/chattice/transports/http/verifier.py

class GoogleTokenVerifier:
    """Verify Chat bearer tokens using google-auth and the documented flows.

    One audience string supports both documented Authentication Audience
    strategies: the HTTP endpoint URL (OIDC ID token via
    verify_oauth2_token) or the project number (self-signed JWT via the
    Chat service-account certificates). Signature, exp, aud, and
    kid-based certificate selection are handled by google-auth; the
    Google Chat identity (email) and issuer are checked explicitly per
    strategy (the official samples do the same). Fail-closed: any
    verification failure answers VerificationError, never a bypass.
    """

    def __init__(
        self,
        *,
        audience: str,
        request: google_requests.Request | None = None,
        clock_skew_in_seconds: int = 10,
    ) -> None:
        self._audience = audience
        self._request = request if request is not None else google_requests.Request()
        self._clock_skew_in_seconds = clock_skew_in_seconds

    def verify(self, request: IncomingRequest) -> None:
        token = extract_bearer(request)
        try:
            if self._audience.startswith("https://"):
                # Endpoint-URL audience: the OFFICIAL Google Chat OIDC
                # flow. verify_oauth2_token checks the signature and the
                # audience against the standard Google OAuth2
                # certificates and validates the issuer itself — never
                # pass a certs_url here (an explicit None would make
                # google-auth fetch 'https://None').
                claims = verify_oauth2_token(  # type: ignore[no-untyped-call]
                    token, self._request, audience=self._audience
                )
                # Identity check: only the Google Chat service may speak
                # for this endpoint (documented expected identity).
                if claims.get("email") != _CHAT_ISSUER:
                    raise VerificationError(
                        "Token identity is not the Google Chat service"
                    )
                if claims.get("email_verified") is not True:
                    raise VerificationError("Token email is not verified")
            else:
                # Project-number audience: a self-signed JWT by the Chat
                # service account, verified against the Chat
                # service-account certificate endpoint.
                claims = verify_token(
                    token,
                    request=self._request,
                    audience=self._audience,
                    certs_url=_CHAT_CERTS_URL,
                    clock_skew_in_seconds=self._clock_skew_in_seconds,
                )
                _validate_issuer(claims)
        except VerificationError:
            raise
        except google_auth_exceptions.TransportError as error:
            # Fail-closed: an unreachable certificate endpoint is a
            # verification failure, never a bypass.
            raise VerificationError(
                "Cannot reach Google Chat issuer certificates"
            ) from error
        except (ValueError, google_auth_exceptions.GoogleAuthError) as error:
            raise VerificationError("Invalid bearer token") from error

HTTPInteractionAdapter

Framework-neutral adapter: HTTP request snapshot -> domain event.

Source code in src/chattice/transports/http/adapter.py

class HTTPInteractionAdapter:
    """Framework-neutral adapter: HTTP request snapshot -> domain event."""

    def parse(self, request: IncomingRequest) -> Event:
        """Decode the request body and parse it into a domain event."""
        return parse_interaction(request.json())

parse(request)

Decode the request body and parse it into a domain event.

Source code in src/chattice/transports/http/adapter.py

def parse(self, request: IncomingRequest) -> Event:
    """Decode the request body and parse it into a domain event."""
    return parse_interaction(request.json())

HTTPInteractionError

Bases: ValueError

Base class for HTTP transport failures.

Source code in src/chattice/transports/http/errors.py

class HTTPInteractionError(ValueError):
    """Base class for HTTP transport failures."""

IncomingRequest dataclass

Immutable snapshot of an inbound interaction HTTP request.

Source code in src/chattice/transports/http/request.py

@dataclass(frozen=True, slots=True, kw_only=True)
class IncomingRequest:
    """Immutable snapshot of an inbound interaction HTTP request."""

    method: str
    path: str
    body: bytes = b""
    headers: Mapping[str, str] = field(default_factory=dict)
    received_at: datetime = field(default_factory=_utcnow)

    def header(self, name: str) -> str | None:
        """Case-insensitive header lookup."""
        lowered = name.lower()
        for key, value in self.headers.items():
            if key.lower() == lowered:
                return value
        return None

    def json(self) -> Mapping[str, object]:
        """Decode the body as a JSON object (lazy, one call per request)."""
        try:
            decoded = json.loads(self.body)
        except (json.JSONDecodeError, UnicodeDecodeError) as error:
            raise InvalidInteractionPayload("Request body is not valid JSON") from error
        if not isinstance(decoded, Mapping):
            raise InvalidInteractionPayload("Request body must be a JSON object")
        return decoded

header(name)

Case-insensitive header lookup.

Source code in src/chattice/transports/http/request.py

def header(self, name: str) -> str | None:
    """Case-insensitive header lookup."""
    lowered = name.lower()
    for key, value in self.headers.items():
        if key.lower() == lowered:
            return value
    return None

json()

Decode the body as a JSON object (lazy, one call per request).

Source code in src/chattice/transports/http/request.py

def json(self) -> Mapping[str, object]:
    """Decode the body as a JSON object (lazy, one call per request)."""
    try:
        decoded = json.loads(self.body)
    except (json.JSONDecodeError, UnicodeDecodeError) as error:
        raise InvalidInteractionPayload("Request body is not valid JSON") from error
    if not isinstance(decoded, Mapping):
        raise InvalidInteractionPayload("Request body must be a JSON object")
    return decoded

IncomingRequestVerifier

Bases: Protocol

Contract: prove that an inbound request genuinely came from Google Chat.

Source code in src/chattice/transports/http/verifier.py

class IncomingRequestVerifier(Protocol):
    """Contract: prove that an inbound request genuinely came from Google Chat."""

    def verify(self, request: IncomingRequest) -> None:
        """Raise VerificationError when the request cannot be verified."""
        ...

verify(request)

Raise VerificationError when the request cannot be verified.

Source code in src/chattice/transports/http/verifier.py

def verify(self, request: IncomingRequest) -> None:
    """Raise VerificationError when the request cannot be verified."""
    ...

InteractionContext dataclass

HTTP transport-only request/response state available through DI.

The normalized domain event and its response capabilities are separate DI values; this type deliberately does not claim to be a canonical entity or resource context.

Source code in src/chattice/transports/http/adapter.py

@dataclass(frozen=True, slots=True, kw_only=True)
class InteractionContext:
    """HTTP transport-only request/response state available through DI.

    The normalized domain event and its response capabilities are separate DI
    values; this type deliberately does not claim to be a canonical entity or
    resource context.
    """

    request: IncomingRequest
    response: InteractionResponse
    received_at: datetime
    deadline_at: datetime

    @property
    def remaining(self) -> timedelta:
        """Time left before the documented sync response deadline."""
        return self.deadline_at - datetime.now(UTC)

remaining property

Time left before the documented sync response deadline.

InteractionResponse dataclass

Request-scoped mutable response plan; guards against double responses.

This is intentionally not frozen: it is per-request mutable state, never shared between requests.

Source code in src/chattice/transports/http/response.py

@dataclass(slots=True)
class InteractionResponse:
    """Request-scoped mutable response plan; guards against double responses.

    This is intentionally not frozen: it is per-request mutable state, never
    shared between requests.
    """

    payload: object = None
    state: ResponseState = ResponseState.NOT_RESPONDED

    def respond(self, payload: object) -> None:
        """Set the synchronous response payload exactly once."""
        if self.state is ResponseState.RESPONDED:
            raise DoubleResponseError(
                "This interaction already has a synchronous response"
            )
        self.payload = payload
        self.state = ResponseState.RESPONDED

respond(payload)

Set the synchronous response payload exactly once.

Source code in src/chattice/transports/http/response.py

def respond(self, payload: object) -> None:
    """Set the synchronous response payload exactly once."""
    if self.state is ResponseState.RESPONDED:
        raise DoubleResponseError(
            "This interaction already has a synchronous response"
        )
    self.payload = payload
    self.state = ResponseState.RESPONDED

MockVerifier

Accepts (or rejects) any request; for tests and local development only.

Source code in src/chattice/transports/http/verifier.py

class MockVerifier:
    """Accepts (or rejects) any request; for tests and local development only."""

    def __init__(self, *, reject: bool = False) -> None:
        self._reject = reject

    def verify(self, request: IncomingRequest) -> None:
        if self._reject:
            raise VerificationError("Mock verifier rejected the request")

RawInteractionResponse dataclass

Explicit raw-response escape hatch: an arbitrary response mapping.

Still validated against event/channel invariants (e.g. a REMOVED_FROM_SPACE event can never receive a response) — only the payload shape is the caller's responsibility.

Source code in src/chattice/transports/http/response.py

@dataclass(frozen=True, slots=True)
class RawInteractionResponse:
    """Explicit raw-response escape hatch: an arbitrary response mapping.

    Still validated against event/channel invariants (e.g. a
    REMOVED_FROM_SPACE event can never receive a response) — only the
    payload shape is the caller's responsibility.
    """

    payload: Mapping[str, object]

RequestConfigResponse dataclass

Typed REQUEST_CONFIG response: ask the user to open an auth URL.

The missing typed DX for a wire shape previously reachable only through RawInteractionResponse.

Source code in src/chattice/transports/http/response.py

@dataclass(frozen=True, slots=True)
class RequestConfigResponse:
    """Typed REQUEST_CONFIG response: ask the user to open an auth URL.

    The missing typed DX for a wire shape previously reachable only
    through RawInteractionResponse.
    """

    auth_url: str

    def to_dict(self) -> dict[str, object]:
        return {
            "actionResponse": {
                "type": "REQUEST_CONFIG",
                "authUrl": self.auth_url,
            }
        }

ResponseState

Bases: Enum

Whether the synchronous response has already been produced.

Source code in src/chattice/transports/http/response.py

class ResponseState(Enum):
    """Whether the synchronous response has already been produced."""

    NOT_RESPONDED = "not_responded"
    RESPONDED = "responded"

VerificationError

Bases: HTTPInteractionError

Incoming request verification failed.

Source code in src/chattice/transports/http/errors.py

class VerificationError(HTTPInteractionError):
    """Incoming request verification failed."""

WidgetAutocomplete dataclass

Typed UPDATE_WIDGET response: autocomplete suggestions for a widget.

Google's UPDATE_WIDGET response type answers a WIDGET_UPDATED autocomplete query. Each suggestion becomes a SelectionItem with the given text.

Source code in src/chattice/transports/http/response.py

@dataclass(frozen=True, slots=True)
class WidgetAutocomplete:
    """Typed UPDATE_WIDGET response: autocomplete suggestions for a widget.

    Google's UPDATE_WIDGET response type answers a WIDGET_UPDATED
    autocomplete query. Each suggestion becomes a SelectionItem with the
    given text.
    """

    widget_id: str
    suggestions: tuple[str, ...] = ()

    def to_dict(self) -> dict[str, object]:
        return {
            "actionResponse": {
                "type": "UPDATE_WIDGET",
                "updatedWidget": {
                    "widget": self.widget_id,
                    "suggestions": {
                        "items": [{"text": text} for text in self.suggestions]
                    },
                },
            }
        }

Pub/Sub push ingress: envelope -> interaction event, push verification.

The documented push envelope wraps the Chat interaction JSON in message.data (base64). Delivery has NO synchronous response channel — the push router acks with 2xx and ignores handler return values.

Authenticated Pub/Sub push sends an OIDC ID token from the configured push service account in the Authorization header (https://cloud.google.com/pubsub/docs/authenticate-push-subscriptions); GooglePubSubVerifier checks signature, audience, issuer, and the expected service-account email.

GooglePubSubVerifier dataclass

Verify authenticated Pub/Sub push requests.

The configured push endpoint receives an OIDC ID token issued for the Pub/Sub push service account. Signature/exp/aud are validated by google-auth's verify_token; the issuer must be accounts.google.com and the token's email claim must match service_account_email (REQUIRED: an audience match alone does not bind the publisher identity) with email_verified true.

Source code in src/chattice/transports/pubsub.py

@dataclass(frozen=True, slots=True)
class GooglePubSubVerifier:
    """Verify authenticated Pub/Sub push requests.

    The configured push endpoint receives an OIDC ID token issued for the
    Pub/Sub push service account. Signature/exp/aud are validated by
    google-auth's verify_token; the issuer must be accounts.google.com and
    the token's email claim must match ``service_account_email`` (REQUIRED:
    an audience match alone does not bind the publisher identity) with
    ``email_verified`` true.
    """

    audience: str
    service_account_email: str
    clock_skew_in_seconds: int = 10
    request: google_requests.Request | None = None

    def verify(self, incoming: IncomingRequest) -> None:
        token = extract_bearer(incoming)
        request = (
            self.request if self.request is not None else google_requests.Request()
        )
        try:
            claims = verify_token(
                token,
                request=request,
                audience=self.audience,
                clock_skew_in_seconds=self.clock_skew_in_seconds,
            )
        except google_auth_exceptions.TransportError as error:
            raise VerificationError(
                "Cannot reach Google issuer certificates"
            ) from error
        except (ValueError, google_auth_exceptions.GoogleAuthError) as error:
            raise VerificationError("Invalid bearer token") from error
        issuer = claims.get("iss")
        if issuer not in _GOOGLE_ACCOUNT_ISSUERS:
            raise VerificationError("Invalid token issuer")
        if claims.get("email_verified") is not True:
            raise VerificationError("Token email is not verified")
        if claims.get("email") != self.service_account_email:
            raise VerificationError(
                "Token service account does not match the configured push "
                "service account"
            )

MockPubSubVerifier

Accepts (or rejects) any push request; tests and local dev only.

Source code in src/chattice/transports/pubsub.py

class MockPubSubVerifier:
    """Accepts (or rejects) any push request; tests and local dev only."""

    def __init__(self, *, reject: bool = False) -> None:
        self._reject = reject

    def verify(self, request: IncomingRequest) -> None:
        if self._reject:
            raise VerificationError("Mock verifier rejected the request")

PubSubEnvelopeError

Bases: ValueError

The Pub/Sub push envelope is malformed.

Source code in src/chattice/transports/pubsub.py

class PubSubEnvelopeError(ValueError):
    """The Pub/Sub push envelope is malformed."""

PubSubPushAdapter

Decodes a documented Pub/Sub push envelope into a domain event.

Source code in src/chattice/transports/pubsub.py

class PubSubPushAdapter:
    """Decodes a documented Pub/Sub push envelope into a domain event."""

    def parse_envelope(self, payload: Mapping[str, object]) -> Event:
        """Validate the envelope, decode message.data, parse the interaction."""
        if not isinstance(payload, Mapping):
            raise PubSubEnvelopeError("Pub/Sub push payload must be a mapping")
        interaction = decode_message_data(payload)
        if interaction is None:
            raise PubSubEnvelopeError(
                "Pub/Sub push payload requires a 'message' object"
            )
        event = parse_interaction(interaction)
        # The FULL envelope (not just the inner interaction) stays in raw.
        return replace(event, raw=dict(payload))

parse_envelope(payload)

Validate the envelope, decode message.data, parse the interaction.

Source code in src/chattice/transports/pubsub.py

def parse_envelope(self, payload: Mapping[str, object]) -> Event:
    """Validate the envelope, decode message.data, parse the interaction."""
    if not isinstance(payload, Mapping):
        raise PubSubEnvelopeError("Pub/Sub push payload must be a mapping")
    interaction = decode_message_data(payload)
    if interaction is None:
        raise PubSubEnvelopeError(
            "Pub/Sub push payload requires a 'message' object"
        )
    event = parse_interaction(interaction)
    # The FULL envelope (not just the inner interaction) stays in raw.
    return replace(event, raw=dict(payload))

PubSubPushVerifier

Bases: Protocol

Contract: prove that an inbound push genuinely came from Pub/Sub.

Source code in src/chattice/transports/pubsub.py

class PubSubPushVerifier(Protocol):
    """Contract: prove that an inbound push genuinely came from Pub/Sub."""

    def verify(self, request: IncomingRequest) -> None:
        """Raise VerificationError when the request cannot be verified."""
        ...

verify(request)

Raise VerificationError when the request cannot be verified.

Source code in src/chattice/transports/pubsub.py

def verify(self, request: IncomingRequest) -> None:
    """Raise VerificationError when the request cannot be verified."""
    ...

decode_message_data(payload)

Decode a Pub/Sub push envelope into its inner JSON mapping.

Returns None when the payload is not a push envelope (e.g. a raw CloudEvent delivered to an HTTPS endpoint). Raises PubSubEnvelopeError for a malformed envelope.

Source code in src/chattice/transports/pubsub.py

def decode_message_data(payload: Mapping[str, object]) -> Mapping[str, object] | None:
    """Decode a Pub/Sub push envelope into its inner JSON mapping.

    Returns None when the payload is not a push envelope (e.g. a raw
    CloudEvent delivered to an HTTPS endpoint). Raises PubSubEnvelopeError
    for a malformed envelope.
    """
    message = payload.get("message")
    if message is None:
        return None
    if not isinstance(message, Mapping):
        raise PubSubEnvelopeError("Pub/Sub push payload requires a 'message' object")
    data = message.get("data")
    if not isinstance(data, str):
        raise PubSubEnvelopeError("'message.data' must be a base64 string")
    try:
        decoded = base64.b64decode(data, validate=True)
    except (binascii.Error, ValueError) as error:
        raise PubSubEnvelopeError("'message.data' is not valid base64") from error
    try:
        inner = json.loads(decoded)
    except (json.JSONDecodeError, UnicodeDecodeError) as error:
        raise PubSubEnvelopeError(
            "'message.data' does not contain valid JSON"
        ) from error
    if not isinstance(inner, Mapping):
        raise PubSubEnvelopeError("'message.data' must decode to a JSON object")
    return inner

FastAPI integration (optional extra chattice[fastapi]).

create_chat_router(dispatcher, verifier, *, path='/')

Build a POST route wiring Google Chat interactions into the dispatcher.

Works under FastAPI via app.include_router(...) and under plain Starlette via Starlette(routes=chat_router.routes).

Source code in src/chattice/integrations/fastapi/router.py

def create_chat_router(
    dispatcher: Dispatcher,
    verifier: IncomingRequestVerifier,
    *,
    path: str = "/",
) -> APIRouter:
    """Build a POST route wiring Google Chat interactions into the dispatcher.

    Works under FastAPI via app.include_router(...) and under plain Starlette
    via Starlette(routes=chat_router.routes).
    """
    adapter = HTTPInteractionAdapter()
    router = APIRouter()

    async def chat_endpoint(request: Request) -> Response:
        incoming = IncomingRequest(
            method=request.method,
            path=request.url.path,
            headers=dict(request.headers),
        )
        try:
            await asyncio.to_thread(verifier.verify, incoming)
        except VerificationError as error:
            logger.info(
                "verification failed: error=%s path=%s",
                type(error).__name__,
                incoming.path,
            )
            return Response(status_code=401)
        incoming = IncomingRequest(
            method=request.method,
            path=request.url.path,
            body=await request.body(),
            headers=incoming.headers,
            received_at=incoming.received_at,
        )
        try:
            event: Event = adapter.parse(incoming)
        except GoogleInteractionError as error:
            # The GoogleInteractionError hierarchy (Invalid/Unsupported/Conflicting)
            # intentionally collapses to 400: these payloads cannot be interpreted.
            # Unknown-but-valid future event types do NOT reach this path — they
            # become UnknownEvent and are dispatched normally (200 empty).
            # error CLASS only — pydantic validation messages embed
            # input_value and would leak attacker-controlled form data.
            logger.info(
                "invalid interaction payload: error=%s path=%s",
                type(error).__name__,
                incoming.path,
            )
            return JSONResponse(
                {"error": "invalid_interaction_payload"}, status_code=400
            )
        response = InteractionResponse()
        context = InteractionContext(
            request=incoming,
            response=response,
            received_at=incoming.received_at,
            deadline_at=incoming.received_at + SYNC_RESPONSE_DEADLINE,
        )
        started = time.monotonic()
        # The webhook surface is auth-less: handlers get exactly the
        # incoming/sync-response capability set (SYNC_RESPONSE + DIALOGS).
        capabilities = ResponseCapabilities.resolve(transport="http", event=event)
        try:
            result = await dispatcher.feed_update(
                event,
                request=incoming,
                response=response,
                interaction=context,
                capabilities=capabilities,
            )
        except Exception as error:
            # Exception CLASS only — never the message: handler messages may
            # contain secrets such as tokens or form values.
            logger.error(
                "handler failed: event_type=%s path=%s error=%s",
                event.event_type,
                incoming.path,
                type(error).__name__,
            )
            return Response(status_code=500)
        latency_ms = (time.monotonic() - started) * 1000
        if datetime.now(UTC) > context.deadline_at:
            logger.warning(
                "sync response deadline exceeded: event_type=%s latency_ms=%.1f",
                event.event_type,
                latency_ms,
            )
        logger.info(
            "interaction handled: event_type=%s latency_ms=%.1f",
            event.event_type,
            latency_ms,
        )
        payload = (
            response.payload if response.state is ResponseState.RESPONDED else result
        )
        try:
            return _serialize(payload, event=event)
        except TypeError as error:
            logger.error(
                "response serialization failed: event_type=%s error=%s",
                event.event_type,
                type(error).__name__,
            )
            return Response(status_code=500)

    # Plain starlette Route (not APIRoute) keeps the router usable under
    # Starlette(routes=chat_router.routes); FastAPI's include_router supports
    # mixed plain Routes and generates no request schema for the webhook.
    router.add_route(path, chat_endpoint, methods=["POST"])
    return router

create_pubsub_router(dispatcher, *, path='/pubsub', idempotency_storage=None, verifier=None, allow_unverified=False)

Push endpoint for Pub/Sub-delivered interactions (ack-only, 204).

Secure by default: pass verifier= (authenticated push) or an explicit allow_unverified=True (test/local environments). Without either the router refuses to be created.

Source code in src/chattice/integrations/fastapi/router.py

def create_pubsub_router(
    dispatcher: Dispatcher,
    *,
    path: str = "/pubsub",
    idempotency_storage: IdempotencyStorage | None = None,
    verifier: PubSubPushVerifier | None = None,
    allow_unverified: bool = False,
) -> APIRouter:
    """Push endpoint for Pub/Sub-delivered interactions (ack-only, 204).

    Secure by default: pass ``verifier=`` (authenticated push) or an
    explicit ``allow_unverified=True`` (test/local environments). Without
    either the router refuses to be created.
    """
    if verifier is None and not allow_unverified:
        raise ValueError(
            "push endpoints require verification: pass verifier= "
            "(authenticated push) or allow_unverified=True"
        )
    adapter = PubSubPushAdapter()
    router = APIRouter()
    delivery = _PushDeliveryCoordinator(idempotency_storage)

    async def pubsub_endpoint(request: Request) -> Response:
        if verifier is not None:
            try:
                await asyncio.to_thread(
                    verifier.verify,
                    IncomingRequest(
                        method=request.method,
                        path=request.url.path,
                        headers=dict(request.headers),
                    ),
                )
            except VerificationError as error:
                push_logger.info(
                    "push verification failed: error=%s", type(error).__name__
                )
                return Response(status_code=401)
        try:
            payload = json.loads(await request.body())
        except (json.JSONDecodeError, UnicodeDecodeError) as error:
            push_logger.info(
                "pubsub payload is not JSON: error=%s path=%s",
                type(error).__name__,
                request.url.path,
            )
            return Response(status_code=400)
        try:
            event = adapter.parse_envelope(payload)
        except (PubSubEnvelopeError, GoogleInteractionError) as error:
            push_logger.info(
                "invalid pubsub envelope: error=%s path=%s",
                type(error).__name__,
                request.url.path,
            )
            return Response(status_code=400)
        message_id, subscription = delivery.envelope_metadata(payload)
        short, dedupe_key, owner = await delivery.claim(
            message_id=message_id, subscription=subscription, label="pubsub"
        )
        if short is not None:
            return short
        # The push surface is ack-only: the response-channel capability
        # set is empty (no sync channel for handlers).
        capabilities = ResponseCapabilities.resolve(transport="pubsub", event=event)
        try:
            await dispatcher.feed_update(event, capabilities=capabilities)
        except Exception as error:
            push_logger.error(
                "pubsub handler failed: event_type=%s error=%s",
                event.event_type,
                type(error).__name__,
            )
            await delivery.release(
                message_id=message_id, dedupe_key=dedupe_key, owner=owner
            )
            return Response(status_code=500)
        await delivery.complete(
            message_id=message_id, dedupe_key=dedupe_key, owner=owner
        )
        return Response(status_code=204)

    router.add_route(path, pubsub_endpoint, methods=["POST"])
    return router

create_workspace_events_router(dispatcher, *, path='/workspace-events', idempotency_storage=None, verifier=None, allow_unverified=False)

Push endpoint for Workspace Events (ack-only, 204).

Google delivers Workspace Events exclusively as Pub/Sub push messages: the CloudEvents context attributes travel in message.attributes (ce-* keys) and message.data (base64) holds the event resource data. A structured CloudEvent POSTed directly is NOT a supported delivery mode and is rejected.

idempotency_storage dedupes redeliveries by the Pub/Sub message id with the same claim/complete/release semantics as the classic push router: a failed dispatch releases the claim so a redelivery re-dispatches.

Secure by default: pass verifier= or an explicit allow_unverified=True.

Source code in src/chattice/integrations/fastapi/router.py

def create_workspace_events_router(
    dispatcher: EventsDispatcher,
    *,
    path: str = "/workspace-events",
    idempotency_storage: IdempotencyStorage | None = None,
    verifier: PubSubPushVerifier | None = None,
    allow_unverified: bool = False,
) -> APIRouter:
    """Push endpoint for Workspace Events (ack-only, 204).

    Google delivers Workspace Events exclusively as Pub/Sub push messages:
    the CloudEvents context attributes travel in ``message.attributes``
    (``ce-*`` keys) and ``message.data`` (base64) holds the event resource
    data. A structured CloudEvent POSTed directly is NOT a supported
    delivery mode and is rejected.

    ``idempotency_storage`` dedupes redeliveries by the Pub/Sub message id
    with the same claim/complete/release semantics as the classic push
    router: a failed dispatch releases the claim so a redelivery
    re-dispatches.

    Secure by default: pass ``verifier=`` or an explicit
    ``allow_unverified=True``.
    """
    if verifier is None and not allow_unverified:
        raise ValueError(
            "push endpoints require verification: pass verifier= "
            "(authenticated push) or allow_unverified=True"
        )
    router = APIRouter()
    delivery = _PushDeliveryCoordinator(idempotency_storage)

    async def workspace_endpoint(request: Request) -> Response:
        if verifier is not None:
            try:
                await asyncio.to_thread(
                    verifier.verify,
                    IncomingRequest(
                        method=request.method,
                        path=request.url.path,
                        headers=dict(request.headers),
                    ),
                )
            except VerificationError as error:
                push_logger.info(
                    "push verification failed: error=%s", type(error).__name__
                )
                return Response(status_code=401)
        try:
            payload = json.loads(await request.body())
        except (json.JSONDecodeError, UnicodeDecodeError) as error:
            push_logger.info(
                "workspace event payload is not JSON: error=%s path=%s",
                type(error).__name__,
                request.url.path,
            )
            return Response(status_code=400)
        if not isinstance(payload, dict):
            push_logger.info("workspace event payload must be a JSON object")
            return Response(status_code=400)
        try:
            # Parse BEFORE claiming: an invalid envelope must never be
            # recorded as a completed delivery (it would be swallowed as a
            # duplicate on retry).
            event = parse_workspace_envelope(payload)
        except WorkspaceEventError as error:
            push_logger.info(
                "invalid workspace push envelope: error=%s path=%s",
                type(error).__name__,
                request.url.path,
            )
            return Response(status_code=400)
        message_id, subscription = delivery.envelope_metadata(payload)
        short, dedupe_key, owner = await delivery.claim(
            message_id=message_id, subscription=subscription, label="workspace"
        )
        if short is not None:
            return short
        try:
            await dispatcher.feed_event(event)
        except Exception as error:
            push_logger.error(
                "workspace handler failed: type=%s error=%s",
                event.cloud_type,
                type(error).__name__,
            )
            await delivery.release(
                message_id=message_id, dedupe_key=dedupe_key, owner=owner
            )
            return Response(status_code=500)
        await delivery.complete(
            message_id=message_id, dedupe_key=dedupe_key, owner=owner
        )
        return Response(status_code=204)

    router.add_route(path, workspace_endpoint, methods=["POST"])
    return router

Google Cloud Storage AssetPublisher (optional extra chattice[gcs]).

GCSAssetPublisher

Publish non-sensitive Card images to a preconfigured GCS bucket.

The bucket or public_url_base must already expose uploaded objects over anonymous HTTPS. This integration never changes bucket IAM. Cache metadata is opt-in and never grants public access to an object.

Source code in src/chattice/integrations/gcs/__init__.py

class GCSAssetPublisher:
    """Publish non-sensitive Card images to a preconfigured GCS bucket.

    The bucket or ``public_url_base`` must already expose uploaded objects
    over anonymous HTTPS. This integration never changes bucket IAM. Cache
    metadata is opt-in and never grants public access to an object.
    """

    def __init__(
        self,
        *,
        bucket: str,
        namespaces: Mapping[str, str] | None = None,
        default_namespace: str | None = None,
        client: Any = None,
        credentials: Any = None,
        project: str | None = None,
        public_url_base: str | None = None,
        upload_timeout: float = 60.0,
        cache_control: str | None = None,
    ) -> None:
        if not bucket or bucket != bucket.strip():
            raise ValueError("bucket must be a non-empty GCS bucket name")
        if upload_timeout <= 0:
            raise ValueError("upload_timeout must be positive")
        if client is not None and (credentials is not None or project is not None):
            raise ValueError(
                "client cannot be combined with credentials or project; configure "
                "the supplied client directly"
            )
        mapping = dict(namespaces or {})
        for namespace, prefix in mapping.items():
            if not namespace or namespace != namespace.strip():
                raise ValueError("namespace keys must be non-empty logical names")
            mapping[namespace] = _prefix(prefix)
        if default_namespace is not None and default_namespace not in mapping:
            raise ValueError("default_namespace must be a key in namespaces")
        if public_url_base is None:
            public_url_base = f"https://storage.googleapis.com/{quote(bucket, safe='')}"
        parsed_url_base = urlparse(public_url_base)
        if parsed_url_base.scheme != "https" or not parsed_url_base.netloc:
            raise ValueError("public_url_base must be an absolute HTTPS URL")

        if client is None:
            storage = _load_storage()
            client = storage.Client(project=project, credentials=credentials)
        self._bucket = client.bucket(bucket)
        self._namespaces = mapping
        self._default_namespace = default_namespace
        self._public_url_base = public_url_base.rstrip("/")
        self._upload_timeout = upload_timeout
        self._cache_control = cache_control

    def _object_name(self, filename: str, namespace: str | None) -> str:
        if not filename or Path(filename).name != filename or "\\" in filename:
            raise AssetPublishError("filename must be a non-empty basename")
        effective_namespace = (
            self._default_namespace if namespace is None else namespace
        )
        if effective_namespace is None:
            prefix = ""
        else:
            try:
                prefix = self._namespaces[effective_namespace]
            except KeyError as error:
                raise AssetPublishError(
                    f"Unknown GCS asset namespace {effective_namespace!r}"
                ) from error
        return f"{prefix}{uuid4().hex}-{filename}"

    async def publish(
        self,
        data: bytes,
        *,
        filename: str,
        content_type: str,
        namespace: str | None = None,
    ) -> str:
        """Upload bytes off the event loop and return their public HTTPS URL."""
        object_name = self._object_name(filename, namespace)
        blob = self._bucket.blob(object_name)
        if self._cache_control is not None:
            blob.cache_control = self._cache_control
        await asyncio.to_thread(
            blob.upload_from_string,
            data,
            content_type=content_type,
            timeout=self._upload_timeout,
        )
        return f"{self._public_url_base}/{quote(object_name, safe='/')}"

publish(data, *, filename, content_type, namespace=None) async

Upload bytes off the event loop and return their public HTTPS URL.

Source code in src/chattice/integrations/gcs/__init__.py

async def publish(
    self,
    data: bytes,
    *,
    filename: str,
    content_type: str,
    namespace: str | None = None,
) -> str:
    """Upload bytes off the event loop and return their public HTTPS URL."""
    object_name = self._object_name(filename, namespace)
    blob = self._bucket.blob(object_name)
    if self._cache_control is not None:
        blob.cache_control = self._cache_control
    await asyncio.to_thread(
        blob.upload_from_string,
        data,
        content_type=content_type,
        timeout=self._upload_timeout,
    )
    return f"{self._public_url_base}/{quote(object_name, safe='/')}"

Workspace Events

Workspace Events ingress family (separate from Chat interactions).

EventsDispatcher

Bases: EventsRouter

Independent feed for Workspace Events; never accepts interactions.

Source code in src/chattice/workspace_events/runtime.py

class EventsDispatcher(EventsRouter):
    """Independent feed for Workspace Events; never accepts interactions."""

    def __init__(self, *, name: str = "events_dispatcher") -> None:
        super().__init__(name=name)
        self._is_dispatcher = True

    async def feed_event(self, event: WorkspaceEvent, **context: object) -> object:
        if not isinstance(event, WorkspaceEvent):
            raise TypeError("feed_event() accepts WorkspaceEvent instances only")
        for router, middleware in self._walk():
            for handler in router.workspace_event.handlers:
                data = dict(context)
                try:
                    if not await _evaluate_filters(handler.filters, event, data):
                        continue
                    return await _invoke(handler, middleware, event, data)
                except SkipHandler:
                    continue
                except StopPropagation:
                    return None
        return None

EventsRouter

Router tree dedicated to Workspace resource-change events.

Source code in src/chattice/workspace_events/runtime.py

class EventsRouter:
    """Router tree dedicated to Workspace resource-change events."""

    def __init__(self, *, name: str | None = None) -> None:
        self.name = name or "events_router"
        if not self.name.strip():
            raise ValueError("EventsRouter name cannot be empty")
        self.workspace_event = _WorkspaceEventObserver()
        self.middleware = _EventsMiddlewareManager()
        self._parent: EventsRouter | None = None
        self._children: list[EventsRouter] = []
        self._is_dispatcher = False

    @property
    def parent(self) -> EventsRouter | None:
        return self._parent

    @property
    def children(self) -> tuple[EventsRouter, ...]:
        return tuple(self._children)

    def include_router(self, router: EventsRouter) -> EventsRouter:
        if not isinstance(router, EventsRouter):
            raise TypeError("include_router() requires an EventsRouter")
        if router is self:
            raise RouterConfigurationError("An EventsRouter cannot include itself")
        if router._is_dispatcher:
            raise RouterConfigurationError(
                "An EventsDispatcher cannot be attached as a child"
            )
        if router._parent is not None:
            raise RouterConfigurationError(
                f"EventsRouter {router.name!r} is already attached to "
                f"{router._parent.name!r}"
            )
        if router._contains(self):
            raise RouterConfigurationError(
                f"Including {router.name!r} in {self.name!r} would create a cycle"
            )
        router._parent = self
        self._children.append(router)
        return router

    def _contains(self, candidate: EventsRouter) -> bool:
        return self is candidate or any(
            child._contains(candidate) for child in self._children
        )

    def _walk(
        self, inherited: tuple[EventsMiddleware, ...] = ()
    ) -> Iterator[tuple[EventsRouter, tuple[EventsMiddleware, ...]]]:
        middleware = (*inherited, *tuple(self.middleware))
        yield self, middleware
        for child in tuple(self._children):
            yield from child._walk(middleware)

WorkspaceEvent dataclass

A Workspace resource-change event (NOT a Chat interaction).

Source code in src/chattice/workspace_events/parser.py

@dataclass(frozen=True, slots=True, kw_only=True)
class WorkspaceEvent:
    """A Workspace resource-change event (NOT a Chat interaction)."""

    event_type: str = field(default="workspace_event", init=False)
    event_id: str
    source: str
    subject: str | None = None
    event_time: datetime | None = None
    data: Mapping[str, object] = field(default_factory=dict)
    cloud_type: str = ""
    raw: Mapping[str, object] = field(default_factory=dict)

WorkspaceEventError

Bases: ValueError

The CloudEvent payload is malformed for Workspace Events.

Source code in src/chattice/workspace_events/envelope.py

class WorkspaceEventError(ValueError):
    """The CloudEvent payload is malformed for Workspace Events."""

WorkspaceEventType

Documented Chat Workspace event type strings (forward-compatible).

Sources: https://developers.google.com/workspace/events/guides/events-chat and https://developers.google.com/workspace/events/guides/events-lifecycle (including batch event types).

Source code in src/chattice/workspace_events/parser.py

class WorkspaceEventType:
    """Documented Chat Workspace event type strings (forward-compatible).

    Sources: https://developers.google.com/workspace/events/guides/events-chat
    and https://developers.google.com/workspace/events/guides/events-lifecycle
    (including batch event types).
    """

    MESSAGE_CREATED = "google.workspace.chat.message.v1.created"
    MESSAGE_UPDATED = "google.workspace.chat.message.v1.updated"
    MESSAGE_DELETED = "google.workspace.chat.message.v1.deleted"
    REACTION_CREATED = "google.workspace.chat.reaction.v1.created"
    REACTION_DELETED = "google.workspace.chat.reaction.v1.deleted"
    MEMBERSHIP_CREATED = "google.workspace.chat.membership.v1.created"
    MEMBERSHIP_UPDATED = "google.workspace.chat.membership.v1.updated"
    MEMBERSHIP_DELETED = "google.workspace.chat.membership.v1.deleted"
    SPACE_UPDATED = "google.workspace.chat.space.v1.updated"
    SPACE_DELETED = "google.workspace.chat.space.v1.deleted"
    SPACE_READ_STATE_UPDATED = "google.workspace.chat.spaceReadState.v1.updated"
    THREAD_READ_STATE_UPDATED = "google.workspace.chat.threadReadState.v1.updated"
    AVAILABILITY_UPDATED = "google.workspace.chat.availability.v1.updated"
    # Batch event types (output only): delivered automatically for any
    # subscribed type, never specified when creating a subscription.
    # No space.v1.batchDeleted / batch availability / batch read-state
    # creation-deletion types are documented.
    MESSAGE_BATCH_CREATED = "google.workspace.chat.message.v1.batchCreated"
    MESSAGE_BATCH_UPDATED = "google.workspace.chat.message.v1.batchUpdated"
    MESSAGE_BATCH_DELETED = "google.workspace.chat.message.v1.batchDeleted"
    REACTION_BATCH_CREATED = "google.workspace.chat.reaction.v1.batchCreated"
    REACTION_BATCH_DELETED = "google.workspace.chat.reaction.v1.batchDeleted"
    MEMBERSHIP_BATCH_CREATED = "google.workspace.chat.membership.v1.batchCreated"
    MEMBERSHIP_BATCH_UPDATED = "google.workspace.chat.membership.v1.batchUpdated"
    MEMBERSHIP_BATCH_DELETED = "google.workspace.chat.membership.v1.batchDeleted"
    SPACE_BATCH_UPDATED = "google.workspace.chat.space.v1.batchUpdated"
    SPACE_READ_STATE_BATCH_UPDATED = (
        "google.workspace.chat.spaceReadState.v1.batchUpdated"
    )
    THREAD_READ_STATE_BATCH_UPDATED = (
        "google.workspace.chat.threadReadState.v1.batchUpdated"
    )
    SUBSCRIPTION_SUSPENDED = "google.workspace.events.subscription.v1.suspended"
    SUBSCRIPTION_EXPIRATION_REMINDER = (
        "google.workspace.events.subscription.v1.expirationReminder"
    )
    SUBSCRIPTION_EXPIRED = "google.workspace.events.subscription.v1.expired"

parse_workspace_envelope(payload)

Parse an official Pub/Sub push envelope for Workspace Events.

Google's binding (https://developers.google.com/workspace/events): the CloudEvents context attributes travel in message.attributes (ce-id, ce-source, ce-specversion, ce-time, ce-type, optional ce-subject/ce-datacontenttype), while base64-decoded message.data contains ONLY the event resource data (or resource names for names-only payloads). The full envelope is preserved in .raw.

Source code in src/chattice/workspace_events/parser.py

def parse_workspace_envelope(payload: Mapping[str, object]) -> WorkspaceEvent:
    """Parse an official Pub/Sub push envelope for Workspace Events.

    Google's binding (https://developers.google.com/workspace/events):
    the CloudEvents context attributes travel in ``message.attributes``
    (``ce-id``, ``ce-source``, ``ce-specversion``, ``ce-time``,
    ``ce-type``, optional ``ce-subject``/``ce-datacontenttype``), while
    base64-decoded ``message.data`` contains ONLY the event resource data
    (or resource names for names-only payloads). The full envelope is
    preserved in ``.raw``.
    """
    if not isinstance(payload, Mapping):
        raise WorkspaceEventError("Workspace event payload must be a mapping")
    message = payload.get("message")
    if message is None or not isinstance(message, Mapping):
        raise WorkspaceEventError(
            "Workspace push payload must be a Pub/Sub envelope with a 'message' object"
        )
    attributes = message.get("attributes")
    if attributes is None or not isinstance(attributes, Mapping):
        raise WorkspaceEventError(
            "Workspace push envelope requires 'message.attributes' (ce-* fields)"
        )
    for required in _REQUIRED_CE_ATTRIBUTES:
        value = attributes.get(required)
        if not isinstance(value, str) or not value:
            raise WorkspaceEventError(
                f"Workspace push envelope requires attribute {required!r}"
            )
    specversion = attributes["ce-specversion"]
    if specversion != REQUIRED_SPECVERSION:
        raise WorkspaceEventError(
            f"Unsupported specversion {specversion!r}; "
            f"expected {REQUIRED_SPECVERSION!r}"
        )
    cloud_type = attributes["ce-type"]
    if not cloud_type.startswith(TYPE_PREFIX):
        raise WorkspaceEventError(
            f"Workspace event 'type' must be a {TYPE_PREFIX!r}-prefixed string"
        )
    datacontenttype = attributes.get("ce-datacontenttype")
    if datacontenttype is not None and datacontenttype != "application/json":
        raise WorkspaceEventError(
            f"Unsupported ce-datacontenttype {datacontenttype!r}; "
            "expected 'application/json'"
        )
    subject = attributes.get("ce-subject")
    if subject is not None and not isinstance(subject, str):
        raise WorkspaceEventError(
            "Workspace event 'ce-subject' must be a string or absent"
        )
    event_time = parse_event_time(attributes.get("ce-time"))
    data = _decode_envelope_data(message)
    return WorkspaceEvent(
        event_id=attributes["ce-id"],
        source=attributes["ce-source"],
        subject=subject,
        event_time=event_time,
        data=MappingProxyType(dict(data)),
        cloud_type=cloud_type,
        raw=MappingProxyType(dict(payload)),
    )

parse_workspace_event(payload)

Parse a CloudEvents 1.0 Workspace event into a WorkspaceEvent.

Accepts a STRUCTURED CloudEvent (all fields at the top level). This form is for offline use (fixtures, replays, tests) — Google delivers Workspace Events exclusively through Pub/Sub push messages; see :func:parse_workspace_envelope for the wire format.

Source code in src/chattice/workspace_events/parser.py

def parse_workspace_event(payload: Mapping[str, object]) -> WorkspaceEvent:
    """Parse a CloudEvents 1.0 Workspace event into a WorkspaceEvent.

    Accepts a STRUCTURED CloudEvent (all fields at the top level). This form
    is for offline use (fixtures, replays, tests) — Google delivers Workspace
    Events exclusively through Pub/Sub push messages; see
    :func:`parse_workspace_envelope` for the wire format.
    """
    if not isinstance(payload, Mapping):
        raise WorkspaceEventError("Workspace event payload must be a mapping")
    if payload.get("specversion") != REQUIRED_SPECVERSION:
        raise WorkspaceEventError(
            f"Unsupported specversion {payload.get('specversion')!r}; "
            f"expected {REQUIRED_SPECVERSION!r}"
        )
    event_id = payload.get("id")
    if not isinstance(event_id, str):
        raise WorkspaceEventError("Workspace event 'id' must be a string")
    source = payload.get("source")
    if not isinstance(source, str):
        raise WorkspaceEventError("Workspace event 'source' must be a string")
    cloud_type = payload.get("type")
    if not isinstance(cloud_type, str) or not cloud_type.startswith(TYPE_PREFIX):
        raise WorkspaceEventError(
            f"Workspace event 'type' must be a {TYPE_PREFIX!r}-prefixed string"
        )
    subject = payload.get("subject")
    if subject is not None and not isinstance(subject, str):
        raise WorkspaceEventError(
            "Workspace event 'subject' must be a string or absent"
        )
    event_time = parse_event_time(payload.get("time"))
    data = payload.get("data")
    if data is None:
        data = {}
    if not isinstance(data, Mapping):
        raise WorkspaceEventError("Workspace event 'data' must be a mapping or absent")
    # deep snapshots — mutating the caller's nested values (e.g.
    # data.message.text) after parsing must not change the parsed event.
    return WorkspaceEvent(
        event_id=event_id,
        source=source,
        subject=subject,
        event_time=event_time,
        data=cast(
            Mapping[str, object],
            MappingProxyType(
                cast(
                    dict[str, object], deep_snapshot(data, where="WorkspaceEvent.data")
                )
            ),
        ),
        cloud_type=cloud_type,
        raw=cast(
            Mapping[str, object],
            MappingProxyType(
                cast(
                    dict[str, object],
                    deep_snapshot(payload, where="WorkspaceEvent.raw"),
                )
            ),
        ),
    )

Experimental namespace

chattice.experimental carries optional integration contracts (e.g. chattice.experimental.ai). The namespace has NO compatibility promise; use it explicitly and pin the package version.

Experimental (Developer Preview) features live here and ONLY here.

Core flows must never depend on this namespace; APIs in this package may change or disappear without notice. See docs/architecture/capabilities.md.

Testing

Testing toolkit: mocks, factories, assertions (first-party testing toolkit use).

EventFactory

Static builders producing frozen domain events directly.

user/space accept either a reference or a Google resource name string (e.g. "users/user-a"); None uses the test defaults.

Source code in src/chattice/testing/event_factory.py

class EventFactory:
    """Static builders producing frozen domain events directly.

    ``user``/``space`` accept either a reference or a Google resource name
    string (e.g. ``"users/user-a"``); ``None`` uses the test defaults.
    """

    @staticmethod
    def message(
        text: str,
        *,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> MessageEvent:
        """Build a chat message event."""
        return MessageEvent(
            text=text,
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

    @staticmethod
    def action(
        name: str,
        parameters: Mapping[str, object] | None = None,
        *,
        form_inputs: FormInputs | None = None,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        dialog: DialogMetadata | None = None,
        event_time: datetime | None = None,
    ) -> ActionEvent:
        """Build a named action event."""
        return ActionEvent(
            name=name,
            parameters=parameters if parameters is not None else {},
            form_inputs=form_inputs if form_inputs is not None else FormInputs(),
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
            dialog=dialog,
        )

    @staticmethod
    def added_to_space(
        *,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> AddedToSpaceEvent:
        """Build an app-added-to-space event."""
        return AddedToSpaceEvent(
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

    @staticmethod
    def removed_from_space(
        *,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> RemovedFromSpaceEvent:
        """Build an app-removed-from-space event."""
        return RemovedFromSpaceEvent(
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

    @staticmethod
    def app_home(
        *,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> AppHomeEvent:
        """Build an App Home open event."""
        return AppHomeEvent(
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

    @staticmethod
    def form_submit(
        function_name: str,
        *,
        parameters: Mapping[str, str] | None = None,
        form_inputs: FormInputs | None = None,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> FormSubmitEvent:
        """Build a form submission event from App Home."""
        return FormSubmitEvent(
            function_name=function_name,
            parameters=parameters if parameters is not None else {},
            form_inputs=form_inputs if form_inputs is not None else FormInputs(),
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

    @staticmethod
    def workspace_event(
        cloud_type: str,
        *,
        event_id: str = "evt-test",
        source: str = "//chat.googleapis.com/test",
        subject: str | None = None,
        event_time: datetime | None = None,
        data: Mapping[str, object] | None = None,
    ) -> WorkspaceEvent:
        """Build a Workspace resource-change event."""
        return WorkspaceEvent(
            event_id=event_id,
            source=source,
            subject=subject,
            event_time=event_time,
            data=data if data is not None else {},
            cloud_type=cloud_type,
        )

    @staticmethod
    def unknown_event(
        original_type: str,
        *,
        user: UserRef | str | None = None,
        space: SpaceRef | str | None = None,
        thread: ThreadRef | None = None,
        event_time: datetime | None = None,
    ) -> UnknownEvent:
        """Build an event of an unrecognized external type."""
        return UnknownEvent(
            original_type=original_type,
            event_time=event_time,
            actor=_user(user),
            space=_space(space),
            thread=thread,
        )

action(name, parameters=None, *, form_inputs=None, user=None, space=None, thread=None, dialog=None, event_time=None) staticmethod

Build a named action event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def action(
    name: str,
    parameters: Mapping[str, object] | None = None,
    *,
    form_inputs: FormInputs | None = None,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    dialog: DialogMetadata | None = None,
    event_time: datetime | None = None,
) -> ActionEvent:
    """Build a named action event."""
    return ActionEvent(
        name=name,
        parameters=parameters if parameters is not None else {},
        form_inputs=form_inputs if form_inputs is not None else FormInputs(),
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
        dialog=dialog,
    )

added_to_space(*, user=None, space=None, thread=None, event_time=None) staticmethod

Build an app-added-to-space event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def added_to_space(
    *,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> AddedToSpaceEvent:
    """Build an app-added-to-space event."""
    return AddedToSpaceEvent(
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

app_home(*, user=None, space=None, thread=None, event_time=None) staticmethod

Build an App Home open event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def app_home(
    *,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> AppHomeEvent:
    """Build an App Home open event."""
    return AppHomeEvent(
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

form_submit(function_name, *, parameters=None, form_inputs=None, user=None, space=None, thread=None, event_time=None) staticmethod

Build a form submission event from App Home.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def form_submit(
    function_name: str,
    *,
    parameters: Mapping[str, str] | None = None,
    form_inputs: FormInputs | None = None,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> FormSubmitEvent:
    """Build a form submission event from App Home."""
    return FormSubmitEvent(
        function_name=function_name,
        parameters=parameters if parameters is not None else {},
        form_inputs=form_inputs if form_inputs is not None else FormInputs(),
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

message(text, *, user=None, space=None, thread=None, event_time=None) staticmethod

Build a chat message event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def message(
    text: str,
    *,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> MessageEvent:
    """Build a chat message event."""
    return MessageEvent(
        text=text,
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

removed_from_space(*, user=None, space=None, thread=None, event_time=None) staticmethod

Build an app-removed-from-space event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def removed_from_space(
    *,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> RemovedFromSpaceEvent:
    """Build an app-removed-from-space event."""
    return RemovedFromSpaceEvent(
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

unknown_event(original_type, *, user=None, space=None, thread=None, event_time=None) staticmethod

Build an event of an unrecognized external type.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def unknown_event(
    original_type: str,
    *,
    user: UserRef | str | None = None,
    space: SpaceRef | str | None = None,
    thread: ThreadRef | None = None,
    event_time: datetime | None = None,
) -> UnknownEvent:
    """Build an event of an unrecognized external type."""
    return UnknownEvent(
        original_type=original_type,
        event_time=event_time,
        actor=_user(user),
        space=_space(space),
        thread=thread,
    )

workspace_event(cloud_type, *, event_id='evt-test', source='//chat.googleapis.com/test', subject=None, event_time=None, data=None) staticmethod

Build a Workspace resource-change event.

Source code in src/chattice/testing/event_factory.py

@staticmethod
def workspace_event(
    cloud_type: str,
    *,
    event_id: str = "evt-test",
    source: str = "//chat.googleapis.com/test",
    subject: str | None = None,
    event_time: datetime | None = None,
    data: Mapping[str, object] | None = None,
) -> WorkspaceEvent:
    """Build a Workspace resource-change event."""
    return WorkspaceEvent(
        event_id=event_id,
        source=source,
        subject=subject,
        event_time=event_time,
        data=data if data is not None else {},
        cloud_type=cloud_type,
    )

FakeChatTransport

Bases: ChatServiceTransport

Subclass of the SDK transport; only the methods Bot uses are real.

Mirrors the gapic transport contract: the client invokes self._transport._wrapped_methods[method](request, retry=..., timeout=..., metadata=...) and reads self._transport.host.

Source code in src/chattice/testing/fake_transport.py

class FakeChatTransport(transports.ChatServiceTransport):
    """Subclass of the SDK transport; only the methods Bot uses are real.

    Mirrors the gapic transport contract: the client invokes
    ``self._transport._wrapped_methods[method](request, retry=..., timeout=...,
    metadata=...)`` and reads ``self._transport.host``.
    """

    _matching_host = "chat.googleapis.com"

    def __init__(
        self, error: Exception | None = None, *, credentials: object = None
    ) -> None:
        # The gapic client requires a transport instance to carry credentials
        # itself ("When providing a transport instance, provide its credentials
        # directly") — we accept and store them without using them.
        self.credentials = credentials
        self.error = error
        self.messages: dict[str, Message] = {}
        self.spaces: dict[str, Space] = {}
        self.memberships: dict[str, Membership] = {}
        self.attachments: dict[str, Attachment] = {}
        self.requests: list[CreateMessageRequest] = []
        self.updates: list[UpdateMessageRequest] = []
        self.timeouts: list[object] = []
        self.membership_creates: list[CreateMembershipRequest] = []
        self.membership_gets: list[GetMembershipRequest] = []
        self.membership_lists: list[ListMembershipsRequest] = []
        self.membership_deletes: list[DeleteMembershipRequest] = []
        self.space_lists: list[ListSpacesRequest] = []
        self.list_page_size = 100
        self.delay: float = 0.0
        self.calls: list[int] = []
        # Keys MUST be bound methods: the client looks up
        # self._transport._wrapped_methods[self._transport.create_message].
        self._wrapped_methods = {
            self.create_message: self._wrap(self.create_message),
            self.get_message: self._wrap(self.get_message),
            self.update_message: self._wrap(self.update_message),
            self.delete_message: self._wrap(self.delete_message),
            self.get_space: self._wrap(self.get_space),
            self.list_spaces: self._wrap(self.list_spaces),
            self.create_membership: self._wrap(self.create_membership),
            self.get_membership: self._wrap(self.get_membership),
            self.list_memberships: self._wrap(self.list_memberships),
            self.delete_membership: self._wrap(self.delete_membership),
            self.get_attachment: self._wrap(self.get_attachment),
        }

    @property
    def host(self) -> str:
        return "fake-chat"

    def _wrap(
        self, method: Callable[..., Awaitable[object]]
    ) -> Callable[..., Awaitable[object]]:
        async def wrapped(
            request: object,
            *,
            retry: object = None,
            timeout: object = None,
            metadata: object = (),
        ) -> object:
            return await method(
                request, retry=retry, timeout=timeout, metadata=metadata
            )

        return wrapped

    def _check_error(self) -> None:
        if self.error is not None:
            raise self.error

    async def create_message(
        self,
        request: CreateMessageRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Message:
        # Record the attempt on entry so a call that fails (error configured)
        # is still counted: the framework made exactly one transport call.
        self.calls.append(1)
        self.timeouts.append(timeout)
        self._check_error()
        if self.delay:
            await asyncio.sleep(self.delay)
        self.requests.append(request)
        if request.message_id:
            name = f"{request.parent}/messages/{request.message_id}"
        else:
            name = f"{request.parent}/messages/fake-{len(self.requests)}"
        message = Message(name=name, text=request.message.text)
        if request.message.thread.name:
            message.thread.name = request.message.thread.name
        self.messages[name] = message
        return message

    async def get_message(
        self,
        request: GetMessageRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Message:
        self._check_error()
        message = self.messages.get(request.name)
        if message is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"message not found: {request.name}")  # type: ignore[no-untyped-call]
        return message

    async def update_message(
        self,
        request: UpdateMessageRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Message:
        self._check_error()
        self.updates.append(request)
        current = self.messages.get(request.message.name)
        if current is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"message not found: {request.message.name}")  # type: ignore[no-untyped-call]
        updated = Message(name=current.name, text=request.message.text)
        self.messages[current.name] = updated
        return updated

    async def delete_message(
        self,
        request: DeleteMessageRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> None:
        self._check_error()
        self.messages.pop(request.name, None)

    async def get_space(
        self,
        request: GetSpaceRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Space:
        self._check_error()
        space = self.spaces.get(request.name)
        if space is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"space not found: {request.name}")  # type: ignore[no-untyped-call]
        return space

    async def list_spaces(
        self,
        request: ListSpacesRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> ListSpacesResponse:
        self._check_error()
        self.space_lists.append(ListSpacesRequest(request))
        values = [self.spaces[name] for name in sorted(self.spaces)]
        offset = int(request.page_token or "0")
        end = min(offset + self.list_page_size, len(values))
        next_token = str(end) if end < len(values) else ""
        return ListSpacesResponse(
            spaces=values[offset:end],
            next_page_token=next_token,
        )

    async def create_membership(
        self,
        request: CreateMembershipRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Membership:
        self._check_error()
        self.membership_creates.append(CreateMembershipRequest(request))
        member_id = request.membership.member.name.removeprefix("users/")
        membership = Membership(request.membership)
        membership.name = f"{request.parent}/members/{member_id}"
        self.memberships[membership.name] = membership
        return membership

    async def get_membership(
        self,
        request: GetMembershipRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Membership:
        self._check_error()
        self.membership_gets.append(GetMembershipRequest(request))
        membership = self.memberships.get(request.name)
        if membership is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"membership not found: {request.name}")  # type: ignore[no-untyped-call]
        return membership

    async def list_memberships(
        self,
        request: ListMembershipsRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> ListMembershipsResponse:
        self._check_error()
        self.membership_lists.append(ListMembershipsRequest(request))
        prefix = f"{request.parent}/members/"
        values = [
            self.memberships[name]
            for name in sorted(self.memberships)
            if name.startswith(prefix)
        ]
        offset = int(request.page_token or "0")
        end = min(offset + self.list_page_size, len(values))
        next_token = str(end) if end < len(values) else ""
        return ListMembershipsResponse(
            memberships=values[offset:end],
            next_page_token=next_token,
        )

    async def delete_membership(
        self,
        request: DeleteMembershipRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Membership:
        self._check_error()
        self.membership_deletes.append(DeleteMembershipRequest(request))
        membership = self.memberships.pop(request.name, None)
        if membership is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"membership not found: {request.name}")  # type: ignore[no-untyped-call]
        return membership

    async def get_attachment(
        self,
        request: GetAttachmentRequest,
        *,
        retry: object = None,
        timeout: object = None,
        metadata: object = (),
    ) -> Attachment:
        self._check_error()
        attachment = self.attachments.get(request.name)
        if attachment is None:
            from google.api_core import exceptions

            raise exceptions.NotFound(f"attachment not found: {request.name}")  # type: ignore[no-untyped-call]
        return attachment

MockBot

Records outgoing calls and fabricates SDK proto responses.

No transport, no network. Handlers call the same facade as the real client (bot.app.messages.create(...), DI-compatible: handlers receive it by name (feed_update(event, bot=mock_bot))). Recorded call kinds are the recording layer's names (send_message, update_message); subclasses may override those recorders.

Source code in src/chattice/testing/mock_bot.py

class MockBot:
    """Records outgoing calls and fabricates SDK proto responses.

    No transport, no network. Handlers call the same facade as the real
    client (``bot.app.messages.create(...)``, DI-compatible: handlers
    receive it by name (``feed_update(event, bot=mock_bot)``)). Recorded
    call kinds are the recording layer's names (``send_message``,
    ``update_message``); subclasses may override those recorders.
    """

    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, Any]]] = []

    @property
    def app(self) -> _MockIdentity:
        return _MockIdentity(self)

    @property
    def user(self) -> _MockIdentity:
        return _MockIdentity(self)

    async def send_message(
        self,
        space: Any,
        text: str | None = None,
        *,
        thread: Any = None,
        reply_option: Any = None,
        request_id: str | None = None,
        message_id: str | None = None,
        timeout: float | None = None,
        markup_syntax: Any = None,
        accessory_widgets: Any = None,
        card: Any = None,
        notify: Any = None,
        private_to: Any = None,
        attachments: Any = None,
    ) -> Message:
        parent = space if isinstance(space, str) else space.name
        self.calls.append(
            (
                "send_message",
                {
                    "space": parent,
                    "text": text,
                    "card": card.to_dict() if card is not None else None,
                    "notify": notify,
                    "private_to": (
                        private_to.name if hasattr(private_to, "name") else private_to
                    ),
                    "attachments": (
                        [a.filename for a in attachments] if attachments else None
                    ),
                },
            )
        )
        name = f"{parent}/messages/{len(self.calls)}"
        message = Message(name=name, text=text or "")
        if markup_syntax is not None:
            self.calls[-1][1]["markup_syntax"] = markup_syntax
            message.markup_syntax = markup_syntax
        return message

    async def upload_attachment(
        self, space: Any, file: Any, *, timeout: float | None = None
    ) -> UploadedAttachment:
        parent = space if isinstance(space, str) else space.name
        self.calls.append(
            (
                "upload_attachment",
                {"space": parent, "filename": file.filename},
            )
        )
        return UploadedAttachment(
            space=parent,
            filename=file.filename,
            attachment_data_ref={
                "resourceName": (f"{parent}/attachments/upload/{len(self.calls)}")
            },
        )

    async def download_attachment(
        self,
        attachment: Any,
        *,
        destination: str | Path | None = None,
        timeout: float | None = None,
    ) -> bytes | Path:
        name = (
            attachment.resource_name
            if hasattr(attachment, "resource_name")
            else str(attachment)
        )
        self.calls.append(("download_attachment", {"resource_name": name}))
        if destination is None:
            return b""
        path = Path(destination)
        path.write_bytes(b"")  # noqa: ASYNC240 — test stub, no real I/O guard
        return path

    async def get_attachment(
        self, name: str, *, timeout: float | None = None
    ) -> AttachmentRef:
        self.calls.append(("get_attachment", {"name": name}))
        return AttachmentRef(
            name=name,
            source=AttachmentSource.UPLOADED_CONTENT,
            attachment_data_ref={"resourceName": name},
        )

    async def get_message(self, name: str, *, timeout: float | None = None) -> Message:
        self.calls.append(("get_message", {"name": name}))
        return Message(name=name, text="")

    async def update_message(
        self,
        name: str,
        text: str | None = None,
        *,
        card: Any = None,
        timeout: float | None = None,
    ) -> Message:
        # Mirrors the real messages.update(name, text=None, *, card=...):
        # the card path records the card payload (the mirror must stay
        # in sync with the real client, including card=).
        self.calls.append(
            (
                "update_message",
                {
                    "name": name,
                    "text": text,
                    "card": card.to_dict() if card is not None else None,
                },
            )
        )
        return Message(name=name, text=text or "")

    async def delete_message(self, name: str, *, timeout: float | None = None) -> None:
        self.calls.append(("delete_message", {"name": name}))

    async def get_space(self, name: str, *, timeout: float | None = None) -> Space:
        self.calls.append(("get_space", {"name": name}))
        return Space(name=name)

    def _sent_texts(self) -> list[str]:
        return [
            args["text"] or "" for kind, args in self.calls if kind == "send_message"
        ]

    def assert_message_sent(self, text: str | None = None, *, count: int = 1) -> None:
        """Assert send_message was called `count` times (optionally with text)."""
        sent = self._sent_texts()
        if len(sent) != count:
            raise AssertionError(
                f"expected {count} sent message(s), got {len(sent)}: {sent!r}"
            )
        if text is not None and sent != [text] * count:
            raise AssertionError(f"expected sent text {text!r} x{count}, got {sent!r}")

    def assert_updated(self, name: str, text: str) -> None:
        """Assert update_message was called with the given name/text."""
        for kind, args in self.calls:
            if (
                kind == "update_message"
                and args.get("name") == name
                and args.get("text") == text
            ):
                return
        raise AssertionError(
            f"expected update_message({name!r}, {text!r}), "
            f"calls: {[(k, a) for k, a in self.calls if k == 'update_message']!r}"
        )

    def assert_no_messages(self) -> None:
        """Assert nothing was sent."""
        sent = self._sent_texts()
        if sent:
            raise AssertionError(f"expected no sent messages, got {sent!r}")

assert_message_sent(text=None, *, count=1)

Assert send_message was called count times (optionally with text).

Source code in src/chattice/testing/mock_bot.py

def assert_message_sent(self, text: str | None = None, *, count: int = 1) -> None:
    """Assert send_message was called `count` times (optionally with text)."""
    sent = self._sent_texts()
    if len(sent) != count:
        raise AssertionError(
            f"expected {count} sent message(s), got {len(sent)}: {sent!r}"
        )
    if text is not None and sent != [text] * count:
        raise AssertionError(f"expected sent text {text!r} x{count}, got {sent!r}")

assert_no_messages()

Assert nothing was sent.

Source code in src/chattice/testing/mock_bot.py

def assert_no_messages(self) -> None:
    """Assert nothing was sent."""
    sent = self._sent_texts()
    if sent:
        raise AssertionError(f"expected no sent messages, got {sent!r}")

assert_updated(name, text)

Assert update_message was called with the given name/text.

Source code in src/chattice/testing/mock_bot.py

def assert_updated(self, name: str, text: str) -> None:
    """Assert update_message was called with the given name/text."""
    for kind, args in self.calls:
        if (
            kind == "update_message"
            and args.get("name") == name
            and args.get("text") == text
        ):
            return
    raise AssertionError(
        f"expected update_message({name!r}, {text!r}), "
        f"calls: {[(k, a) for k, a in self.calls if k == 'update_message']!r}"
    )

assert_card_has_button(card, *, action=None, text=None)

Assert the card contains a button matching the given action/text.

Source code in src/chattice/testing/assertions.py

def assert_card_has_button(
    card: Card, *, action: str | None = None, text: str | None = None
) -> None:
    """Assert the card contains a button matching the given action/text."""
    buttons = _buttons(card)
    for button in buttons:
        if action is not None and button.action != action:
            continue
        if text is not None and button.text != text:
            continue
        return
    wanted = f"action={action!r}" if action else f"text={text!r}"
    raise AssertionError(f"no button with {wanted} found on the card")

assert_card_header(card, *, title=None, subtitle=None)

Assert the card header matches the given fields.

Source code in src/chattice/testing/assertions.py

def assert_card_header(
    card: Card, *, title: str | None = None, subtitle: str | None = None
) -> None:
    """Assert the card header matches the given fields."""
    header = card.header
    if header is None:
        raise AssertionError("card has no header")
    if title is not None and header.title != title:
        raise AssertionError(f"expected header title {title!r}, got {header.title!r}")
    if subtitle is not None and header.subtitle != subtitle:
        raise AssertionError(
            f"expected header subtitle {subtitle!r}, got {header.subtitle!r}"
        )

set_state_for(storage, key, state) async

Seed a workflow state for an application test.

Source code in src/chattice/testing/fsm.py

async def set_state_for(storage: BaseStorage, key: StorageKey, state: State) -> None:
    """Seed a workflow state for an application test."""
    await storage.set_state(key, state.state)