Chat Module

Service

unique_toolkit.services.chat_service

ChatService

Bases: ChatServiceDeprecated

Provides all functionalities to manage the chat session.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

class ChatService(ChatServiceDeprecated):
    """Provides all functionalities to manage the chat session."""

    @deprecated(
        "Use UniqueContext.from_chat_event(event) with UniqueServiceFactory instead."
    )
    def __init__(
        self,
        event: "ChatEvent",
        content_scope_chat_id: str | None = None,
    ) -> None:
        """Initialize ChatService from an event (deprecated). Use :meth:`from_context` instead."""
        super().__init__(event, content_scope_chat_id)
        self._elicitation_service: ElicitationService | None = None
        self._elicitation_message_id: str | None = None
        self._cancellation_watcher: CancellationWatcher = CancellationWatcher(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._chat_id,
            assistant_message_id=self._assistant_message_id,
        )

    @classmethod
    def from_settings(cls, settings: UniqueSettings, **kwargs: Any) -> Self:
        """Create a ChatService from a :class:`UniqueSettings` or :class:`UniqueContext`.

        This is the preferred constructor when using the service factory pattern.

        Args:
            settings: Either a :class:`UniqueSettings` (whose ``context`` is used)
                or a :class:`UniqueContext` carrying auth and chat information directly.

        Raises:
            ValueError: If the resolved context has no chat or ``last_user_message_id`` is missing.
        """
        return cls.from_context(context=settings.context)

    @classmethod
    def from_context(cls, context: UniqueContext) -> Self:
        if context.chat is None:
            raise ValueError(
                "ChatService requires a chat context (context.chat is None)"
            )
        chat = context.chat
        user_message_id = chat.last_user_message_id
        instance: ChatService = cls.__new__(cls)
        instance._event = None
        instance._company_id = context.auth.get_confidential_company_id()
        instance._user_id = context.auth.get_confidential_user_id()
        instance._chat_id = chat.chat_id
        instance._assistant_id = chat.assistant_id
        instance._assistant_message_id = chat.last_assistant_message_id
        instance._user_message_id = user_message_id
        instance._user_message_text = chat.last_user_message_text
        instance._content_scope_chat_id = chat.parent_chat_id or chat.chat_id
        instance._elicitation_service = None
        instance._elicitation_message_id = chat.last_assistant_message_id
        instance._cancellation_watcher = CancellationWatcher(
            user_id=instance._user_id,
            company_id=instance._company_id,
            chat_id=instance._chat_id,
            assistant_message_id=instance._assistant_message_id,
        )
        return instance

    @property
    def cancellation(self) -> CancellationWatcher:
        """Cancellation watcher for this chat session."""
        return self._cancellation_watcher

    @property
    def elicitation(self) -> ElicitationService:
        """Get the ElicitationService for this chat session."""
        if self._elicitation_service is not None:
            return self._elicitation_service

        if self._event is not None:
            self._elicitation_service = ElicitationService.from_chat_event(self._event)
        else:
            self._elicitation_service = ElicitationService(
                user_id=self._user_id,
                company_id=self._company_id,
                chat_id=self._content_scope_chat_id,
                message_id=self._elicitation_message_id,
            )

        return self._elicitation_service

    def get_debug_info(self) -> dict[str, Any]:
        """Retrieves the debug information from the current user message.

        Returns:
            dict[str, Any]: The debug information stored on the user message,
                or an empty dict if retrieval fails or no debug info exists.

        """
        try:
            raw_msg = unique_sdk.Message.retrieve(
                user_id=self._user_id,
                company_id=self._company_id,
                id=self._user_message_id,
                chatId=self._chat_id,
            )
            return getattr(raw_msg, "debugInfo", {}) or {}
        except Exception:
            logging.getLogger(__name__).warning(
                "Failed to retrieve debug info from user message", exc_info=True
            )
            return {}

    async def get_debug_info_async(self) -> dict[str, Any]:
        """Retrieves the debug information from the current user message asynchronously.

        Returns:
            dict[str, Any]: The debug information stored on the user message,
                or an empty dict if retrieval fails or no debug info exists.

        """
        try:
            raw_msg = await unique_sdk.Message.retrieve_async(
                user_id=self._user_id,
                company_id=self._company_id,
                id=self._user_message_id,
                chatId=self._chat_id,
            )
            return getattr(raw_msg, "debugInfo", {}) or {}
        except Exception:
            logging.getLogger(__name__).warning(
                "Failed to retrieve debug info from user message", exc_info=True
            )
            return {}

    async def update_debug_info_async(self, debug_info: dict):
        """Updates the debug information for the chat session.

        Args:
            debug_info (dict): The new debug information.

        """
        return await modify_message_async(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=False,
            debug_info=debug_info,
        )

    def replace_debug_info(self, debug_info: dict):
        """Replace the debug information in the last user message

        Args:
            debug_info (dict): The new debug information.

        """
        return modify_message(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=False,
            debug_info=debug_info,
        )

    # Message Methods
    ############################################################################

    def modify_user_message(
        self,
        content: str,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        message_id: str | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Modifies a user message in the chat session synchronously.

        Args:
            content (str): The new content for the message.
            references (list[ContentReference]): list of ContentReference objects.
            debug_info (dict[str, Any]]]): Debug information.
            message_id (str, optional): The message ID, if not specified the last user message is edited.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The modified message.

        Raises:
            Exception: If the modification fails.

        """
        return modify_message(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=False,
            content=content,
            references=references,
            debug_info=debug_info,
            message_id=message_id,
            set_completed_at=set_completed_at or False,
        )

    async def modify_user_message_async(
        self,
        content: str,
        references: list[ContentReference] = [],
        debug_info: dict = {},
        message_id: str | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Modifies a message in the chat session asynchronously.

        Args:
            content (str): The new content for the message.
            message_id (str, optional): The message ID. Defaults to None, then the ChatState user message id is used.
            references (list[ContentReference]): list of ContentReference objects. Defaults to None.
            debug_info (dict[str, Any]]]): Debug information. Defaults to {}.
            set_completed_at (bool, optional): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The modified message.

        Raises:
            Exception: If the modification fails.

        """
        return await modify_message_async(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=False,
            content=content,
            references=references,
            debug_info=debug_info,
            message_id=message_id,
            set_completed_at=set_completed_at or False,
        )

    def modify_assistant_message(
        self,
        content: str | None = None,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        message_id: str | None = None,
        set_completed_at: bool | None = None,
    ) -> ChatMessage:
        """Modifies a message in the chat session synchronously if parameter is not specified the corresponding field will remain as is.

        Args:
            content (str, optional): The new content for the message.
            original_content (str, optional): The original content for the message.
            references (list[ContentReference]): list of ContentReference objects. Defaults to [].
            debug_info (dict[str, Any]]]): Debug information. Defaults to {}.
            message_id (Optional[str]): The message ID. Defaults to None.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The modified message.

        Raises:
            Exception: If the modification fails.

        """

        return modify_message(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=True,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            message_id=message_id,
            set_completed_at=set_completed_at or False,
        )

    async def modify_assistant_message_async(
        self,
        content: str | None = None,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        message_id: str | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Modifies a message in the chat session asynchronously.

        Args:
            content (str, optional): The new content for the message.
            original_content (str, optional): The original content for the message.
            message_id (str, optional): The message ID. Defaults to None, then the ChatState assistant message id is used.
            references (list[ContentReference]): list of ContentReference objects. Defaults to None.
            debug_info (dict[str, Any]], optional): Debug information. Defaults to None.
            set_completed_at (bool, optional): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The modified message.

        Raises:
            Exception: If the modification fails.

        """
        return await modify_message_async(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=self._assistant_message_id,
            chat_id=self._chat_id,
            user_message_id=self._user_message_id,
            user_message_text=self._user_message_text,
            assistant=True,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            message_id=message_id,
            set_completed_at=set_completed_at or False,
        )

    def create_assistant_message(
        self,
        content: str,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Creates a message in the chat session synchronously.

        Args:
            content (str): The content for the message.
            original_content (str, optional): The original content for the message.
            references (list[ContentReference]): list of ContentReference objects. Defaults to None.
            debug_info (dict[str, Any]]): Debug information. Defaults to None.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The created message.

        Raises:
            Exception: If the creation fails.

        """
        chat_message = create_message(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            role=ChatMessageRole.ASSISTANT,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            set_completed_at=set_completed_at,
        )
        # Update the assistant message id
        self._assistant_message_id = chat_message.id or "unknown"
        return chat_message

    async def create_assistant_message_async(
        self,
        content: str,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Creates a message in the chat session asynchronously.

        Args:
            content (str): The content for the message.
            original_content (str, optional): The original content for the message.
            references (list[ContentReference]): list of references. Defaults to None.
            debug_info (dict[str, Any]]): Debug information. Defaults to None.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The created message.

        Raises:
            Exception: If the creation fails.

        """
        chat_message = await create_message_async(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            role=ChatMessageRole.ASSISTANT,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            set_completed_at=set_completed_at,
        )
        # Update the assistant message id
        self._assistant_message_id = chat_message.id or "unknown"
        return chat_message

    def create_user_message(
        self,
        content: str,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Creates a user message in the chat session synchronously.

        Args:
            content (str): The content for the message.
            original_content (str, optional): The original content for the message.
            references (list[ContentReference]): list of ContentReference objects. Defaults to None.
            debug_info (dict[str, Any]]): Debug information. Defaults to None.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The created message.

        Raises:
            Exception: If the creation fails.

        """
        chat_message = create_message(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            role=ChatMessageRole.USER,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            set_completed_at=set_completed_at,
        )
        # Update the user message id
        self._user_message_id = chat_message.id or "unknown"
        return chat_message

    async def create_user_message_async(
        self,
        content: str,
        original_content: str | None = None,
        references: list[ContentReference] | None = None,
        debug_info: dict | None = None,
        set_completed_at: bool | None = False,
    ) -> ChatMessage:
        """Creates a user message in the chat session asynchronously.

        Args:
            content (str): The content for the message.
            original_content (str, optional): The original content for the message.
            references (list[ContentReference]): list of references. Defaults to None.
            debug_info (dict[str, Any]]): Debug information. Defaults to None.
            set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

        Returns:
            ChatMessage: The created message.

        Raises:
            Exception: If the creation fails.

        """
        chat_message = await create_message_async(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            role=ChatMessageRole.USER,
            content=content,
            original_content=original_content,
            references=references,
            debug_info=debug_info,
            set_completed_at=set_completed_at,
        )
        # Update the user message id
        self._user_message_id = chat_message.id or "unknown"
        return chat_message

    def free_user_input(self) -> None:
        """Unblocks the next user input"""
        self.modify_assistant_message(set_completed_at=True)

    # History Methods
    ############################################################################

    def get_full_history(self) -> list[ChatMessage]:
        """Loads the full chat history for the chat session synchronously.

        Returns:
            list[ChatMessage]: The full chat history.

        Raises:
            Exception: If the loading fails.

        """
        return get_full_history(
            event_user_id=self._user_id,
            event_company_id=self._company_id,
            event_payload_chat_id=self._chat_id,
        )

    async def get_full_history_async(self) -> list[ChatMessage]:
        """Loads the full chat history for the chat session asynchronously.

        Returns:
            list[ChatMessage]: The full chat history.

        Raises:
            Exception: If the loading fails.

        """
        return await get_full_history_async(
            event_user_id=self._user_id,
            event_company_id=self._company_id,
            event_payload_chat_id=self._chat_id,
        )

    def get_full_and_selected_history(
        self,
        token_limit: int,
        percent_of_max_tokens: float = DEFAULT_PERCENT_OF_MAX_TOKENS,
        max_messages: int = DEFAULT_MAX_MESSAGES,
    ) -> tuple[list[ChatMessage], list[ChatMessage]]:
        """Loads the chat history for the chat session synchronously.

        Args:
            token_limit (int): The maximum number of tokens to load.
            percent_of_max_tokens (float): The percentage of the maximum tokens to load. Defaults to 0.15.
            max_messages (int): The maximum number of messages to load. Defaults to 4.

        Returns:
            tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

        Raises:
            Exception: If the loading fails.

        """
        full_history = get_full_history(
            event_user_id=self._user_id,
            event_company_id=self._company_id,
            event_payload_chat_id=self._chat_id,
        )
        selected_history = get_selection_from_history(
            full_history=full_history,
            max_tokens=int(round(token_limit * percent_of_max_tokens)),
            max_messages=max_messages,
            model_info=None,  # TODO: Pass language_model when available
        )

        return full_history, selected_history

    async def get_full_and_selected_history_async(
        self,
        token_limit: int,
        percent_of_max_tokens: float = DEFAULT_PERCENT_OF_MAX_TOKENS,
        max_messages: int = DEFAULT_MAX_MESSAGES,
    ) -> tuple[list[ChatMessage], list[ChatMessage]]:
        """Loads the chat history for the chat session asynchronously.

        Args:
            token_limit (int): The maximum number of tokens to load.
            percent_of_max_tokens (float): The percentage of the maximum tokens to load. Defaults to 0.15.
            max_messages (int): The maximum number of messages to load. Defaults to 4.

        Returns:
            tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

        Raises:
            Exception: If the loading fails.

        """
        full_history = await get_full_history_async(
            event_user_id=self._user_id,
            event_company_id=self._company_id,
            event_payload_chat_id=self._chat_id,
        )
        selected_history = get_selection_from_history(
            full_history=full_history,
            max_tokens=int(round(token_limit * percent_of_max_tokens)),
            max_messages=max_messages,
            model_info=None,  # TODO: Pass language_model when available
        )

        return full_history, selected_history

    # Message Assessment Methods
    ############################################################################

    def create_message_assessment(
        self,
        assistant_message_id: str,
        status: ChatMessageAssessmentStatus,
        type: ChatMessageAssessmentType,
        title: str | None = None,
        explanation: str | None = None,
        label: ChatMessageAssessmentLabel | None = None,
        is_visible: bool = True,
    ) -> ChatMessageAssessment:
        """Creates a message assessment for an assistant message synchronously.

        Args:
            assistant_message_id (str): The ID of the assistant message to assess
            status (MessageAssessmentStatus): The status of the assessment (e.g. "DONE")
            type (MessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")
            title (str | None): The title of the assessment
            explanation (str | None): Explanation of the assessment
            label (MessageAssessmentLabel | None): The assessment label (e.g. "RED")
            is_visible (bool): Whether the assessment is visible to users. Defaults to True.

        Returns:
            ChatMessageAssessment: The created message assessment

        Raises:
            Exception: If the creation fails

        """
        return create_message_assessment(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=assistant_message_id,
            status=status,
            type=type,
            title=title,
            explanation=explanation,
            label=label,
            is_visible=is_visible,
        )

    async def create_message_assessment_async(
        self,
        assistant_message_id: str,
        status: ChatMessageAssessmentStatus,
        type: ChatMessageAssessmentType,
        title: str | None = None,
        explanation: str | None = None,
        label: ChatMessageAssessmentLabel | None = None,
        is_visible: bool = True,
    ) -> ChatMessageAssessment:
        """Creates a message assessment for an assistant message asynchronously.

        Args:
            assistant_message_id (str): The ID of the assistant message to assess
            status (ChatMessageAssessmentStatus): The status of the assessment (e.g. "DONE")
            type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")
            title (str | None): The title of the assessment
            explanation (str | None): Explanation of the assessment
            label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
            is_visible (bool): Whether the assessment is visible to users. Defaults to True.

        Returns:
            ChatMessageAssessment: The created message assessment

        Raises:
            Exception: If the creation fails

        """
        return await create_message_assessment_async(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=assistant_message_id,
            status=status,
            type=type,
            title=title,
            explanation=explanation,
            label=label,
            is_visible=is_visible,
        )

    def modify_message_assessment(
        self,
        assistant_message_id: str,
        status: ChatMessageAssessmentStatus,
        type: ChatMessageAssessmentType,
        title: str | None = None,
        explanation: str | None = None,
        label: ChatMessageAssessmentLabel | None = None,
    ) -> ChatMessageAssessment:
        """Modifies a message assessment for an assistant message synchronously.

        Args:
            assistant_message_id (str): The ID of the assistant message to assess
            status (MessageAssessmentStatus): The status of the assessment (e.g. "DONE")
            title (str | None): The title of the assessment
            explanation (str | None): Explanation of the assessment
            label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
            type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")

        Returns:
            dict: The modified message assessment

        Raises:
            Exception: If the modification fails

        """
        return modify_message_assessment(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=assistant_message_id,
            status=status,
            type=type,
            title=title,
            explanation=explanation,
            label=label,
        )

    async def modify_message_assessment_async(
        self,
        assistant_message_id: str,
        type: ChatMessageAssessmentType,
        title: str | None = None,
        status: ChatMessageAssessmentStatus | None = None,
        explanation: str | None = None,
        label: ChatMessageAssessmentLabel | None = None,
    ) -> ChatMessageAssessment:
        """Modifies a message assessment for an assistant message asynchronously.

        Args:
            assistant_message_id (str): The ID of the assistant message to assess
            status (ChatMessageAssessmentStatus): The status of the assessment (e.g. "DONE")
            title (str | None): The title of the assessment
            explanation (str | None): Explanation of the assessment
            label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
            type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")

        Returns:
            ChatMessageAssessment: The modified message assessment

        Raises:
            Exception: If the modification fails

        """
        return await modify_message_assessment_async(
            user_id=self._user_id,
            company_id=self._company_id,
            assistant_message_id=assistant_message_id,
            status=status,
            type=type,
            title=title,
            explanation=explanation,
            label=label,
        )

    # Message Log Methods
    ############################################################################

    def create_message_log(
        self,
        *,
        message_id: str,
        text: str,
        status: MessageLogStatus,
        order: int,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Creates a message log for tracking execution steps synchronously.

        Args:
            message_id (str): The ID of the message this log belongs to
            text (str): The log text content
            status (MessageLogStatus): The status of this log entry
            order (int): The order/sequence number of this log entry
            details (MessageLogDetails | None): Additional details about this log entry
            uncited_references (MessageLogUncitedReferences | None): References that are not cited
            references (list[ContentReference] | None): List of references for this log

        Returns:
            MessageLog: The created message log

        Raises:
            Exception: If the creation fails

        """
        return create_message_log(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            text=text,
            status=status,
            order=order,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    async def create_message_log_async(
        self,
        *,
        message_id: str,
        text: str,
        status: MessageLogStatus,
        order: int,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Creates a message log for tracking execution steps asynchronously.

        Args:
            message_id (str): The ID of the message this log belongs to
            text (str): The log text content
            status (MessageLogStatus): The status of this log entry
            order (int): The order/sequence number of this log entry
            details (MessageLogDetails | None): Additional details about this log entry
            uncited_references (MessageLogUncitedReferences | None): References that are not cited
            references (list[ContentReference] | None): List of references for this log

        Returns:
            MessageLog: The created message log

        Raises:
            Exception: If the creation fails

        """
        return await create_message_log_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            text=text,
            status=status,
            order=order,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    def update_message_log(
        self,
        *,
        message_log_id: str,
        order: int,
        text: str | None = None,
        status: MessageLogStatus | None = None,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Updates a message log synchronously.

        Args:
            message_log_id (str): The ID of the message log to update
            order (int): The order/sequence number (required)
            text (str | None): The updated log text content
            status (MessageLogStatus | None): The updated status
            details (MessageLogDetails | None): Updated additional details
            uncited_references (MessageLogUncitedReferences | None): Updated uncited references
            references (list[ContentReference] | None): Updated list of references

        Returns:
            MessageLog: The updated message log

        Raises:
            Exception: If the update fails

        """
        return update_message_log(
            user_id=self._user_id,
            company_id=self._company_id,
            message_log_id=message_log_id,
            order=order,
            text=text,
            status=status,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    async def update_message_log_async(
        self,
        *,
        message_log_id: str,
        order: int,
        text: str | None = None,
        status: MessageLogStatus | None = None,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Updates a message log asynchronously.

        Args:
            message_log_id (str): The ID of the message log to update
            order (int): The order/sequence number (required)
            text (str | None): The updated log text content
            status (MessageLogStatus | None): The updated status
            details (MessageLogDetails | None): Updated additional details
            uncited_references (MessageLogUncitedReferences | None): Updated uncited references
            references (list[ContentReference] | None): Updated list of references

        Returns:
            MessageLog: The updated message log

        Raises:
            Exception: If the update fails

        """
        return await update_message_log_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_log_id=message_log_id,
            order=order,
            text=text,
            status=status,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    def create_assistant_message_log(
        self,
        *,
        text: str,
        status: MessageLogStatus,
        order: int,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Creates a message log for the current assistant message synchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            text (str): The log text content
            status (MessageLogStatus): The status of this log entry
            order (int): The order/sequence number of this log entry
            details (MessageLogDetails | None): Additional details about this log entry
            uncited_references (MessageLogUncitedReferences | None): References that are not cited
            references (list[ContentReference] | None): List of references for this log

        Returns:
            MessageLog: The created message log

        Raises:
            Exception: If the creation fails

        """
        return self.create_message_log(
            message_id=self._assistant_message_id,
            text=text,
            status=status,
            order=order,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    async def create_assistant_message_log_async(
        self,
        *,
        text: str,
        status: MessageLogStatus,
        order: int,
        details: MessageLogDetails | None = None,
        uncited_references: MessageLogUncitedReferences | None = None,
        references: list[ContentReference] | None = None,
    ) -> MessageLog:
        """Creates a message log for the current assistant message asynchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            text (str): The log text content
            status (MessageLogStatus): The status of this log entry
            order (int): The order/sequence number of this log entry
            details (MessageLogDetails | None): Additional details about this log entry
            uncited_references (MessageLogUncitedReferences | None): References that are not cited
            references (list[ContentReference] | None): List of references for this log

        Returns:
            MessageLog: The created message log

        Raises:
            Exception: If the creation fails

        """
        return await self.create_message_log_async(
            message_id=self._assistant_message_id,
            text=text,
            status=status,
            order=order,
            details=details,
            uncited_references=uncited_references,
            references=references,
        )

    # Message Execution Methods
    ############################################################################

    def create_message_execution(
        self,
        *,
        message_id: str,
        type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        is_queueable: bool = True,
        execution_options: dict | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Creates a message execution for tracking long-running operations synchronously.

        Args:
            message_id (str): The ID of the message this execution belongs to
            type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
            seconds_remaining (int | None): Estimated seconds remaining for completion
            percentage_completed (int | None): Percentage of completion (0-100)
            is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
            execution_options (dict | None): Additional execution options. Defaults to None.
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The created message execution

        Raises:
            Exception: If the creation fails

        """
        return create_message_execution(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            chat_id=self._chat_id,
            type=type,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            is_queueable=is_queueable,
            execution_options=execution_options,
            progress_title=progress_title,
        )

    async def create_message_execution_async(
        self,
        *,
        message_id: str,
        type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        is_queueable: bool = True,
        execution_options: dict | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Creates a message execution for tracking long-running operations asynchronously.

        Args:
            message_id (str): The ID of the message this execution belongs to
            type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
            seconds_remaining (int | None): Estimated seconds remaining for completion
            percentage_completed (int | None): Percentage of completion (0-100)
            is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
            execution_options (dict | None): Additional execution options. Defaults to None.
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The created message execution

        Raises:
            Exception: If the creation fails

        """
        return await create_message_execution_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            chat_id=self._chat_id,
            type=type,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            is_queueable=is_queueable,
            execution_options=execution_options,
            progress_title=progress_title,
        )

    def get_message_execution(
        self,
        *,
        message_id: str,
    ) -> MessageExecution:
        """Gets a message execution by message ID synchronously.

        Args:
            message_id (str): The ID of the message to get execution for

        Returns:
            MessageExecution: The message execution

        Raises:
            Exception: If the retrieval fails

        """
        return get_message_execution(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
        )

    async def get_message_execution_async(
        self,
        *,
        message_id: str,
    ) -> MessageExecution:
        """Gets a message execution by message ID asynchronously.

        Args:
            message_id (str): The ID of the message to get execution for

        Returns:
            MessageExecution: The message execution

        Raises:
            Exception: If the retrieval fails

        """
        return await get_message_execution_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
        )

    def update_message_execution(
        self,
        *,
        message_id: str,
        status: MessageExecutionUpdateStatus | None = None,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Updates a message execution synchronously.

        Args:
            message_id (str): The ID of the message to update execution for
            status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
            seconds_remaining (int | None): Updated estimated seconds remaining
            percentage_completed (int | None): Updated percentage of completion (0-100)
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The updated message execution

        Raises:
            Exception: If the update fails

        """
        return update_message_execution(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            status=status,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            progress_title=progress_title,
        )

    async def update_message_execution_async(
        self,
        *,
        message_id: str,
        status: MessageExecutionUpdateStatus | None = None,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Updates a message execution asynchronously.

        Args:
            message_id (str): The ID of the message to update execution for
            status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
            seconds_remaining (int | None): Updated estimated seconds remaining
            percentage_completed (int | None): Updated percentage of completion (0-100)
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The updated message execution

        Raises:
            Exception: If the update fails

        """
        return await update_message_execution_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            status=status,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            progress_title=progress_title,
        )

    def create_assistant_message_execution(
        self,
        *,
        type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        is_queueable: bool = True,
        execution_options: dict | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Creates a message execution for the current assistant message synchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
            seconds_remaining (int | None): Estimated seconds remaining for completion
            percentage_completed (int | None): Percentage of completion (0-100)
            is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
            execution_options (dict | None): Additional execution options. Defaults to None.
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The created message execution

        Raises:
            Exception: If the creation fails

        """
        return self.create_message_execution(
            message_id=self._assistant_message_id,
            type=type,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            is_queueable=is_queueable,
            execution_options=execution_options,
            progress_title=progress_title,
        )

    async def create_assistant_message_execution_async(
        self,
        *,
        type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
        is_queueable: bool = True,
        execution_options: dict | None = None,
        progress_title: str | None = None,
    ) -> MessageExecution:
        """Creates a message execution for the current assistant message asynchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
            seconds_remaining (int | None): Estimated seconds remaining for completion
            percentage_completed (int | None): Percentage of completion (0-100)
            is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
            execution_options (dict | None): Additional execution options. Defaults to None.
            progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

        Returns:
            MessageExecution: The created message execution

        Raises:
            Exception: If the creation fails

        """
        return await self.create_message_execution_async(
            message_id=self._assistant_message_id,
            type=type,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
            is_queueable=is_queueable,
            execution_options=execution_options,
            progress_title=progress_title,
        )

    def get_assistant_message_execution(self) -> MessageExecution:
        """Gets the message execution for the current assistant message synchronously.

        This is a convenience method that uses the current assistant message ID.

        Returns:
            MessageExecution: The message execution

        Raises:
            Exception: If the retrieval fails

        """
        return self.get_message_execution(message_id=self._assistant_message_id)

    async def get_assistant_message_execution_async(self) -> MessageExecution:
        """Gets the message execution for the current assistant message asynchronously.

        This is a convenience method that uses the current assistant message ID.

        Returns:
            MessageExecution: The message execution

        Raises:
            Exception: If the retrieval fails

        """
        return await self.get_message_execution_async(
            message_id=self._assistant_message_id
        )

    def update_assistant_message_execution(
        self,
        *,
        status: MessageExecutionUpdateStatus | None = None,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
    ) -> MessageExecution:
        """Updates the message execution for the current assistant message synchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
            seconds_remaining (int | None): Updated estimated seconds remaining
            percentage_completed (int | None): Updated percentage of completion (0-100)

        Returns:
            MessageExecution: The updated message execution

        Raises:
            Exception: If the update fails

        """
        return self.update_message_execution(
            message_id=self._assistant_message_id,
            status=status,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
        )

    async def update_assistant_message_execution_async(
        self,
        *,
        status: MessageExecutionUpdateStatus | None = None,
        seconds_remaining: int | None = None,
        percentage_completed: int | None = None,
    ) -> MessageExecution:
        """Updates the message execution for the current assistant message asynchronously.

        This is a convenience method that uses the current assistant message ID.

        Args:
            status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
            seconds_remaining (int | None): Updated estimated seconds remaining
            percentage_completed (int | None): Updated percentage of completion (0-100)

        Returns:
            MessageExecution: The updated message execution

        Raises:
            Exception: If the update fails

        """
        return await self.update_message_execution_async(
            message_id=self._assistant_message_id,
            status=status,
            seconds_remaining=seconds_remaining,
            percentage_completed=percentage_completed,
        )

    # Language Model Methods
    ############################################################################

    @deprecated("Use complete_with_references instead")
    def stream_complete(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None = None,
        debug_info: dict = {},
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        start_text: str | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelStreamResponse:
        return self.complete_with_references(
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

    def complete_with_references(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None = None,
        debug_info: dict | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        start_text: str | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelStreamResponse:
        """Streams a completion in the chat session synchronously."""
        return stream_complete_with_references(
            company_id=self._company_id,
            user_id=self._user_id,
            assistant_message_id=self._assistant_message_id,
            user_message_id=self._user_message_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

    def complete(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None = None,
        debug_info: dict | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        start_text: str | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelResponse:
        response = self.complete_with_references(
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

        return LanguageModelResponse.from_stream_response(response)

    @deprecated("use complete_with_references_async instead.")
    async def stream_complete_async(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None = None,
        debug_info: dict | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        start_text: str | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelStreamResponse:
        """Stream a completion in the chat session asynchronously."""
        return await self.complete_with_references_async(
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

    async def complete_with_references_async(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None = None,
        debug_info: dict | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        start_text: str | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelStreamResponse:
        return await stream_complete_with_references_async(
            company_id=self._company_id,
            user_id=self._user_id,
            assistant_message_id=self._assistant_message_id,
            user_message_id=self._user_message_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

    async def complete_async(
        self,
        messages: LanguageModelMessages | list[ChatCompletionMessageParam],
        model_name: LanguageModelName | str,
        content_chunks: list[ContentChunk] | None,
        debug_info: dict | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        timeout: int = DEFAULT_COMPLETE_TIMEOUT,
        tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
        start_text: str | None = None,
        tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
        other_options: dict | None = None,
    ) -> LanguageModelResponse:
        response = self.complete_with_references_async(
            messages=messages,
            model_name=model_name,
            content_chunks=content_chunks,
            debug_info=debug_info,
            temperature=temperature,
            timeout=timeout,
            tools=tools,
            start_text=start_text,
            tool_choice=tool_choice,
            other_options=other_options,
        )

        return LanguageModelResponse.from_stream_response(await response)

    def complete_responses_with_references(
        self,
        *,
        model_name: LanguageModelName | str,
        messages: str
        | LanguageModelMessages
        | Sequence[
            ResponseInputItemParam
            | LanguageModelMessageOptions
            | ResponseOutputItem  # History is automatically convertible
        ],
        content_chunks: list[ContentChunk] | None = None,
        tools: Sequence[LanguageModelToolDescription | ToolParam] | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        debug_info: dict | None = None,
        start_text: str | None = None,
        include: list[ResponseIncludable] | None = None,
        instructions: str | None = None,
        max_output_tokens: int | None = None,
        metadata: Metadata | None = None,
        parallel_tool_calls: bool | None = None,
        text: ResponseTextConfigParam | None = None,
        tool_choice: response_create_params.ToolChoice | None = None,
        top_p: float | None = None,
        reasoning: Reasoning | None = None,
        other_options: dict | None = None,
    ) -> ResponsesLanguageModelStreamResponse:
        return stream_responses_with_references(
            company_id=self._company_id,
            user_id=self._user_id,
            assistant_message_id=self._assistant_message_id,
            user_message_id=self._user_message_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            model_name=model_name,
            messages=messages,
            content_chunks=content_chunks,
            tools=tools,
            temperature=temperature,
            debug_info=debug_info,
            start_text=start_text,
            include=include,
            instructions=instructions,
            max_output_tokens=max_output_tokens,
            metadata=metadata,
            parallel_tool_calls=parallel_tool_calls,
            text=text,
            tool_choice=tool_choice,
            top_p=top_p,
            reasoning=reasoning,
            other_options=other_options,
        )

    async def complete_responses_with_references_async(
        self,
        *,
        model_name: LanguageModelName | str,
        messages: str
        | LanguageModelMessages
        | Sequence[
            ResponseInputItemParam | LanguageModelMessageOptions | ResponseOutputItem
        ],
        content_chunks: list[ContentChunk] | None = None,
        tools: Sequence[LanguageModelToolDescription | ToolParam] | None = None,
        temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
        debug_info: dict | None = None,
        start_text: str | None = None,
        include: list[ResponseIncludable] | None = None,
        instructions: str | None = None,
        max_output_tokens: int | None = None,
        metadata: Metadata | None = None,
        parallel_tool_calls: bool | None = None,
        text: ResponseTextConfigParam | None = None,
        tool_choice: response_create_params.ToolChoice | None = None,
        top_p: float | None = None,
        reasoning: Reasoning | None = None,
        other_options: dict | None = None,
    ) -> ResponsesLanguageModelStreamResponse:
        async def _on_rate_limit_retry(attempt: int, wait_secs: float) -> None:
            if not rate_limit_retry_config.log_message_on_retry:
                return
            if not feature_flags.enable_new_answers_ui_un_14411.is_enabled(
                self._company_id
            ):
                return
            from unique_toolkit.agentic.message_log_manager.service import (
                _request_counters,
            )

            msg_id = self._assistant_message_id
            _request_counters[msg_id] += 1
            order = _request_counters[msg_id]
            try:
                await self.create_message_log_async(
                    message_id=msg_id,
                    text=f"Rate limit reached; retrying in {wait_secs:.0f}s (attempt {attempt})",
                    status=MessageLogStatus.RUNNING,
                    order=order,
                )
            except Exception:
                _LOGGER.warning(
                    "Failed to write rate-limit retry message log",
                    exc_info=True,
                )

        return await stream_responses_with_references_async(
            company_id=self._company_id,
            user_id=self._user_id,
            assistant_message_id=self._assistant_message_id,
            user_message_id=self._user_message_id,
            chat_id=self._chat_id,
            assistant_id=self._assistant_id,
            model_name=model_name,
            messages=messages,
            content_chunks=content_chunks,
            tools=tools,
            temperature=temperature,
            debug_info=debug_info,
            start_text=start_text,
            include=include,
            instructions=instructions,
            max_output_tokens=max_output_tokens,
            metadata=metadata,
            parallel_tool_calls=parallel_tool_calls,
            text=text,
            tool_choice=tool_choice,
            top_p=top_p,
            reasoning=reasoning,
            other_options=other_options,
            on_rate_limit_retry=_on_rate_limit_retry,
        )

    # Chat Content Methods
    ############################################################################

    def upload_to_chat_from_bytes(
        self,
        *,
        content: bytes,
        content_name: str,
        mime_type: str,
        scope_id: str | None = None,
        skip_ingestion: bool = False,
        hide_in_chat: bool = False,
        ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> Content:
        return upload_content_from_bytes(
            user_id=self._user_id,
            company_id=self._company_id,
            content=content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=self._chat_id,
            skip_ingestion=skip_ingestion,
            hide_in_chat=hide_in_chat,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )

    async def upload_to_chat_from_bytes_async(
        self,
        *,
        content: bytes,
        content_name: str,
        mime_type: str,
        scope_id: str | None = None,
        skip_ingestion: bool = False,
        hide_in_chat: bool = False,
        ingestion_config: unique_sdk.Content.IngestionConfig | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> Content:
        return await upload_content_from_bytes_async(
            user_id=self._user_id,
            company_id=self._company_id,
            content=content,
            content_name=content_name,
            mime_type=mime_type,
            scope_id=scope_id,
            chat_id=self._chat_id,
            skip_ingestion=skip_ingestion,
            hide_in_chat=hide_in_chat,
            ingestion_config=ingestion_config,
            metadata=metadata,
        )

    def download_chat_content_to_bytes(self, *, content_id: str) -> bytes:
        """Download content by id from the content-scope chat (e.g. parent chat when subagent).

        Uses the service's content-scope chat id, so when running as a subagent
        with correlation, this accesses files from the primary chat session.

        Args:
            content_id: The content id to download.

        Returns:
            bytes: The raw content bytes.
        """
        return download_content_to_bytes(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            chat_id=self._content_scope_chat_id,
        )

    async def download_chat_content_to_bytes_async(self, *, content_id: str) -> bytes:
        """Async download content by id from the content-scope chat (e.g. parent chat when subagent).

        Uses the service's content-scope chat id, so when running as a subagent
        with correlation, this accesses files from the primary chat session.

        Args:
            content_id: The content id to download.

        Returns:
            bytes: The raw content bytes.
        """
        return await download_content_to_bytes_async(
            user_id=self._user_id,
            company_id=self._company_id,
            content_id=content_id,
            chat_id=self._content_scope_chat_id,
        )

    def download_chat_images_and_documents(self) -> tuple[list[Content], list[Content]]:
        """Return images and documents from the content-scope chat (e.g. parent chat when subagent).

        Uses the service's content-scope chat id, so when running as a subagent
        with correlation, this accesses files from the primary chat session.

        Returns:
            tuple[list[Content], list[Content]]: (images, documents) from the content-scope chat.
        """
        images: list[Content] = []
        files: list[Content] = []
        for c in search_contents(
            user_id=self._user_id,
            company_id=self._company_id,
            chat_id=self._content_scope_chat_id,
            where={"ownerId": {"equals": self._content_scope_chat_id}},
        ):
            if is_file_content(filename=c.key):
                files.append(c)
            if is_image_content(filename=c.key):
                images.append(c)
        return images, files

    # Short Term Memories
    ############################################################################

    def create_chat_memory_by_id(
        self, *, chat_id: str, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for a specific chat synchronously.

        Args:
            chat_id (str): The chat ID
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        # Convert BaseModel to JSON string if needed
        if isinstance(value, BaseModel):
            value = value.model_dump_json()

        return create_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            chat_id=chat_id,
        )

    async def create_chat_memory_by_id_async(
        self, *, chat_id: str, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for a specific chat asynchronously.

        Args:
            chat_id (str): The chat ID
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        # Convert BaseModel to JSON string if needed
        if isinstance(value, BaseModel):
            value = value.model_dump_json()

        return await create_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            chat_id=chat_id,
        )

    def create_message_memory_by_id(
        self, *, message_id: str, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for a specific message synchronously.

        Args:
            message_id (str): The message ID
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        # Convert BaseModel to JSON string if needed
        if isinstance(value, BaseModel):
            value = value.model_dump_json()

        return create_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            message_id=message_id,
        )

    async def create_message_memory_by_id_async(
        self, *, message_id: str, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for a specific message asynchronously.

        Args:
            message_id (str): The message ID
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        # Convert BaseModel to JSON string if needed
        if isinstance(value, BaseModel):
            value = value.model_dump_json()

        return await create_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            message_id=message_id,
        )

    def find_chat_memory_by_id(self, *, chat_id: str, key: str) -> ShortTermMemory:
        """Finds the latest short-term memory for a specific chat synchronously.

        Args:
            chat_id (str): The chat ID
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return find_latest_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            chat_id=chat_id,
        )

    async def find_chat_memory_by_id_async(
        self, *, chat_id: str, key: str
    ) -> ShortTermMemory:
        """Finds the latest short-term memory for a specific chat asynchronously.

        Args:
            chat_id (str): The chat ID
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return await find_latest_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            chat_id=chat_id,
        )

    def find_message_memory_by_id(
        self, *, message_id: str, key: str
    ) -> ShortTermMemory:
        """Finds the latest short-term memory for a specific message synchronously.

        Args:
            message_id (str): The message ID
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return find_latest_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            message_id=message_id,
        )

    async def find_message_memory_by_id_async(
        self, *, message_id: str, key: str
    ) -> ShortTermMemory:
        """Finds the latest short-term memory for a specific message asynchronously.

        Args:
            message_id (str): The message ID
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return await find_latest_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            message_id=message_id,
        )

    # Convenience methods using current chat/message IDs
    ############################################################################

    def create_chat_memory(
        self, *, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for the current chat synchronously.

        Args:
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        return self.create_chat_memory_by_id(
            chat_id=self._chat_id,
            key=key,
            value=value,
        )

    async def create_chat_memory_async(
        self, *, key: str, value: str | dict | BaseModel
    ) -> ShortTermMemory:
        """Creates a short-term memory for the current chat asynchronously.

        Args:
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        return await self.create_chat_memory_by_id_async(
            chat_id=self._chat_id,
            key=key,
            value=value,
        )

    @overload
    def create_message_memory(
        self,
        *,
        key: str,
        value: str | dict | BaseModel,
    ) -> ShortTermMemory: ...

    @overload
    def create_message_memory(
        self, *, key: str, value: str | dict | BaseModel, message_id: str
    ) -> ShortTermMemory: ...

    def create_message_memory(
        self, *, key: str, value: str | dict | BaseModel, message_id: str | None = None
    ) -> ShortTermMemory:
        """Creates a short-term memory for the current assistant message synchronously.

        Args:
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        return self.create_message_memory_by_id(
            key=key,
            value=value,
            message_id=message_id or self._assistant_message_id,
        )

    @overload
    async def create_message_memory_async(
        self,
        *,
        key: str,
        value: str | dict | BaseModel,
    ) -> ShortTermMemory: ...

    @overload
    async def create_message_memory_async(
        self, *, key: str, value: str | dict | BaseModel, message_id: str
    ) -> ShortTermMemory: ...

    async def create_message_memory_async(
        self, *, key: str, value: str | dict | BaseModel, message_id: str | None = None
    ) -> ShortTermMemory:
        """Creates a short-term memory for the current assistant message asynchronously.

        Args:
            key (str): The memory key
            value (str | dict | BaseModel): The memory value

        Returns:
            ShortTermMemory: The created short-term memory

        Raises:
            Exception: If the creation fails
        """
        return await self.create_message_memory_by_id_async(
            message_id=message_id or self._assistant_message_id,
            key=key,
            value=value,
        )

    def find_chat_memory(self, *, key: str) -> ShortTermMemory:
        """Finds the latest short-term memory for the current chat synchronously.

        Args:
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return self.find_chat_memory_by_id(
            chat_id=self._chat_id,
            key=key,
        )

    async def find_chat_memory_async(self, *, key: str) -> ShortTermMemory:
        """Finds the latest short-term memory for the current chat asynchronously.

        Args:
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return await self.find_chat_memory_by_id_async(
            chat_id=self._chat_id,
            key=key,
        )

    @overload
    def find_message_memory(self, *, key: str) -> ShortTermMemory: ...

    @overload
    def find_message_memory(self, *, key: str, message_id: str) -> ShortTermMemory: ...

    def find_message_memory(
        self, *, key: str, message_id: str | None = None
    ) -> ShortTermMemory:
        """Finds the latest short-term memory for the current assistant message synchronously.

        Args:
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return self.find_message_memory_by_id(
            message_id=message_id or self._assistant_message_id,
            key=key,
        )

    @overload
    async def find_message_memory_async(self, *, key: str) -> ShortTermMemory: ...

    @overload
    async def find_message_memory_async(
        self, *, key: str, message_id: str
    ) -> ShortTermMemory: ...

    async def find_message_memory_async(
        self, *, key: str, message_id: str | None = None
    ) -> ShortTermMemory:
        """Finds the latest short-term memory for the current assistant message asynchronously.

        Args:
            key (str): The memory key

        Returns:
            ShortTermMemory: The latest short-term memory

        Raises:
            Exception: If the retrieval fails
        """
        return await self.find_message_memory_by_id_async(
            message_id=message_id or self._assistant_message_id,
            key=key,
        )

    # Message Tool Methods
    ############################################################################

    def create_message_tools(
        self,
        *,
        tool_calls: list[ChatMessageTool],
        message_id: str | None = None,
    ) -> list[ChatMessageTool]:
        """Persist tool call records for an assistant message.

        Args:
            tool_calls: The tool call records to persist.
            message_id: The assistant message to attach records to.
                Defaults to the current turn's assistant message.
        """
        return create_message_tools(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id or self._assistant_message_id,
            tool_calls=tool_calls,
        )

    async def create_message_tools_async(
        self,
        *,
        tool_calls: list[ChatMessageTool],
        message_id: str | None = None,
    ) -> list[ChatMessageTool]:
        """Async variant of create_message_tools.

        Args:
            tool_calls: The tool call records to persist.
            message_id: The assistant message to attach records to.
                Defaults to the current turn's assistant message.
        """
        return await create_message_tools_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id or self._assistant_message_id,
            tool_calls=tool_calls,
        )

    def get_message_tools(
        self,
        *,
        message_id: str | None = None,
        message_ids: list[str] | None = None,
    ) -> list[ChatMessageTool]:
        """Fetch persisted tool call records for one or more assistant messages."""
        return get_message_tools(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            message_ids=message_ids,
        )

    async def get_message_tools_async(
        self,
        *,
        message_id: str | None = None,
        message_ids: list[str] | None = None,
    ) -> list[ChatMessageTool]:
        """Async variant of get_message_tools."""
        return await get_message_tools_async(
            user_id=self._user_id,
            company_id=self._company_id,
            message_id=message_id,
            message_ids=message_ids,
        )

cancellation property

Cancellation watcher for this chat session.

elicitation property

Get the ElicitationService for this chat session.

__init__(event, content_scope_chat_id=None)

Initialize ChatService from an event (deprecated). Use :meth:from_context instead.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

@deprecated(
    "Use UniqueContext.from_chat_event(event) with UniqueServiceFactory instead."
)
def __init__(
    self,
    event: "ChatEvent",
    content_scope_chat_id: str | None = None,
) -> None:
    """Initialize ChatService from an event (deprecated). Use :meth:`from_context` instead."""
    super().__init__(event, content_scope_chat_id)
    self._elicitation_service: ElicitationService | None = None
    self._elicitation_message_id: str | None = None
    self._cancellation_watcher: CancellationWatcher = CancellationWatcher(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._chat_id,
        assistant_message_id=self._assistant_message_id,
    )

complete_with_references(messages, model_name, content_chunks=None, debug_info=None, temperature=DEFAULT_COMPLETE_TEMPERATURE, timeout=DEFAULT_COMPLETE_TIMEOUT, tools=None, start_text=None, tool_choice=None, other_options=None)

Streams a completion in the chat session synchronously.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def complete_with_references(
    self,
    messages: LanguageModelMessages | list[ChatCompletionMessageParam],
    model_name: LanguageModelName | str,
    content_chunks: list[ContentChunk] | None = None,
    debug_info: dict | None = None,
    temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
    timeout: int = DEFAULT_COMPLETE_TIMEOUT,
    tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
    start_text: str | None = None,
    tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
    other_options: dict | None = None,
) -> LanguageModelStreamResponse:
    """Streams a completion in the chat session synchronously."""
    return stream_complete_with_references(
        company_id=self._company_id,
        user_id=self._user_id,
        assistant_message_id=self._assistant_message_id,
        user_message_id=self._user_message_id,
        chat_id=self._chat_id,
        assistant_id=self._assistant_id,
        messages=messages,
        model_name=model_name,
        content_chunks=content_chunks,
        debug_info=debug_info,
        temperature=temperature,
        timeout=timeout,
        tools=tools,
        start_text=start_text,
        tool_choice=tool_choice,
        other_options=other_options,
    )

create_assistant_message(content, original_content=None, references=None, debug_info=None, set_completed_at=False)

Creates a message in the chat session synchronously.

Parameters:

Name	Type	Description	Default
content	str	The content for the message.	required
original_content	str	The original content for the message.	None
references	list[ContentReference]	list of ContentReference objects. Defaults to None.	None
debug_info	dict[str, Any]]	Debug information. Defaults to None.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The created message.

Raises:

Type	Description
Exception	If the creation fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_assistant_message(
    self,
    content: str,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Creates a message in the chat session synchronously.

    Args:
        content (str): The content for the message.
        original_content (str, optional): The original content for the message.
        references (list[ContentReference]): list of ContentReference objects. Defaults to None.
        debug_info (dict[str, Any]]): Debug information. Defaults to None.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The created message.

    Raises:
        Exception: If the creation fails.

    """
    chat_message = create_message(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._chat_id,
        assistant_id=self._assistant_id,
        role=ChatMessageRole.ASSISTANT,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        set_completed_at=set_completed_at,
    )
    # Update the assistant message id
    self._assistant_message_id = chat_message.id or "unknown"
    return chat_message

create_assistant_message_async(content, original_content=None, references=None, debug_info=None, set_completed_at=False) async

Creates a message in the chat session asynchronously.

Parameters:

Name	Type	Description	Default
content	str	The content for the message.	required
original_content	str	The original content for the message.	None
references	list[ContentReference]	list of references. Defaults to None.	None
debug_info	dict[str, Any]]	Debug information. Defaults to None.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The created message.

Raises:

Type	Description
Exception	If the creation fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_assistant_message_async(
    self,
    content: str,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Creates a message in the chat session asynchronously.

    Args:
        content (str): The content for the message.
        original_content (str, optional): The original content for the message.
        references (list[ContentReference]): list of references. Defaults to None.
        debug_info (dict[str, Any]]): Debug information. Defaults to None.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The created message.

    Raises:
        Exception: If the creation fails.

    """
    chat_message = await create_message_async(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._chat_id,
        assistant_id=self._assistant_id,
        role=ChatMessageRole.ASSISTANT,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        set_completed_at=set_completed_at,
    )
    # Update the assistant message id
    self._assistant_message_id = chat_message.id or "unknown"
    return chat_message

create_assistant_message_execution(*, type=MessageExecutionType.DEEP_RESEARCH, seconds_remaining=None, percentage_completed=None, is_queueable=True, execution_options=None, progress_title=None)

Creates a message execution for the current assistant message synchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
type	MessageExecutionType	The type of execution. Defaults to DEEP_RESEARCH.	DEEP_RESEARCH
seconds_remaining	int | None	Estimated seconds remaining for completion	None
percentage_completed	int | None	Percentage of completion (0-100)	None
is_queueable	bool	Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.	True
execution_options	dict | None	Additional execution options. Defaults to None.	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The created message execution

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_assistant_message_execution(
    self,
    *,
    type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    is_queueable: bool = True,
    execution_options: dict | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Creates a message execution for the current assistant message synchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
        seconds_remaining (int | None): Estimated seconds remaining for completion
        percentage_completed (int | None): Percentage of completion (0-100)
        is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
        execution_options (dict | None): Additional execution options. Defaults to None.
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The created message execution

    Raises:
        Exception: If the creation fails

    """
    return self.create_message_execution(
        message_id=self._assistant_message_id,
        type=type,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        is_queueable=is_queueable,
        execution_options=execution_options,
        progress_title=progress_title,
    )

create_assistant_message_execution_async(*, type=MessageExecutionType.DEEP_RESEARCH, seconds_remaining=None, percentage_completed=None, is_queueable=True, execution_options=None, progress_title=None) async

Creates a message execution for the current assistant message asynchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
type	MessageExecutionType	The type of execution. Defaults to DEEP_RESEARCH.	DEEP_RESEARCH
seconds_remaining	int | None	Estimated seconds remaining for completion	None
percentage_completed	int | None	Percentage of completion (0-100)	None
is_queueable	bool	Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.	True
execution_options	dict | None	Additional execution options. Defaults to None.	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The created message execution

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_assistant_message_execution_async(
    self,
    *,
    type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    is_queueable: bool = True,
    execution_options: dict | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Creates a message execution for the current assistant message asynchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
        seconds_remaining (int | None): Estimated seconds remaining for completion
        percentage_completed (int | None): Percentage of completion (0-100)
        is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
        execution_options (dict | None): Additional execution options. Defaults to None.
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The created message execution

    Raises:
        Exception: If the creation fails

    """
    return await self.create_message_execution_async(
        message_id=self._assistant_message_id,
        type=type,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        is_queueable=is_queueable,
        execution_options=execution_options,
        progress_title=progress_title,
    )

create_assistant_message_log(*, text, status, order, details=None, uncited_references=None, references=None)

Creates a message log for the current assistant message synchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
text	str	The log text content	required
status	MessageLogStatus	The status of this log entry	required
order	int	The order/sequence number of this log entry	required
details	MessageLogDetails | None	Additional details about this log entry	None
uncited_references	MessageLogUncitedReferences | None	References that are not cited	None
references	list[ContentReference] | None	List of references for this log	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The created message log

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_assistant_message_log(
    self,
    *,
    text: str,
    status: MessageLogStatus,
    order: int,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Creates a message log for the current assistant message synchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        text (str): The log text content
        status (MessageLogStatus): The status of this log entry
        order (int): The order/sequence number of this log entry
        details (MessageLogDetails | None): Additional details about this log entry
        uncited_references (MessageLogUncitedReferences | None): References that are not cited
        references (list[ContentReference] | None): List of references for this log

    Returns:
        MessageLog: The created message log

    Raises:
        Exception: If the creation fails

    """
    return self.create_message_log(
        message_id=self._assistant_message_id,
        text=text,
        status=status,
        order=order,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )

create_assistant_message_log_async(*, text, status, order, details=None, uncited_references=None, references=None) async

Creates a message log for the current assistant message asynchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
text	str	The log text content	required
status	MessageLogStatus	The status of this log entry	required
order	int	The order/sequence number of this log entry	required
details	MessageLogDetails | None	Additional details about this log entry	None
uncited_references	MessageLogUncitedReferences | None	References that are not cited	None
references	list[ContentReference] | None	List of references for this log	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The created message log

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_assistant_message_log_async(
    self,
    *,
    text: str,
    status: MessageLogStatus,
    order: int,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Creates a message log for the current assistant message asynchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        text (str): The log text content
        status (MessageLogStatus): The status of this log entry
        order (int): The order/sequence number of this log entry
        details (MessageLogDetails | None): Additional details about this log entry
        uncited_references (MessageLogUncitedReferences | None): References that are not cited
        references (list[ContentReference] | None): List of references for this log

    Returns:
        MessageLog: The created message log

    Raises:
        Exception: If the creation fails

    """
    return await self.create_message_log_async(
        message_id=self._assistant_message_id,
        text=text,
        status=status,
        order=order,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )

create_chat_memory(*, key, value)

Creates a short-term memory for the current chat synchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_chat_memory(
    self, *, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for the current chat synchronously.

    Args:
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    return self.create_chat_memory_by_id(
        chat_id=self._chat_id,
        key=key,
        value=value,
    )

create_chat_memory_async(*, key, value) async

Creates a short-term memory for the current chat asynchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_chat_memory_async(
    self, *, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for the current chat asynchronously.

    Args:
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    return await self.create_chat_memory_by_id_async(
        chat_id=self._chat_id,
        key=key,
        value=value,
    )

create_chat_memory_by_id(*, chat_id, key, value)

Creates a short-term memory for a specific chat synchronously.

Parameters:

Name	Type	Description	Default
chat_id	str	The chat ID	required
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_chat_memory_by_id(
    self, *, chat_id: str, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for a specific chat synchronously.

    Args:
        chat_id (str): The chat ID
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    # Convert BaseModel to JSON string if needed
    if isinstance(value, BaseModel):
        value = value.model_dump_json()

    return create_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        chat_id=chat_id,
    )

create_chat_memory_by_id_async(*, chat_id, key, value) async

Creates a short-term memory for a specific chat asynchronously.

Parameters:

Name	Type	Description	Default
chat_id	str	The chat ID	required
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_chat_memory_by_id_async(
    self, *, chat_id: str, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for a specific chat asynchronously.

    Args:
        chat_id (str): The chat ID
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    # Convert BaseModel to JSON string if needed
    if isinstance(value, BaseModel):
        value = value.model_dump_json()

    return await create_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        chat_id=chat_id,
    )

create_message_assessment(assistant_message_id, status, type, title=None, explanation=None, label=None, is_visible=True)

Creates a message assessment for an assistant message synchronously.

Parameters:

Name	Type	Description	Default
assistant_message_id	str	The ID of the assistant message to assess	required
status	MessageAssessmentStatus	The status of the assessment (e.g. "DONE")	required
type	MessageAssessmentType	The type of assessment (e.g. "HALLUCINATION")	required
title	str | None	The title of the assessment	None
explanation	str | None	Explanation of the assessment	None
label	MessageAssessmentLabel | None	The assessment label (e.g. "RED")	None
is_visible	bool	Whether the assessment is visible to users. Defaults to True.	True

Returns:

Name	Type	Description
ChatMessageAssessment	ChatMessageAssessment	The created message assessment

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_assessment(
    self,
    assistant_message_id: str,
    status: ChatMessageAssessmentStatus,
    type: ChatMessageAssessmentType,
    title: str | None = None,
    explanation: str | None = None,
    label: ChatMessageAssessmentLabel | None = None,
    is_visible: bool = True,
) -> ChatMessageAssessment:
    """Creates a message assessment for an assistant message synchronously.

    Args:
        assistant_message_id (str): The ID of the assistant message to assess
        status (MessageAssessmentStatus): The status of the assessment (e.g. "DONE")
        type (MessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")
        title (str | None): The title of the assessment
        explanation (str | None): Explanation of the assessment
        label (MessageAssessmentLabel | None): The assessment label (e.g. "RED")
        is_visible (bool): Whether the assessment is visible to users. Defaults to True.

    Returns:
        ChatMessageAssessment: The created message assessment

    Raises:
        Exception: If the creation fails

    """
    return create_message_assessment(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=assistant_message_id,
        status=status,
        type=type,
        title=title,
        explanation=explanation,
        label=label,
        is_visible=is_visible,
    )

create_message_assessment_async(assistant_message_id, status, type, title=None, explanation=None, label=None, is_visible=True) async

Creates a message assessment for an assistant message asynchronously.

Parameters:

Name	Type	Description	Default
assistant_message_id	str	The ID of the assistant message to assess	required
status	ChatMessageAssessmentStatus	The status of the assessment (e.g. "DONE")	required
type	ChatMessageAssessmentType	The type of assessment (e.g. "HALLUCINATION")	required
title	str | None	The title of the assessment	None
explanation	str | None	Explanation of the assessment	None
label	ChatMessageAssessmentLabel | None	The assessment label (e.g. "RED")	None
is_visible	bool	Whether the assessment is visible to users. Defaults to True.	True

Returns:

Name	Type	Description
ChatMessageAssessment	ChatMessageAssessment	The created message assessment

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_assessment_async(
    self,
    assistant_message_id: str,
    status: ChatMessageAssessmentStatus,
    type: ChatMessageAssessmentType,
    title: str | None = None,
    explanation: str | None = None,
    label: ChatMessageAssessmentLabel | None = None,
    is_visible: bool = True,
) -> ChatMessageAssessment:
    """Creates a message assessment for an assistant message asynchronously.

    Args:
        assistant_message_id (str): The ID of the assistant message to assess
        status (ChatMessageAssessmentStatus): The status of the assessment (e.g. "DONE")
        type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")
        title (str | None): The title of the assessment
        explanation (str | None): Explanation of the assessment
        label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
        is_visible (bool): Whether the assessment is visible to users. Defaults to True.

    Returns:
        ChatMessageAssessment: The created message assessment

    Raises:
        Exception: If the creation fails

    """
    return await create_message_assessment_async(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=assistant_message_id,
        status=status,
        type=type,
        title=title,
        explanation=explanation,
        label=label,
        is_visible=is_visible,
    )

create_message_execution(*, message_id, type=MessageExecutionType.DEEP_RESEARCH, seconds_remaining=None, percentage_completed=None, is_queueable=True, execution_options=None, progress_title=None)

Creates a message execution for tracking long-running operations synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message this execution belongs to	required
type	MessageExecutionType	The type of execution. Defaults to DEEP_RESEARCH.	DEEP_RESEARCH
seconds_remaining	int | None	Estimated seconds remaining for completion	None
percentage_completed	int | None	Percentage of completion (0-100)	None
is_queueable	bool	Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.	True
execution_options	dict | None	Additional execution options. Defaults to None.	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The created message execution

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_execution(
    self,
    *,
    message_id: str,
    type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    is_queueable: bool = True,
    execution_options: dict | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Creates a message execution for tracking long-running operations synchronously.

    Args:
        message_id (str): The ID of the message this execution belongs to
        type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
        seconds_remaining (int | None): Estimated seconds remaining for completion
        percentage_completed (int | None): Percentage of completion (0-100)
        is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
        execution_options (dict | None): Additional execution options. Defaults to None.
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The created message execution

    Raises:
        Exception: If the creation fails

    """
    return create_message_execution(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        chat_id=self._chat_id,
        type=type,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        is_queueable=is_queueable,
        execution_options=execution_options,
        progress_title=progress_title,
    )

create_message_execution_async(*, message_id, type=MessageExecutionType.DEEP_RESEARCH, seconds_remaining=None, percentage_completed=None, is_queueable=True, execution_options=None, progress_title=None) async

Creates a message execution for tracking long-running operations asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message this execution belongs to	required
type	MessageExecutionType	The type of execution. Defaults to DEEP_RESEARCH.	DEEP_RESEARCH
seconds_remaining	int | None	Estimated seconds remaining for completion	None
percentage_completed	int | None	Percentage of completion (0-100)	None
is_queueable	bool	Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.	True
execution_options	dict | None	Additional execution options. Defaults to None.	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The created message execution

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_execution_async(
    self,
    *,
    message_id: str,
    type: MessageExecutionType = MessageExecutionType.DEEP_RESEARCH,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    is_queueable: bool = True,
    execution_options: dict | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Creates a message execution for tracking long-running operations asynchronously.

    Args:
        message_id (str): The ID of the message this execution belongs to
        type (MessageExecutionType): The type of execution. Defaults to DEEP_RESEARCH.
        seconds_remaining (int | None): Estimated seconds remaining for completion
        percentage_completed (int | None): Percentage of completion (0-100)
        is_queueable (bool): Whether the execution is queueable. Defaults to True. If true, then the progress will be updated in the background by the execution pipeline. Set to False if you want to update the progress manually.
        execution_options (dict | None): Additional execution options. Defaults to None.
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The created message execution

    Raises:
        Exception: If the creation fails

    """
    return await create_message_execution_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        chat_id=self._chat_id,
        type=type,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        is_queueable=is_queueable,
        execution_options=execution_options,
        progress_title=progress_title,
    )

create_message_log(*, message_id, text, status, order, details=None, uncited_references=None, references=None)

Creates a message log for tracking execution steps synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message this log belongs to	required
text	str	The log text content	required
status	MessageLogStatus	The status of this log entry	required
order	int	The order/sequence number of this log entry	required
details	MessageLogDetails | None	Additional details about this log entry	None
uncited_references	MessageLogUncitedReferences | None	References that are not cited	None
references	list[ContentReference] | None	List of references for this log	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The created message log

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_log(
    self,
    *,
    message_id: str,
    text: str,
    status: MessageLogStatus,
    order: int,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Creates a message log for tracking execution steps synchronously.

    Args:
        message_id (str): The ID of the message this log belongs to
        text (str): The log text content
        status (MessageLogStatus): The status of this log entry
        order (int): The order/sequence number of this log entry
        details (MessageLogDetails | None): Additional details about this log entry
        uncited_references (MessageLogUncitedReferences | None): References that are not cited
        references (list[ContentReference] | None): List of references for this log

    Returns:
        MessageLog: The created message log

    Raises:
        Exception: If the creation fails

    """
    return create_message_log(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        text=text,
        status=status,
        order=order,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )

create_message_log_async(*, message_id, text, status, order, details=None, uncited_references=None, references=None) async

Creates a message log for tracking execution steps asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message this log belongs to	required
text	str	The log text content	required
status	MessageLogStatus	The status of this log entry	required
order	int	The order/sequence number of this log entry	required
details	MessageLogDetails | None	Additional details about this log entry	None
uncited_references	MessageLogUncitedReferences | None	References that are not cited	None
references	list[ContentReference] | None	List of references for this log	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The created message log

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_log_async(
    self,
    *,
    message_id: str,
    text: str,
    status: MessageLogStatus,
    order: int,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Creates a message log for tracking execution steps asynchronously.

    Args:
        message_id (str): The ID of the message this log belongs to
        text (str): The log text content
        status (MessageLogStatus): The status of this log entry
        order (int): The order/sequence number of this log entry
        details (MessageLogDetails | None): Additional details about this log entry
        uncited_references (MessageLogUncitedReferences | None): References that are not cited
        references (list[ContentReference] | None): List of references for this log

    Returns:
        MessageLog: The created message log

    Raises:
        Exception: If the creation fails

    """
    return await create_message_log_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        text=text,
        status=status,
        order=order,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )

create_message_memory(*, key, value, message_id=None)

create_message_memory(
    *, key: str, value: str | dict | BaseModel
) -> ShortTermMemory

create_message_memory(
    *,
    key: str,
    value: str | dict | BaseModel,
    message_id: str,
) -> ShortTermMemory

Creates a short-term memory for the current assistant message synchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_memory(
    self, *, key: str, value: str | dict | BaseModel, message_id: str | None = None
) -> ShortTermMemory:
    """Creates a short-term memory for the current assistant message synchronously.

    Args:
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    return self.create_message_memory_by_id(
        key=key,
        value=value,
        message_id=message_id or self._assistant_message_id,
    )

create_message_memory_async(*, key, value, message_id=None) async

create_message_memory_async(
    *, key: str, value: str | dict | BaseModel
) -> ShortTermMemory

create_message_memory_async(
    *,
    key: str,
    value: str | dict | BaseModel,
    message_id: str,
) -> ShortTermMemory

Creates a short-term memory for the current assistant message asynchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_memory_async(
    self, *, key: str, value: str | dict | BaseModel, message_id: str | None = None
) -> ShortTermMemory:
    """Creates a short-term memory for the current assistant message asynchronously.

    Args:
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    return await self.create_message_memory_by_id_async(
        message_id=message_id or self._assistant_message_id,
        key=key,
        value=value,
    )

create_message_memory_by_id(*, message_id, key, value)

Creates a short-term memory for a specific message synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The message ID	required
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_memory_by_id(
    self, *, message_id: str, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for a specific message synchronously.

    Args:
        message_id (str): The message ID
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    # Convert BaseModel to JSON string if needed
    if isinstance(value, BaseModel):
        value = value.model_dump_json()

    return create_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        message_id=message_id,
    )

create_message_memory_by_id_async(*, message_id, key, value) async

Creates a short-term memory for a specific message asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The message ID	required
key	str	The memory key	required
value	str | dict | BaseModel	The memory value	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The created short-term memory

Raises:

Type	Description
Exception	If the creation fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_memory_by_id_async(
    self, *, message_id: str, key: str, value: str | dict | BaseModel
) -> ShortTermMemory:
    """Creates a short-term memory for a specific message asynchronously.

    Args:
        message_id (str): The message ID
        key (str): The memory key
        value (str | dict | BaseModel): The memory value

    Returns:
        ShortTermMemory: The created short-term memory

    Raises:
        Exception: If the creation fails
    """
    # Convert BaseModel to JSON string if needed
    if isinstance(value, BaseModel):
        value = value.model_dump_json()

    return await create_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        message_id=message_id,
    )

create_message_tools(*, tool_calls, message_id=None)

Persist tool call records for an assistant message.

Parameters:

Name	Type	Description	Default
tool_calls	list[ChatMessageTool]	The tool call records to persist.	required
message_id	str | None	The assistant message to attach records to. Defaults to the current turn's assistant message.	None

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_message_tools(
    self,
    *,
    tool_calls: list[ChatMessageTool],
    message_id: str | None = None,
) -> list[ChatMessageTool]:
    """Persist tool call records for an assistant message.

    Args:
        tool_calls: The tool call records to persist.
        message_id: The assistant message to attach records to.
            Defaults to the current turn's assistant message.
    """
    return create_message_tools(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id or self._assistant_message_id,
        tool_calls=tool_calls,
    )

create_message_tools_async(*, tool_calls, message_id=None) async

Async variant of create_message_tools.

Parameters:

Name	Type	Description	Default
tool_calls	list[ChatMessageTool]	The tool call records to persist.	required
message_id	str | None	The assistant message to attach records to. Defaults to the current turn's assistant message.	None

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_message_tools_async(
    self,
    *,
    tool_calls: list[ChatMessageTool],
    message_id: str | None = None,
) -> list[ChatMessageTool]:
    """Async variant of create_message_tools.

    Args:
        tool_calls: The tool call records to persist.
        message_id: The assistant message to attach records to.
            Defaults to the current turn's assistant message.
    """
    return await create_message_tools_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id or self._assistant_message_id,
        tool_calls=tool_calls,
    )

create_user_message(content, original_content=None, references=None, debug_info=None, set_completed_at=False)

Creates a user message in the chat session synchronously.

Parameters:

Name	Type	Description	Default
content	str	The content for the message.	required
original_content	str	The original content for the message.	None
references	list[ContentReference]	list of ContentReference objects. Defaults to None.	None
debug_info	dict[str, Any]]	Debug information. Defaults to None.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The created message.

Raises:

Type	Description
Exception	If the creation fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def create_user_message(
    self,
    content: str,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Creates a user message in the chat session synchronously.

    Args:
        content (str): The content for the message.
        original_content (str, optional): The original content for the message.
        references (list[ContentReference]): list of ContentReference objects. Defaults to None.
        debug_info (dict[str, Any]]): Debug information. Defaults to None.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The created message.

    Raises:
        Exception: If the creation fails.

    """
    chat_message = create_message(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._chat_id,
        assistant_id=self._assistant_id,
        role=ChatMessageRole.USER,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        set_completed_at=set_completed_at,
    )
    # Update the user message id
    self._user_message_id = chat_message.id or "unknown"
    return chat_message

create_user_message_async(content, original_content=None, references=None, debug_info=None, set_completed_at=False) async

Creates a user message in the chat session asynchronously.

Parameters:

Name	Type	Description	Default
content	str	The content for the message.	required
original_content	str	The original content for the message.	None
references	list[ContentReference]	list of references. Defaults to None.	None
debug_info	dict[str, Any]]	Debug information. Defaults to None.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The created message.

Raises:

Type	Description
Exception	If the creation fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def create_user_message_async(
    self,
    content: str,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Creates a user message in the chat session asynchronously.

    Args:
        content (str): The content for the message.
        original_content (str, optional): The original content for the message.
        references (list[ContentReference]): list of references. Defaults to None.
        debug_info (dict[str, Any]]): Debug information. Defaults to None.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The created message.

    Raises:
        Exception: If the creation fails.

    """
    chat_message = await create_message_async(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._chat_id,
        assistant_id=self._assistant_id,
        role=ChatMessageRole.USER,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        set_completed_at=set_completed_at,
    )
    # Update the user message id
    self._user_message_id = chat_message.id or "unknown"
    return chat_message

download_chat_content_to_bytes(*, content_id)

Download content by id from the content-scope chat (e.g. parent chat when subagent).

Uses the service's content-scope chat id, so when running as a subagent with correlation, this accesses files from the primary chat session.

Parameters:

Name	Type	Description	Default
content_id	str	The content id to download.	required

Returns:

Name	Type	Description
bytes	bytes	The raw content bytes.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def download_chat_content_to_bytes(self, *, content_id: str) -> bytes:
    """Download content by id from the content-scope chat (e.g. parent chat when subagent).

    Uses the service's content-scope chat id, so when running as a subagent
    with correlation, this accesses files from the primary chat session.

    Args:
        content_id: The content id to download.

    Returns:
        bytes: The raw content bytes.
    """
    return download_content_to_bytes(
        user_id=self._user_id,
        company_id=self._company_id,
        content_id=content_id,
        chat_id=self._content_scope_chat_id,
    )

download_chat_content_to_bytes_async(*, content_id) async

Async download content by id from the content-scope chat (e.g. parent chat when subagent).

Uses the service's content-scope chat id, so when running as a subagent with correlation, this accesses files from the primary chat session.

Parameters:

Name	Type	Description	Default
content_id	str	The content id to download.	required

Returns:

Name	Type	Description
bytes	bytes	The raw content bytes.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def download_chat_content_to_bytes_async(self, *, content_id: str) -> bytes:
    """Async download content by id from the content-scope chat (e.g. parent chat when subagent).

    Uses the service's content-scope chat id, so when running as a subagent
    with correlation, this accesses files from the primary chat session.

    Args:
        content_id: The content id to download.

    Returns:
        bytes: The raw content bytes.
    """
    return await download_content_to_bytes_async(
        user_id=self._user_id,
        company_id=self._company_id,
        content_id=content_id,
        chat_id=self._content_scope_chat_id,
    )

download_chat_images_and_documents()

Return images and documents from the content-scope chat (e.g. parent chat when subagent).

Uses the service's content-scope chat id, so when running as a subagent with correlation, this accesses files from the primary chat session.

Returns:

Type	Description
tuple[list[Content], list[Content]]	tuple[list[Content], list[Content]]: (images, documents) from the content-scope chat.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def download_chat_images_and_documents(self) -> tuple[list[Content], list[Content]]:
    """Return images and documents from the content-scope chat (e.g. parent chat when subagent).

    Uses the service's content-scope chat id, so when running as a subagent
    with correlation, this accesses files from the primary chat session.

    Returns:
        tuple[list[Content], list[Content]]: (images, documents) from the content-scope chat.
    """
    images: list[Content] = []
    files: list[Content] = []
    for c in search_contents(
        user_id=self._user_id,
        company_id=self._company_id,
        chat_id=self._content_scope_chat_id,
        where={"ownerId": {"equals": self._content_scope_chat_id}},
    ):
        if is_file_content(filename=c.key):
            files.append(c)
        if is_image_content(filename=c.key):
            images.append(c)
    return images, files

find_chat_memory(*, key)

Finds the latest short-term memory for the current chat synchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def find_chat_memory(self, *, key: str) -> ShortTermMemory:
    """Finds the latest short-term memory for the current chat synchronously.

    Args:
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return self.find_chat_memory_by_id(
        chat_id=self._chat_id,
        key=key,
    )

find_chat_memory_async(*, key) async

Finds the latest short-term memory for the current chat asynchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def find_chat_memory_async(self, *, key: str) -> ShortTermMemory:
    """Finds the latest short-term memory for the current chat asynchronously.

    Args:
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return await self.find_chat_memory_by_id_async(
        chat_id=self._chat_id,
        key=key,
    )

find_chat_memory_by_id(*, chat_id, key)

Finds the latest short-term memory for a specific chat synchronously.

Parameters:

Name	Type	Description	Default
chat_id	str	The chat ID	required
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def find_chat_memory_by_id(self, *, chat_id: str, key: str) -> ShortTermMemory:
    """Finds the latest short-term memory for a specific chat synchronously.

    Args:
        chat_id (str): The chat ID
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return find_latest_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        chat_id=chat_id,
    )

find_chat_memory_by_id_async(*, chat_id, key) async

Finds the latest short-term memory for a specific chat asynchronously.

Parameters:

Name	Type	Description	Default
chat_id	str	The chat ID	required
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def find_chat_memory_by_id_async(
    self, *, chat_id: str, key: str
) -> ShortTermMemory:
    """Finds the latest short-term memory for a specific chat asynchronously.

    Args:
        chat_id (str): The chat ID
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return await find_latest_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        chat_id=chat_id,
    )

find_message_memory(*, key, message_id=None)

find_message_memory(*, key: str) -> ShortTermMemory

find_message_memory(
    *, key: str, message_id: str
) -> ShortTermMemory

Finds the latest short-term memory for the current assistant message synchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def find_message_memory(
    self, *, key: str, message_id: str | None = None
) -> ShortTermMemory:
    """Finds the latest short-term memory for the current assistant message synchronously.

    Args:
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return self.find_message_memory_by_id(
        message_id=message_id or self._assistant_message_id,
        key=key,
    )

find_message_memory_async(*, key, message_id=None) async

find_message_memory_async(*, key: str) -> ShortTermMemory

find_message_memory_async(
    *, key: str, message_id: str
) -> ShortTermMemory

Finds the latest short-term memory for the current assistant message asynchronously.

Parameters:

Name	Type	Description	Default
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def find_message_memory_async(
    self, *, key: str, message_id: str | None = None
) -> ShortTermMemory:
    """Finds the latest short-term memory for the current assistant message asynchronously.

    Args:
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return await self.find_message_memory_by_id_async(
        message_id=message_id or self._assistant_message_id,
        key=key,
    )

find_message_memory_by_id(*, message_id, key)

Finds the latest short-term memory for a specific message synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The message ID	required
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def find_message_memory_by_id(
    self, *, message_id: str, key: str
) -> ShortTermMemory:
    """Finds the latest short-term memory for a specific message synchronously.

    Args:
        message_id (str): The message ID
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return find_latest_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        message_id=message_id,
    )

find_message_memory_by_id_async(*, message_id, key) async

Finds the latest short-term memory for a specific message asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The message ID	required
key	str	The memory key	required

Returns:

Name	Type	Description
ShortTermMemory	ShortTermMemory	The latest short-term memory

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def find_message_memory_by_id_async(
    self, *, message_id: str, key: str
) -> ShortTermMemory:
    """Finds the latest short-term memory for a specific message asynchronously.

    Args:
        message_id (str): The message ID
        key (str): The memory key

    Returns:
        ShortTermMemory: The latest short-term memory

    Raises:
        Exception: If the retrieval fails
    """
    return await find_latest_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        message_id=message_id,
    )

free_user_input()

Unblocks the next user input

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def free_user_input(self) -> None:
    """Unblocks the next user input"""
    self.modify_assistant_message(set_completed_at=True)

from_settings(settings, **kwargs) classmethod

Create a ChatService from a :class:UniqueSettings or :class:UniqueContext.

This is the preferred constructor when using the service factory pattern.

Parameters:

Name	Type	Description	Default
settings	UniqueSettings	Either a :class:UniqueSettings (whose context is used) or a :class:UniqueContext carrying auth and chat information directly.	required

Raises:

Type	Description
ValueError	If the resolved context has no chat or last_user_message_id is missing.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

@classmethod
def from_settings(cls, settings: UniqueSettings, **kwargs: Any) -> Self:
    """Create a ChatService from a :class:`UniqueSettings` or :class:`UniqueContext`.

    This is the preferred constructor when using the service factory pattern.

    Args:
        settings: Either a :class:`UniqueSettings` (whose ``context`` is used)
            or a :class:`UniqueContext` carrying auth and chat information directly.

    Raises:
        ValueError: If the resolved context has no chat or ``last_user_message_id`` is missing.
    """
    return cls.from_context(context=settings.context)

get_assistant_message_execution()

Gets the message execution for the current assistant message synchronously.

This is a convenience method that uses the current assistant message ID.

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The message execution

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_assistant_message_execution(self) -> MessageExecution:
    """Gets the message execution for the current assistant message synchronously.

    This is a convenience method that uses the current assistant message ID.

    Returns:
        MessageExecution: The message execution

    Raises:
        Exception: If the retrieval fails

    """
    return self.get_message_execution(message_id=self._assistant_message_id)

get_assistant_message_execution_async() async

Gets the message execution for the current assistant message asynchronously.

This is a convenience method that uses the current assistant message ID.

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The message execution

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_assistant_message_execution_async(self) -> MessageExecution:
    """Gets the message execution for the current assistant message asynchronously.

    This is a convenience method that uses the current assistant message ID.

    Returns:
        MessageExecution: The message execution

    Raises:
        Exception: If the retrieval fails

    """
    return await self.get_message_execution_async(
        message_id=self._assistant_message_id
    )

get_debug_info()

Retrieves the debug information from the current user message.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: The debug information stored on the user message, or an empty dict if retrieval fails or no debug info exists.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_debug_info(self) -> dict[str, Any]:
    """Retrieves the debug information from the current user message.

    Returns:
        dict[str, Any]: The debug information stored on the user message,
            or an empty dict if retrieval fails or no debug info exists.

    """
    try:
        raw_msg = unique_sdk.Message.retrieve(
            user_id=self._user_id,
            company_id=self._company_id,
            id=self._user_message_id,
            chatId=self._chat_id,
        )
        return getattr(raw_msg, "debugInfo", {}) or {}
    except Exception:
        logging.getLogger(__name__).warning(
            "Failed to retrieve debug info from user message", exc_info=True
        )
        return {}

get_debug_info_async() async

Retrieves the debug information from the current user message asynchronously.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: The debug information stored on the user message, or an empty dict if retrieval fails or no debug info exists.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_debug_info_async(self) -> dict[str, Any]:
    """Retrieves the debug information from the current user message asynchronously.

    Returns:
        dict[str, Any]: The debug information stored on the user message,
            or an empty dict if retrieval fails or no debug info exists.

    """
    try:
        raw_msg = await unique_sdk.Message.retrieve_async(
            user_id=self._user_id,
            company_id=self._company_id,
            id=self._user_message_id,
            chatId=self._chat_id,
        )
        return getattr(raw_msg, "debugInfo", {}) or {}
    except Exception:
        logging.getLogger(__name__).warning(
            "Failed to retrieve debug info from user message", exc_info=True
        )
        return {}

get_full_and_selected_history(token_limit, percent_of_max_tokens=DEFAULT_PERCENT_OF_MAX_TOKENS, max_messages=DEFAULT_MAX_MESSAGES)

Loads the chat history for the chat session synchronously.

Parameters:

Name	Type	Description	Default
token_limit	int	The maximum number of tokens to load.	required
percent_of_max_tokens	float	The percentage of the maximum tokens to load. Defaults to 0.15.	DEFAULT_PERCENT_OF_MAX_TOKENS
max_messages	int	The maximum number of messages to load. Defaults to 4.	DEFAULT_MAX_MESSAGES

Returns:

Type	Description
tuple[list[ChatMessage], list[ChatMessage]]	tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

Raises:

Type	Description
Exception	If the loading fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_full_and_selected_history(
    self,
    token_limit: int,
    percent_of_max_tokens: float = DEFAULT_PERCENT_OF_MAX_TOKENS,
    max_messages: int = DEFAULT_MAX_MESSAGES,
) -> tuple[list[ChatMessage], list[ChatMessage]]:
    """Loads the chat history for the chat session synchronously.

    Args:
        token_limit (int): The maximum number of tokens to load.
        percent_of_max_tokens (float): The percentage of the maximum tokens to load. Defaults to 0.15.
        max_messages (int): The maximum number of messages to load. Defaults to 4.

    Returns:
        tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

    Raises:
        Exception: If the loading fails.

    """
    full_history = get_full_history(
        event_user_id=self._user_id,
        event_company_id=self._company_id,
        event_payload_chat_id=self._chat_id,
    )
    selected_history = get_selection_from_history(
        full_history=full_history,
        max_tokens=int(round(token_limit * percent_of_max_tokens)),
        max_messages=max_messages,
        model_info=None,  # TODO: Pass language_model when available
    )

    return full_history, selected_history

get_full_and_selected_history_async(token_limit, percent_of_max_tokens=DEFAULT_PERCENT_OF_MAX_TOKENS, max_messages=DEFAULT_MAX_MESSAGES) async

Loads the chat history for the chat session asynchronously.

Parameters:

Name	Type	Description	Default
token_limit	int	The maximum number of tokens to load.	required
percent_of_max_tokens	float	The percentage of the maximum tokens to load. Defaults to 0.15.	DEFAULT_PERCENT_OF_MAX_TOKENS
max_messages	int	The maximum number of messages to load. Defaults to 4.	DEFAULT_MAX_MESSAGES

Returns:

Type	Description
tuple[list[ChatMessage], list[ChatMessage]]	tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

Raises:

Type	Description
Exception	If the loading fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_full_and_selected_history_async(
    self,
    token_limit: int,
    percent_of_max_tokens: float = DEFAULT_PERCENT_OF_MAX_TOKENS,
    max_messages: int = DEFAULT_MAX_MESSAGES,
) -> tuple[list[ChatMessage], list[ChatMessage]]:
    """Loads the chat history for the chat session asynchronously.

    Args:
        token_limit (int): The maximum number of tokens to load.
        percent_of_max_tokens (float): The percentage of the maximum tokens to load. Defaults to 0.15.
        max_messages (int): The maximum number of messages to load. Defaults to 4.

    Returns:
        tuple[list[ChatMessage], list[ChatMessage]]: The selected and full chat history.

    Raises:
        Exception: If the loading fails.

    """
    full_history = await get_full_history_async(
        event_user_id=self._user_id,
        event_company_id=self._company_id,
        event_payload_chat_id=self._chat_id,
    )
    selected_history = get_selection_from_history(
        full_history=full_history,
        max_tokens=int(round(token_limit * percent_of_max_tokens)),
        max_messages=max_messages,
        model_info=None,  # TODO: Pass language_model when available
    )

    return full_history, selected_history

get_full_history()

Loads the full chat history for the chat session synchronously.

Returns:

Type	Description
list[ChatMessage]	list[ChatMessage]: The full chat history.

Raises:

Type	Description
Exception	If the loading fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_full_history(self) -> list[ChatMessage]:
    """Loads the full chat history for the chat session synchronously.

    Returns:
        list[ChatMessage]: The full chat history.

    Raises:
        Exception: If the loading fails.

    """
    return get_full_history(
        event_user_id=self._user_id,
        event_company_id=self._company_id,
        event_payload_chat_id=self._chat_id,
    )

get_full_history_async() async

Loads the full chat history for the chat session asynchronously.

Returns:

Type	Description
list[ChatMessage]	list[ChatMessage]: The full chat history.

Raises:

Type	Description
Exception	If the loading fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_full_history_async(self) -> list[ChatMessage]:
    """Loads the full chat history for the chat session asynchronously.

    Returns:
        list[ChatMessage]: The full chat history.

    Raises:
        Exception: If the loading fails.

    """
    return await get_full_history_async(
        event_user_id=self._user_id,
        event_company_id=self._company_id,
        event_payload_chat_id=self._chat_id,
    )

get_message_execution(*, message_id)

Gets a message execution by message ID synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message to get execution for	required

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The message execution

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_message_execution(
    self,
    *,
    message_id: str,
) -> MessageExecution:
    """Gets a message execution by message ID synchronously.

    Args:
        message_id (str): The ID of the message to get execution for

    Returns:
        MessageExecution: The message execution

    Raises:
        Exception: If the retrieval fails

    """
    return get_message_execution(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
    )

get_message_execution_async(*, message_id) async

Gets a message execution by message ID asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message to get execution for	required

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The message execution

Raises:

Type	Description
Exception	If the retrieval fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_message_execution_async(
    self,
    *,
    message_id: str,
) -> MessageExecution:
    """Gets a message execution by message ID asynchronously.

    Args:
        message_id (str): The ID of the message to get execution for

    Returns:
        MessageExecution: The message execution

    Raises:
        Exception: If the retrieval fails

    """
    return await get_message_execution_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
    )

get_message_tools(*, message_id=None, message_ids=None)

Fetch persisted tool call records for one or more assistant messages.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def get_message_tools(
    self,
    *,
    message_id: str | None = None,
    message_ids: list[str] | None = None,
) -> list[ChatMessageTool]:
    """Fetch persisted tool call records for one or more assistant messages."""
    return get_message_tools(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        message_ids=message_ids,
    )

get_message_tools_async(*, message_id=None, message_ids=None) async

Async variant of get_message_tools.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def get_message_tools_async(
    self,
    *,
    message_id: str | None = None,
    message_ids: list[str] | None = None,
) -> list[ChatMessageTool]:
    """Async variant of get_message_tools."""
    return await get_message_tools_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        message_ids=message_ids,
    )

modify_assistant_message(content=None, original_content=None, references=None, debug_info=None, message_id=None, set_completed_at=None)

Modifies a message in the chat session synchronously if parameter is not specified the corresponding field will remain as is.

Parameters:

Name	Type	Description	Default
content	str	The new content for the message.	None
original_content	str	The original content for the message.	None
references	list[ContentReference]	list of ContentReference objects. Defaults to [].	None
debug_info	dict[str, Any]]]	Debug information. Defaults to {}.	None
message_id	Optional[str]	The message ID. Defaults to None.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	None

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The modified message.

Raises:

Type	Description
Exception	If the modification fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def modify_assistant_message(
    self,
    content: str | None = None,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    message_id: str | None = None,
    set_completed_at: bool | None = None,
) -> ChatMessage:
    """Modifies a message in the chat session synchronously if parameter is not specified the corresponding field will remain as is.

    Args:
        content (str, optional): The new content for the message.
        original_content (str, optional): The original content for the message.
        references (list[ContentReference]): list of ContentReference objects. Defaults to [].
        debug_info (dict[str, Any]]]): Debug information. Defaults to {}.
        message_id (Optional[str]): The message ID. Defaults to None.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The modified message.

    Raises:
        Exception: If the modification fails.

    """

    return modify_message(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=True,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        message_id=message_id,
        set_completed_at=set_completed_at or False,
    )

modify_assistant_message_async(content=None, original_content=None, references=None, debug_info=None, message_id=None, set_completed_at=False) async

Modifies a message in the chat session asynchronously.

Parameters:

Name	Type	Description	Default
content	str	The new content for the message.	None
original_content	str	The original content for the message.	None
message_id	str	The message ID. Defaults to None, then the ChatState assistant message id is used.	None
references	list[ContentReference]	list of ContentReference objects. Defaults to None.	None
debug_info	dict[str, Any]]	Debug information. Defaults to None.	None
set_completed_at	bool	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The modified message.

Raises:

Type	Description
Exception	If the modification fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def modify_assistant_message_async(
    self,
    content: str | None = None,
    original_content: str | None = None,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    message_id: str | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Modifies a message in the chat session asynchronously.

    Args:
        content (str, optional): The new content for the message.
        original_content (str, optional): The original content for the message.
        message_id (str, optional): The message ID. Defaults to None, then the ChatState assistant message id is used.
        references (list[ContentReference]): list of ContentReference objects. Defaults to None.
        debug_info (dict[str, Any]], optional): Debug information. Defaults to None.
        set_completed_at (bool, optional): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The modified message.

    Raises:
        Exception: If the modification fails.

    """
    return await modify_message_async(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=True,
        content=content,
        original_content=original_content,
        references=references,
        debug_info=debug_info,
        message_id=message_id,
        set_completed_at=set_completed_at or False,
    )

modify_message_assessment(assistant_message_id, status, type, title=None, explanation=None, label=None)

Modifies a message assessment for an assistant message synchronously.

Parameters:

Name	Type	Description	Default
assistant_message_id	str	The ID of the assistant message to assess	required
status	MessageAssessmentStatus	The status of the assessment (e.g. "DONE")	required
title	str | None	The title of the assessment	None
explanation	str | None	Explanation of the assessment	None
label	ChatMessageAssessmentLabel | None	The assessment label (e.g. "RED")	None
type	ChatMessageAssessmentType	The type of assessment (e.g. "HALLUCINATION")	required

Returns:

Name	Type	Description
dict	ChatMessageAssessment	The modified message assessment

Raises:

Type	Description
Exception	If the modification fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def modify_message_assessment(
    self,
    assistant_message_id: str,
    status: ChatMessageAssessmentStatus,
    type: ChatMessageAssessmentType,
    title: str | None = None,
    explanation: str | None = None,
    label: ChatMessageAssessmentLabel | None = None,
) -> ChatMessageAssessment:
    """Modifies a message assessment for an assistant message synchronously.

    Args:
        assistant_message_id (str): The ID of the assistant message to assess
        status (MessageAssessmentStatus): The status of the assessment (e.g. "DONE")
        title (str | None): The title of the assessment
        explanation (str | None): Explanation of the assessment
        label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
        type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")

    Returns:
        dict: The modified message assessment

    Raises:
        Exception: If the modification fails

    """
    return modify_message_assessment(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=assistant_message_id,
        status=status,
        type=type,
        title=title,
        explanation=explanation,
        label=label,
    )

modify_message_assessment_async(assistant_message_id, type, title=None, status=None, explanation=None, label=None) async

Modifies a message assessment for an assistant message asynchronously.

Parameters:

Name	Type	Description	Default
assistant_message_id	str	The ID of the assistant message to assess	required
status	ChatMessageAssessmentStatus	The status of the assessment (e.g. "DONE")	None
title	str | None	The title of the assessment	None
explanation	str | None	Explanation of the assessment	None
label	ChatMessageAssessmentLabel | None	The assessment label (e.g. "RED")	None
type	ChatMessageAssessmentType	The type of assessment (e.g. "HALLUCINATION")	required

Returns:

Name	Type	Description
ChatMessageAssessment	ChatMessageAssessment	The modified message assessment

Raises:

Type	Description
Exception	If the modification fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def modify_message_assessment_async(
    self,
    assistant_message_id: str,
    type: ChatMessageAssessmentType,
    title: str | None = None,
    status: ChatMessageAssessmentStatus | None = None,
    explanation: str | None = None,
    label: ChatMessageAssessmentLabel | None = None,
) -> ChatMessageAssessment:
    """Modifies a message assessment for an assistant message asynchronously.

    Args:
        assistant_message_id (str): The ID of the assistant message to assess
        status (ChatMessageAssessmentStatus): The status of the assessment (e.g. "DONE")
        title (str | None): The title of the assessment
        explanation (str | None): Explanation of the assessment
        label (ChatMessageAssessmentLabel | None): The assessment label (e.g. "RED")
        type (ChatMessageAssessmentType): The type of assessment (e.g. "HALLUCINATION")

    Returns:
        ChatMessageAssessment: The modified message assessment

    Raises:
        Exception: If the modification fails

    """
    return await modify_message_assessment_async(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=assistant_message_id,
        status=status,
        type=type,
        title=title,
        explanation=explanation,
        label=label,
    )

modify_user_message(content, references=None, debug_info=None, message_id=None, set_completed_at=False)

Modifies a user message in the chat session synchronously.

Parameters:

Name	Type	Description	Default
content	str	The new content for the message.	required
references	list[ContentReference]	list of ContentReference objects.	None
debug_info	dict[str, Any]]]	Debug information.	None
message_id	str	The message ID, if not specified the last user message is edited.	None
set_completed_at	Optional[bool]	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The modified message.

Raises:

Type	Description
Exception	If the modification fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def modify_user_message(
    self,
    content: str,
    references: list[ContentReference] | None = None,
    debug_info: dict | None = None,
    message_id: str | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Modifies a user message in the chat session synchronously.

    Args:
        content (str): The new content for the message.
        references (list[ContentReference]): list of ContentReference objects.
        debug_info (dict[str, Any]]]): Debug information.
        message_id (str, optional): The message ID, if not specified the last user message is edited.
        set_completed_at (Optional[bool]): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The modified message.

    Raises:
        Exception: If the modification fails.

    """
    return modify_message(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=False,
        content=content,
        references=references,
        debug_info=debug_info,
        message_id=message_id,
        set_completed_at=set_completed_at or False,
    )

modify_user_message_async(content, references=[], debug_info={}, message_id=None, set_completed_at=False) async

Modifies a message in the chat session asynchronously.

Parameters:

Name	Type	Description	Default
content	str	The new content for the message.	required
message_id	str	The message ID. Defaults to None, then the ChatState user message id is used.	None
references	list[ContentReference]	list of ContentReference objects. Defaults to None.	[]
debug_info	dict[str, Any]]]	Debug information. Defaults to {}.	{}
set_completed_at	bool	Whether to set the completedAt field with the current date time. Defaults to False.	False

Returns:

Name	Type	Description
ChatMessage	ChatMessage	The modified message.

Raises:

Type	Description
Exception	If the modification fails.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def modify_user_message_async(
    self,
    content: str,
    references: list[ContentReference] = [],
    debug_info: dict = {},
    message_id: str | None = None,
    set_completed_at: bool | None = False,
) -> ChatMessage:
    """Modifies a message in the chat session asynchronously.

    Args:
        content (str): The new content for the message.
        message_id (str, optional): The message ID. Defaults to None, then the ChatState user message id is used.
        references (list[ContentReference]): list of ContentReference objects. Defaults to None.
        debug_info (dict[str, Any]]]): Debug information. Defaults to {}.
        set_completed_at (bool, optional): Whether to set the completedAt field with the current date time. Defaults to False.

    Returns:
        ChatMessage: The modified message.

    Raises:
        Exception: If the modification fails.

    """
    return await modify_message_async(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=False,
        content=content,
        references=references,
        debug_info=debug_info,
        message_id=message_id,
        set_completed_at=set_completed_at or False,
    )

replace_debug_info(debug_info)

Replace the debug information in the last user message

Parameters:

Name	Type	Description	Default
debug_info	dict	The new debug information.	required

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def replace_debug_info(self, debug_info: dict):
    """Replace the debug information in the last user message

    Args:
        debug_info (dict): The new debug information.

    """
    return modify_message(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=False,
        debug_info=debug_info,
    )

stream_complete_async(messages, model_name, content_chunks=None, debug_info=None, temperature=DEFAULT_COMPLETE_TEMPERATURE, timeout=DEFAULT_COMPLETE_TIMEOUT, tools=None, start_text=None, tool_choice=None, other_options=None) async

Stream a completion in the chat session asynchronously.

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

@deprecated("use complete_with_references_async instead.")
async def stream_complete_async(
    self,
    messages: LanguageModelMessages | list[ChatCompletionMessageParam],
    model_name: LanguageModelName | str,
    content_chunks: list[ContentChunk] | None = None,
    debug_info: dict | None = None,
    temperature: float = DEFAULT_COMPLETE_TEMPERATURE,
    timeout: int = DEFAULT_COMPLETE_TIMEOUT,
    tools: Sequence[LanguageModelTool | LanguageModelToolDescription] | None = None,
    start_text: str | None = None,
    tool_choice: ChatCompletionToolChoiceOptionParam | None = None,
    other_options: dict | None = None,
) -> LanguageModelStreamResponse:
    """Stream a completion in the chat session asynchronously."""
    return await self.complete_with_references_async(
        messages=messages,
        model_name=model_name,
        content_chunks=content_chunks,
        debug_info=debug_info,
        temperature=temperature,
        timeout=timeout,
        tools=tools,
        start_text=start_text,
        tool_choice=tool_choice,
        other_options=other_options,
    )

update_assistant_message_execution(*, status=None, seconds_remaining=None, percentage_completed=None)

Updates the message execution for the current assistant message synchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
status	MessageExecutionUpdateStatus | None	The updated status (COMPLETED or FAILED). Defaults to None	None
seconds_remaining	int | None	Updated estimated seconds remaining	None
percentage_completed	int | None	Updated percentage of completion (0-100)	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The updated message execution

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def update_assistant_message_execution(
    self,
    *,
    status: MessageExecutionUpdateStatus | None = None,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
) -> MessageExecution:
    """Updates the message execution for the current assistant message synchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
        seconds_remaining (int | None): Updated estimated seconds remaining
        percentage_completed (int | None): Updated percentage of completion (0-100)

    Returns:
        MessageExecution: The updated message execution

    Raises:
        Exception: If the update fails

    """
    return self.update_message_execution(
        message_id=self._assistant_message_id,
        status=status,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
    )

update_assistant_message_execution_async(*, status=None, seconds_remaining=None, percentage_completed=None) async

Updates the message execution for the current assistant message asynchronously.

This is a convenience method that uses the current assistant message ID.

Parameters:

Name	Type	Description	Default
status	MessageExecutionUpdateStatus | None	The updated status (COMPLETED or FAILED). Defaults to None	None
seconds_remaining	int | None	Updated estimated seconds remaining	None
percentage_completed	int | None	Updated percentage of completion (0-100)	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The updated message execution

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def update_assistant_message_execution_async(
    self,
    *,
    status: MessageExecutionUpdateStatus | None = None,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
) -> MessageExecution:
    """Updates the message execution for the current assistant message asynchronously.

    This is a convenience method that uses the current assistant message ID.

    Args:
        status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
        seconds_remaining (int | None): Updated estimated seconds remaining
        percentage_completed (int | None): Updated percentage of completion (0-100)

    Returns:
        MessageExecution: The updated message execution

    Raises:
        Exception: If the update fails

    """
    return await self.update_message_execution_async(
        message_id=self._assistant_message_id,
        status=status,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
    )

update_debug_info_async(debug_info) async

Updates the debug information for the chat session.

Parameters:

Name	Type	Description	Default
debug_info	dict	The new debug information.	required

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def update_debug_info_async(self, debug_info: dict):
    """Updates the debug information for the chat session.

    Args:
        debug_info (dict): The new debug information.

    """
    return await modify_message_async(
        user_id=self._user_id,
        company_id=self._company_id,
        assistant_message_id=self._assistant_message_id,
        chat_id=self._chat_id,
        user_message_id=self._user_message_id,
        user_message_text=self._user_message_text,
        assistant=False,
        debug_info=debug_info,
    )

update_message_execution(*, message_id, status=None, seconds_remaining=None, percentage_completed=None, progress_title=None)

Updates a message execution synchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message to update execution for	required
status	MessageExecutionUpdateStatus | None	The updated status (COMPLETED or FAILED). Defaults to None	None
seconds_remaining	int | None	Updated estimated seconds remaining	None
percentage_completed	int | None	Updated percentage of completion (0-100)	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The updated message execution

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def update_message_execution(
    self,
    *,
    message_id: str,
    status: MessageExecutionUpdateStatus | None = None,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Updates a message execution synchronously.

    Args:
        message_id (str): The ID of the message to update execution for
        status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
        seconds_remaining (int | None): Updated estimated seconds remaining
        percentage_completed (int | None): Updated percentage of completion (0-100)
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The updated message execution

    Raises:
        Exception: If the update fails

    """
    return update_message_execution(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        status=status,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        progress_title=progress_title,
    )

update_message_execution_async(*, message_id, status=None, seconds_remaining=None, percentage_completed=None, progress_title=None) async

Updates a message execution asynchronously.

Parameters:

Name	Type	Description	Default
message_id	str	The ID of the message to update execution for	required
status	MessageExecutionUpdateStatus | None	The updated status (COMPLETED or FAILED). Defaults to None	None
seconds_remaining	int | None	Updated estimated seconds remaining	None
percentage_completed	int | None	Updated percentage of completion (0-100)	None
progress_title	str | None	The title of the progress bar. If not provided, the title of the last message log is taken.	None

Returns:

Name	Type	Description
MessageExecution	MessageExecution	The updated message execution

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def update_message_execution_async(
    self,
    *,
    message_id: str,
    status: MessageExecutionUpdateStatus | None = None,
    seconds_remaining: int | None = None,
    percentage_completed: int | None = None,
    progress_title: str | None = None,
) -> MessageExecution:
    """Updates a message execution asynchronously.

    Args:
        message_id (str): The ID of the message to update execution for
        status (MessageExecutionUpdateStatus | None): The updated status (COMPLETED or FAILED). Defaults to None
        seconds_remaining (int | None): Updated estimated seconds remaining
        percentage_completed (int | None): Updated percentage of completion (0-100)
        progress_title (str | None): The title of the progress bar. If not provided, the title of the last message log is taken.

    Returns:
        MessageExecution: The updated message execution

    Raises:
        Exception: If the update fails

    """
    return await update_message_execution_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_id=message_id,
        status=status,
        seconds_remaining=seconds_remaining,
        percentage_completed=percentage_completed,
        progress_title=progress_title,
    )

update_message_log(*, message_log_id, order, text=None, status=None, details=None, uncited_references=None, references=None)

Updates a message log synchronously.

Parameters:

Name	Type	Description	Default
message_log_id	str	The ID of the message log to update	required
order	int	The order/sequence number (required)	required
text	str | None	The updated log text content	None
status	MessageLogStatus | None	The updated status	None
details	MessageLogDetails | None	Updated additional details	None
uncited_references	MessageLogUncitedReferences | None	Updated uncited references	None
references	list[ContentReference] | None	Updated list of references	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The updated message log

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

def update_message_log(
    self,
    *,
    message_log_id: str,
    order: int,
    text: str | None = None,
    status: MessageLogStatus | None = None,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Updates a message log synchronously.

    Args:
        message_log_id (str): The ID of the message log to update
        order (int): The order/sequence number (required)
        text (str | None): The updated log text content
        status (MessageLogStatus | None): The updated status
        details (MessageLogDetails | None): Updated additional details
        uncited_references (MessageLogUncitedReferences | None): Updated uncited references
        references (list[ContentReference] | None): Updated list of references

    Returns:
        MessageLog: The updated message log

    Raises:
        Exception: If the update fails

    """
    return update_message_log(
        user_id=self._user_id,
        company_id=self._company_id,
        message_log_id=message_log_id,
        order=order,
        text=text,
        status=status,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )

update_message_log_async(*, message_log_id, order, text=None, status=None, details=None, uncited_references=None, references=None) async

Updates a message log asynchronously.

Parameters:

Name	Type	Description	Default
message_log_id	str	The ID of the message log to update	required
order	int	The order/sequence number (required)	required
text	str | None	The updated log text content	None
status	MessageLogStatus | None	The updated status	None
details	MessageLogDetails | None	Updated additional details	None
uncited_references	MessageLogUncitedReferences | None	Updated uncited references	None
references	list[ContentReference] | None	Updated list of references	None

Returns:

Name	Type	Description
MessageLog	MessageLog	The updated message log

Raises:

Type	Description
Exception	If the update fails

Source code in unique_toolkit/unique_toolkit/services/chat_service.py

async def update_message_log_async(
    self,
    *,
    message_log_id: str,
    order: int,
    text: str | None = None,
    status: MessageLogStatus | None = None,
    details: MessageLogDetails | None = None,
    uncited_references: MessageLogUncitedReferences | None = None,
    references: list[ContentReference] | None = None,
) -> MessageLog:
    """Updates a message log asynchronously.

    Args:
        message_log_id (str): The ID of the message log to update
        order (int): The order/sequence number (required)
        text (str | None): The updated log text content
        status (MessageLogStatus | None): The updated status
        details (MessageLogDetails | None): Updated additional details
        uncited_references (MessageLogUncitedReferences | None): Updated uncited references
        references (list[ContentReference] | None): Updated list of references

    Returns:
        MessageLog: The updated message log

    Raises:
        Exception: If the update fails

    """
    return await update_message_log_async(
        user_id=self._user_id,
        company_id=self._company_id,
        message_log_id=message_log_id,
        order=order,
        text=text,
        status=status,
        details=details,
        uncited_references=uncited_references,
        references=references,
    )