Event System API

This document provides a detailed introduction to the ErisPulse event system API.

The event system categorizes platform events and distributes them to five types of handlers:

flowchart LR
    A["Platform Event<br/> (OneBot12 Standard)"] --> B{"Event Type"}
    B --> C["command<br/>Command Handler"]
    B --> D["message<br/>Message Handler"]
    B --> E["notice<br/>Notice Handler"]
    B --> F["request<br/>Request Handler"]
    B --> G["meta<br/>Meta Event Handler"]
    C & D & E & F & G --> H["Event Wrapper Class<br/>reply / get_text / done, etc."]

Command Module

Registering Commands

from ErisPulse.Core.Event import command

# Basic command
@command("hello", help="Send greeting")
async def hello_handler(event):
    await event.reply("Hello!")

# Command with aliases
@command(["help", "h"], aliases=["help"], help="Show help")
async def help_handler(event):
    pass

# Command with permission
def is_admin(event):
    return event.get("user_id") in admin_ids

@command("admin", permission=is_admin, help="Admin command")
async def admin_handler(event):
    pass

# Hidden command
@command("secret", hidden=True, help="Secret command")
async def secret_handler(event):
    pass

# Command group
@command("admin.reload", group="admin", help="Reload module")
async def reload_handler(event):
    pass

# Subcommands (multi-token command names separated by spaces)
# Matching uses longest prefix: /admin add x matches admin add (args is ["x"]);
# Subcommands inherit permission from the nearest ancestor command if permission not declared
@command("admin add", help="Add admin")
async def admin_add_handler(event):
    pass

Naming Conflict Rules: Command names take precedence over aliases. When registering an alias that conflicts with an existing command, the alias is ignored and a warning is issued; when registering a command that conflicts with an existing alias, the command takes precedence and the conflicting alias is removed—both types of conflicts generate WARNING logs and do not silently hijack.

Command Information

All command query APIs support optional session context: pass event= (Event or dict) or explicitly platform= / bot_id= / session_id= (explicit parameters take precedence when overlapped with event), i.e., filter commands unavailable in the current session by scope module dimension (see advanced/scope.md); all are optional keyword arguments, and behavior remains unchanged if none are provided.

# Get command help
help_text = command.help()

# Session-aware help: only list commands available in the current session
help_text = command.help(event=event)

# Get specific command (returns merged and overridden effective parameters; returns None if session unavailable)
cmd_info = command.get_command("admin")
cmd_info = command.get_command("admin", event=event)

# Get all commands (filters unavailable module commands in session-aware mode)
all_commands = command.get_commands()
all_commands = command.get_commands(event=event)

# Get all commands in a command group (supports session-aware filtering)
admin_commands = command.get_group_commands("admin")
admin_commands = command.get_group_commands("admin", event=event)

# Get all visible commands
visible_commands = command.get_visible_commands()

# Session-aware visible commands (either event or explicit keyword argument is sufficient)
visible_commands = command.get_visible_commands(event=event)
visible_commands = command.get_visible_commands(
    platform=event.get("platform"),
    bot_id=event.get_self_account_id(),
    session_id=event.get_session_id(),
)

Waiting for Replies

# Wait for user reply
@command("ask", help="Ask user information")
async def ask_command(event):
    reply = await command.wait_reply(
        event,
        prompt="Please enter your name:",  # Already sent above
        timeout=30.0
    )
    
    if reply:
        name = reply.get_text()
        await event.reply(f"Hello, {name}!")

# Wait for reply with validation
def validate_age(event_data):
    try:
        age = int(event_data.get_text())
        return 0 <= age <= 150
    except ValueError:
        return False

@command("age", help="Ask user age")
async def age_command(event):
    await event.reply("Please enter your age:")
    
    reply = await command.wait_reply(
        event,
        timeout=60,
        validator=validate_age
    )
    
    if reply:
        age = int(reply.get_text())
        await event.reply(f"Your age is {age} years old.")

# Wait for reply with callback
async def handle_confirmation(reply_event):
    text = reply_event.get_text().lower()
    if text in ["是", "yes", "y"]:
        await event.reply("Operation confirmed!")
    else:
        await event.reply("Operation cancelled.")

@command("confirm", help="Confirm operation")
async def confirm_command(event):
    await command.wait_reply(
        event,
        prompt="Please enter '是' or '否':",
        callback=handle_confirmation
    )

Message Module

Message Events

from ErisPulse.Core.Event import message

# Listen to all messages
@message.on_message()
async def message_handler(event):
    sdk.logger.info(f"Received message: {event.get_text()}")

# Listen to private messages
@message.on_private_message()
async def private_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"Private message from: {user_id}")

# Listen to group messages
@message.on_group_message()
async def group_handler(event):
    group_id = event.get_group_id()
    sdk.logger.info(f"Group message from: {group_id}")

# Listen to @messages
@message.on_at_message()
async def at_handler(event):
    mentions = event.get_mentions()
    sdk.logger.info(f"Users mentioned: {mentions}")

Conditional Listening

# Use priority to control execution order
@message.on_message(priority=10)  # Higher value means higher priority
async def high_priority_handler(event):
    pass

# Implement conditional filtering inside the handler
@message.on_message()
async def filtered_handler(event):
    if "keyword" not in event.get_text():
        return
    # Process messages containing the keyword
    pass

Notice Module

Notice Events

from ErisPulse.Core.Event import notice

# Friend added
@notice.on_friend_add()
async def friend_add_handler(event):
    user_id = event.get_user_id()
    await event.reply("Welcome to add me as a friend!")

# Friend removed
@notice.on_friend_remove()
async def friend_remove_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"Friend removed: {user_id}")

# Group member increased
@notice.on_group_increase()
async def member_increase_handler(event):
    user_id = event.get_user_id()
    await event.reply("Welcome new member!")

# Group member decreased
@notice.on_group_decrease()
async def member_decrease_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"Group member left: {user_id}")

Request Module

Request Events

from ErisPulse.Core.Event import request

# Friend request
@request.on_friend_request()
async def friend_request_handler(event):
    user_id = event.get_user_id()
    comment = event.get_comment()
    sdk.logger.info(f"Friend request: {user_id}, comment: {comment}")

# Group invitation request
@request.on_group_request()
async def group_request_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    sdk.logger.info(f"Group invitation: {group_id}, from: {user_id}")

Meta Module

Meta Events

from ErisPulse.Core.Event import meta

# Connection event
@meta.on_connect()
async def connect_handler(event):
    platform = event.get_platform()
    sdk.logger.info(f"Platform {platform} connected successfully")

# Disconnection event
@meta.on_disconnect()
async def disconnect_handler(event):
    platform = event.get_platform()
    sdk.logger.info(f"Platform {platform} disconnected")

# Heartbeat event
@meta.on_heartbeat()
async def heartbeat_handler(event):
    sdk.logger.debug("Received heartbeat")

Bot Status Query

After the adapter sends a meta event, the framework automatically tracks the Bot status. For query APIs and lifecycle event listeners, refer to Adapter System API - Bot Status Management.

Event Wrapper Class

Event module event handlers receive an Event wrapper class instance, which inherits from dict and provides convenient methods.

Core Methods

# Get event information
event_id = event.get_id()
event_time = event.get_time()
event_type = event.get_type()
detail_type = event.get_detail_type()
platform = event.get_platform()

# Get bot information
self_platform = event.get_self_platform()
self_user_id = event.get_self_user_id()
self_info = event.get_self_info()

Session Identifiers

# Unified target ID: returns group_id for group chats, user_id for private chats, etc.
target_id = event.get_target_id()

# Session unique identifier, format: {platform}:{detail_type}:{target_id}
session_id = event.get_session_id()
# Example: "telegram:private:12345", "qq:group:67890"

get_target_id() returns the first non-empty value in the following order: group_id → channel_id → guild_id → thread_id → user_id. This is suitable for scenarios such as context management and state storage that require a unified session identifier.

Message Methods

# Get message content
message_segments = event.get_message()
alt_message = event.get_alt_message()
text = event.get_text()

# Get sender information
user_id = event.get_user_id()
nickname = event.get_user_nickname()
sender = event.get_sender()

# Get group information
group_id = event.get_group_id()

# Check message type
is_msg = event.is_message()
is_private = event.is_private_message()
is_group = event.is_group_message()

# @message related
is_at = event.is_at_message()
has_mention = event.has_mention()
mentions = event.get_mentions()

Command Information

# Get command information
cmd_name = event.get_command_name()
cmd_args = event.get_command_args()
cmd_raw = event.get_command_raw()

# Check if it's a command
is_cmd = event.is_command()

Reply Functionality

# Basic reply
await event.reply("This is a message")

# Specify sending method
await event.reply("http://example.com/image.jpg", method="Image")

# Reply with @users and reply to message
await event.reply("Hello", at_users=["user1"], reply_to="msg_id")

# @all members
await event.reply("Announcement", at_all=True)

# Use platform-specific modifier methods (via parameter)
await event.reply("Board content", method="Board",
                  via=[("Expire", 3600), ("ForMember", "114514")])

# Get send chain, freely append modifier methods and sending methods (suitable for multiple consecutive modifiers/action methods)
await event.send_chain().Expire(3600).Board("Board content")
await event.send_chain().DismissBoard()

# Reply using OneBot12 message segments
from ErisPulse.Core.Event import MessageBuilder
msg = MessageBuilder().text("Hello").image("url").build()
await event.reply_ob12(msg)

# Wait for reply
reply = await event.wait_reply(timeout=30)

Platform Capability Query

# Check if current platform supports a certain sending method
if event.supports("Image"):
    await event.reply(url, method="Image")

# List all available sending methods for current platform
methods = event.available_methods()
# ["Text", "Image", "Voice", ...]

Reply Methods

The reply() method supports specifying the sending type via the method parameter, and two convenient boolean parameters:

# Simple text reply
await event.reply("Hello")

# Reply and @ sender (automatically extract user_id)
await event.reply("Hello", at_sender=True)

# Reply and quote current message
await event.reply("Received", quote=True)

# Combine usage
await event.reply("Received", at_sender=True, quote=True)

# Send image (using method parameter)
if event.supports("Image"):
    await event.reply("http://example.com/img.jpg", method="Image")
else:
    await event.reply("[Image] http://example.com/img.jpg")

Parameter Description:

Parameter	Type	Description
`content`	str	Content to send
`method`	str	Sending method, default "Text", optional "Image"/"Voice"/"Video"/"File" etc.
`at_sender`	bool	Whether to @ sender (automatically extract user_id)
`quote`	bool	Whether to quote and reply to current message (automatically extract message_id)
`at_users`	list[str]	List of users to @
`reply_to`	str	Manually specify the message ID to reply to
`at_all`	bool	Whether to @ all members

Interaction Methods

# confirm — confirmation dialog (returns True/False/None)
if await event.confirm("Are you sure you want to execute this operation?"):
    await event.reply("Confirmed")

# Use non-Text method to send confirmation prompt
if await event.confirm("http://example.com/image.jpg", method="Image"):
    await event.reply("Image prompt confirmed")

# choose — selection menu (returns option index or None)
choice = await event.choose("Please select color:", ["red", "green", "blue"])

# options_format="auto" (default) automatically selects style based on method:
# Markdown→unordered list (- 1. option), Html→ordered list (<ol>), others→plain text list
# Text-based methods (Markdown/Html etc.) default merge options to the end
# merge_prompt=True forces merging for any method; placeholder can customize placeholder
choice = await event.choose(
    "## Please select\n{options}", ["A", "B"],
    method="Markdown", merge_prompt=True,
)

# collect — form collection (returns {key: value} dict or None)
data = await event.collect([
    {"key": "name", "prompt": "Please enter name:"},
    {"key": "age", "prompt": "Please enter age:",
     "validator": lambda e: e.get_text().isdigit()},
    {"key": "avatar", "prompt": "Please send avatar:", "method": "Image"},
])

# wait_for — wait for any event satisfying condition
evt = await event.wait_for(event_type="notice", condition=lambda e: ..., timeout=120)

# conversation — multi-turn conversation context
conv = event.conversation(timeout=60)
await conv.say("Welcome!")

For complete parameter descriptions and more examples of interaction methods, refer to Event Wrapper Class Details and Conversation Multi-Turn Dialog.

Utility Methods

# Convert to dictionary (filter keys starting with _)
event_dict = event.to_dict()

# Get raw data
raw = event.get_raw()
raw_type = event.get_raw_type()

Chain Control

event.done(claim=, stop=) uniformly controls the two orthogonal semantics of "claiming" and "blocking":

Claiming (claim): Marks the event as processed (_processed), the command dispatcher skips it to prevent duplication
Blocking (stop): Prevents propagation to lower-priority handlers (_propagation_stopped)

# Claim + Block (default)
event.done()

# Claim only, do not block (lower-priority observers still see)
event.done(stop=False)

# Block only, do not claim (e.g., firewall / rate limiting)
event.done(claim=False)

# mark_processed is the main method, done is its alias
event.mark_processed()             # equivalent to event.done()
event.mark_processed(stop=False)   # equivalent to event.done(stop=False)

# Query status
event.is_processed()  # whether claimed
event.is_stopped()    # whether propagation blocked

Platform Extension Methods

Adapters can register platform-specific methods for Event, available only on instances of the corresponding platform.

User: Using Platform Extension Methods

After an adapter registers platform-specific methods, you can directly call them in event handlers. The methods vary by platform; refer to the corresponding platform documentation.

from ErisPulse.Core.Event import message

@message.on_message()
async def handle_message(event):
    platform = event.get_platform()

    # Call platform-specific methods based on platform
    if platform == "email":
        subject = event.get_subject()           # Email-specific
        attachments = event.get_attachments()   # Email-specific

Querying Registered Platform Methods

from ErisPulse.Core.Event import get_platform_event_methods

# View which methods are registered for a platform
methods = get_platform_event_methods("email")
# ["get_subject", "get_from", "get_attachments", ...]

# Dynamically check and call
for method_name in get_platform_event_methods(event.get_platform()):
    method = getattr(event, method_name)
    print(f"{method_name}: {method()}")

Platform Method Isolation

Methods registered for different platforms do not interfere with each other:

# Email event - only email methods
event = Event({"platform": "email", "email_raw": {"subject": "Hello"}})
event.get_subject()      # ✅ "Hello"
event.get_chat_type()    # ❌ AttributeError

# Telegram event - only Telegram methods
event = Event({"platform": "telegram", "telegram_raw": {"chat": {"type": "private"}}})
event.get_chat_type()    # ✅ "private"
event.get_subject()      # ❌ AttributeError

`hasattr` / `dir` Support

hasattr(event, "get_subject")   # True only when platform="email"
"get_subject" in dir(event)     # Same as above

Adapter: Registering Platform Extension Methods

Adapters can register platform-specific methods for Event via decorators, where the first parameter of the method is self (Event instance), allowing free access to event data.

Single Method Registration

from ErisPulse.Core.Event import register_event_method

@register_event_method("email")
def get_subject(self):
    """Get email subject"""
    return self.get("email_raw", {}).get("subject", "")

@register_event_method("email")
def get_from(self):
    """Get sender"""
    return self.get("email_raw", {}).get("from", {})

Batch Registration (Mixin Class)

When there are many methods, it is recommended to use a Mixin class for batch registration:

from ErisPulse.Core.Event import register_event_mixin

class EmailEventMixin:
    def get_subject(self):
        return self.get("email_raw", {}).get("subject", "")

    def get_from(self):
        return self.get("email_raw", {}).get("from", {})

    def get_attachments(self):
        return self.get("email_raw", {}).get("attachments", [])

# Register all methods at once
register_event_mixin("email", EmailEventMixin)

Return Value Specification

Scenario	Return Value	User Usage
Return data (text, dict, etc.)	Return the value directly	`subject = event.get_subject()`
Execute operation (send message, etc.)	Return `asyncio.Task`	`task = event.do_something()` (optional await)

Recommendation: Methods that do not return data should return asyncio.Task, so users can decide whether to await, even if not awaited, the operation will complete.

@register_event_method("email")
def forward_email(self, to_address: str):
    """Forward email — returns Task, user can decide whether to await"""
    import asyncio
    return asyncio.create_task(
        self._do_forward(to_address)
    )

# User can await to wait for result
await event.forward_email("[email protected]")

# Or not await, operation executes in background
event.forward_email("[email protected]")

Unregistering Methods

from ErisPulse.Core.Event import unregister_event_method, unregister_platform_event_methods

# Unregister a single method
unregister_event_method("email", "get_subject")

# Unregister all methods for a platform (called during adapter shutdown)
unregister_platform_event_methods("email")

Overriding Built-in Methods

register_event_mixin / register_event_method supports overriding Event built-in methods (such as confirm, choose, collect, wait_reply, reply, etc.). The registered platform methods take precedence over built-in methods via Event.__getattribute__, so adapters can provide platform-specific interactive implementations.

Built-in implementations are exported as _builtin_* functions, and the overwriting side can call them as fallback:

from ErisPulse.Core.Event import register_event_mixin, _builtin_choose

class YunhuEventMixin:
    async def choose(self, prompt, options, timeout=60, method="Text"):
        # Yunhu platform uses button components
        buttons = [[{"text": opt} for opt in options]]
        await self.reply(prompt)
        # ...wait for button callback or text reply...
        # fallback to built-in logic
        return await _builtin_choose(self, None, options, timeout, "Text")

register_event_mixin("yunhu", YunhuEventMixin)

Cross-Platform Extension (Wildcard)

register_event_method and register_event_mixin support passing "*" as the platform name, registering methods available on Event instances for all platforms. Suitable for cross-platform reusable features such as AI chat and context management.

Registering Cross-Platform Methods

from ErisPulse.Core.Event.wrapper import register_event_method

@register_event_method("*")
async def ai_chat(self, prompt: str):
    """self is Event instance, can freely access event data and built-in methods"""
    await self.reply(f"AI: {prompt}")

After registration, all platform event handlers can call:

from ErisPulse.Core.Event import message

@message.on_message()
async def handler(event):
    await event.ai_chat(event.get_text())

Method Resolution Priority

When accessing Event methods via attribute, the resolution order is:

Platform-specific methods (current platform's override)
Wildcard methods ("*" registered cross-platform methods)
Built-in methods (reply, confirm, choose, collect, wait_reply, etc.)
Dictionary key access

Thus, wildcard methods can override built-in methods (such as reply), but will be further overridden by same-named platform-specific methods.

Priority System

Event handlers support priority, with higher values indicating higher priority:

# High-priority handler executes first
@message.on_message(priority=10)
async def high_priority_handler(event):
    pass

# Low-priority handler executes later
@message.on_message(priority=0)
async def low_priority_handler(event):
    pass

Core Module API - Core module API
Adapter System API - Adapter management API
Module Development Guide - Develop custom modules

Event System API

Command Module

Registering Commands

Command Information

Waiting for Replies

Message Module

Message Events

Conditional Listening

Notice Module

Notice Events

Request Module

Request Events

Meta Module

Meta Events

Bot Status Query

Event Wrapper Class

Core Methods

Session Identifiers

Message Methods

Command Information

Reply Functionality

Platform Capability Query

Reply Methods

Interaction Methods

Utility Methods

Chain Control

Platform Extension Methods

User: Using Platform Extension Methods

Querying Registered Platform Methods

Platform Method Isolation

hasattr / dir Support

Adapter: Registering Platform Extension Methods

Single Method Registration

Batch Registration (Mixin Class)

Return Value Specification

Unregistering Methods

Overriding Built-in Methods

Cross-Platform Extension (Wildcard)

Registering Cross-Platform Methods

Method Resolution Priority

Priority System

Related Documentation

`hasattr` / `dir` Support