Pipeline Events and APIs

class PersonMessageReceived(BaseEventModel):
    """When any private chat message is received"""

    event_name: str = "PersonMessageReceived"

    launcher_type: str
    """Launcher object type (person)"""

    launcher_id: typing.Union[int, str]
    """Sender ID"""

    sender_id: typing.Union[int, str]
    """Sender ID, same as launcher_id in private chat"""

    message_event: platform_events.PersonMessage
    """Original message event object. Contains sender information."""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """Message chain"""


class GroupMessageReceived(BaseEventModel):
    """When any group chat message is received"""

    event_name: str = "GroupMessageReceived"

    launcher_type: str
    """Launcher object type (group)"""

    launcher_id: typing.Union[int, str]
    """Group ID"""

    sender_id: typing.Union[int, str]
    """Sender ID"""

    message_event: platform_events.GroupMessage
    """Original message event object. Contains group and sender information."""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """Message chain"""

class PersonNormalMessageReceived(BaseEventModel):
    """Triggered when a private chat normal message that should be processed is determined"""

    event_name: str = "PersonNormalMessageReceived"

    launcher_type: str
    """Launcher object type (person)"""

    launcher_id: typing.Union[int, str]
    """Launcher object ID"""

    sender_id: typing.Union[int, str]
    """Sender ID, same as launcher_id in private chat"""

    text_message: str
    """Message text"""

    message_event: platform_events.PersonMessage
    """Original message event object. Contains sender information."""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """Message chain"""

    # ========== Settable Attributes ==========

    user_message_alter: typing.Optional[provider_message.ContentElement] = None
    """Modified message text, langbot_plugin.api.entities.builtin.provider.message.ContentElement type"""

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """Direct reply message chain, only effective when preventing default behavior"""


class GroupNormalMessageReceived(BaseEventModel):
    """Triggered when a group chat normal message that should be processed is determined"""

    event_name: str = "GroupNormalMessageReceived"

    launcher_type: str
    """Launcher object type (group)"""

    launcher_id: typing.Union[int, str]
    """Group ID"""

    sender_id: typing.Union[int, str]
    """Sender ID"""

    text_message: str
    """Message text"""

    message_event: platform_events.GroupMessage
    """Original message event object. Contains group and sender information."""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """Message chain"""

    # ========== Settable Attributes ==========

    user_message_alter: typing.Optional[provider_message.ContentElement] = None
    """Modified message text, langbot_plugin.api.entities.builtin.provider.message.ContentElement type"""

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """Direct reply message chain, only effective when preventing default behavior"""

class PersonCommandSent(BaseEventModel):
    """Triggered when a private chat command that should be processed is determined"""

    event_name: str = "PersonCommandSent"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    command: str
    """Command text"""

    params: list[str]
    """Command parameters"""

    text_message: str
    """Message text"""

    is_admin: bool
    """Whether it's an administrator"""


class GroupCommandSent(BaseEventModel):
    """Triggered when a group chat command that should be processed is determined"""

    event_name: str = "GroupCommandSent"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    command: str
    """Command text"""

    params: list[str]
    """Command parameters"""

    text_message: str
    """Message text"""

    is_admin: bool
    """Whether it's an administrator"""

class NormalMessageResponded(BaseEventModel):
    """Triggered when replying to a normal message"""

    event_name: str = "NormalMessageResponded"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    session: provider_session.Session
    """Session object"""

    prefix: str
    """Reply message prefix"""

    response_text: str
    """Reply message text"""

    finish_reason: str
    """Response end reason"""

    funcs_called: list[str]
    """List of called functions"""

    # ========== Settable Attributes ==========

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """Reply message component list, only effective when preventing default behavior"""

class PromptPreProcessing(BaseEventModel):
    """Triggered when preprocessing prompt in session"""

    event_name: str = "PromptPreProcessing"

    session_name: str
    """Session name, format is person_1234567890 or group_1234567890"""

    # ========== Settable Attributes ==========

    default_prompt: list[typing.Union[provider_message.Message, provider_message.MessageChunk]]
    """Scenario preset (system prompt) for this conversation, can be modified, langbot_plugin.api.entities.builtin.provider.message.Message or langbot_plugin.api.entities.builtin.provider.message.MessageChunk type"""

    prompt: list[typing.Union[provider_message.Message, provider_message.MessageChunk]]
    """Existing message records for this conversation, can be modified, langbot_plugin.api.entities.builtin.provider.message.Message or langbot_plugin.api.entities.builtin.provider.message.MessageChunk type"""

...
        @self.handler(events.PersonMessageReceived)
        async def handler(event_context: context.EventContext):
            ...

event = event_context.event
print(event.launcher_type)  # Launcher object type (group)
print(event.launcher_id)    # Group ID
print(event.sender_id)      # Sender account ID
print(event.message_chain)  # Message chain

def prevent_default(self):
    """Prevent default behavior"""

# Usage example
event_context.prevent_default()

def prevent_postorder(self):
    """Prevent postorder execution"""

# Usage example
event_context.prevent_postorder()