How to Subscribe to Custom Events in AstrBot: A Complete Guide

To subscribe to custom events in AstrBot, create a PipelineScheduler subclass with an async execute(event) method and register it in the EventBus.pipeline_scheduler_mapping dictionary under the configuration ID that matches your event's unified_msg_origin.

AstrBot is an open-source multi-platform chatbot framework that routes messages through an asynchronous event bus. Understanding how to subscribe to custom events in AstrBot allows you to build dedicated handlers for specific platforms or custom message sources without modifying core framework code.

Understanding the AstrBot Event Bus Architecture

AstrBot implements an asynchronous event bus (astrbot.core.event_bus.EventBus) that acts as the central message router. Every incoming message from any platform—whether Telegram, Discord, Misskey, or custom sources—becomes an AstrMessageEvent that flows through this architecture.

The subscription model relies on three core components:

• PipelineScheduler – The base class that all subscribers must extend. It requires a single async execute(event) method that processes incoming events.
• pipeline_scheduler_mapping – A dictionary that maps configuration IDs to scheduler instances. The EventBus uses this to route events to the correct handler.
• AstrBotConfigManager – Resolves an event's unified_msg_origin (e.g., custom:group:123) to a configuration record containing the target ID.

When EventBus.dispatch() pulls an event from the queue, it calls get_conf_info(event.unified_msg_origin) to determine which scheduler should handle the event, then executes it as an independent asyncio task.

How to Subscribe to Custom Events in AstrBot

Step 1: Create a Custom PipelineScheduler

First, subclass PipelineScheduler and implement the execute method. This method receives an AstrMessageEvent object containing all message metadata.

from astrbot.core.pipeline.scheduler import PipelineScheduler

class CustomEventScheduler(PipelineScheduler):
    async def execute(self, event):
        # Your custom processing logic here

        print(f"Received event from {event.get_sender_name()}: {event.get_message_outline()}")
        # Return True to continue pipeline, False to stop

        return True

Step 2: Register Your Subscriber in the Mapping

Instantiate your scheduler and add it to the pipeline_scheduler_mapping dictionary under the configuration ID you want to listen for.

from astrbot.core.event_bus import EventBus

# Create mapping with your custom scheduler

pipeline_mapping = {
    "my-custom-conf": CustomEventScheduler(),  # Key matches config ID

}

# Initialize EventBus with your mapping

event_bus = EventBus(
    event_queue=asyncio.Queue(),
    pipeline_scheduler_mapping=pipeline_mapping,
    astrbot_config_mgr=config_manager  # Your config manager instance

)

Step 3: Ensure Configuration ID Resolution

Your AstrBotConfigManager must return a configuration record with an id field that matches your mapping key. When an event arrives with unified_msg_origin="custom:group:123", the config manager should resolve this to {"id": "my-custom-conf", ...}.

In astrbot/core/event_bus.py, the dispatch logic specifically looks up the scheduler using this ID:

# From astrbot/core/event_bus.py lines 44-49

conf_info = self.astrbot_config_mgr.get_conf_info(event.unified_msg_origin)
scheduler = self.pipeline_scheduler_mapping.get(conf_info["id"])
if scheduler is None:
    logger.error(f"No scheduler found for config ID: {conf_info['id']}")
    continue

Complete Working Example

Here is a runnable example demonstrating the full subscription flow, including a dummy config manager and event simulation:

import asyncio
from astrbot.core.event_bus import EventBus
from astrbot.core.pipeline.scheduler import PipelineScheduler

# 1. Define custom scheduler

class MyCustomScheduler(PipelineScheduler):
    async def execute(self, event):
        print(f"🔔 Custom handler received: {event.get_message_outline()}")
        return True

# 2. Create dummy config manager

class DummyConfigMgr:
    def get_conf_info(self, origin):
        # Route any "custom:" origin to our custom config

        if origin.startswith("custom:"):
            return {"id": "my-custom-conf", "name": "Custom Source"}
        return {"id": "default", "name": "Default"}

# 3. Setup infrastructure

event_queue = asyncio.Queue()
pipeline_mapping = {
    "my-custom-conf": MyCustomScheduler(),
}
config_mgr = DummyConfigMgr()

bus = EventBus(
    event_queue=event_queue,
    pipeline_scheduler_mapping=pipeline_mapping,
    astrbot_config_mgr=config_mgr,
)

# 4. Simulate a custom event

class DummyEvent:
    unified_msg_origin = "custom:group:123"
    def get_platform_id(self): return "custom"
    def get_platform_name(self): return "Custom"
    def get_sender_name(self): return "Alice"
    def get_sender_id(self): return "alice42"
    def get_message_outline(self): return "Hello from custom source"

async def main():
    # Start event bus

    asyncio.create_task(bus.dispatch())
    
    # Push event

    await event_queue.put(DummyEvent())
    await asyncio.sleep(0.1)  # Let handler process

asyncio.run(main())

Managing Subscriptions at Runtime

The pipeline_scheduler_mapping is a standard Python dictionary, allowing dynamic subscription management without restarting the bot.

Adding a subscriber at runtime:

new_scheduler = MyCustomScheduler()
bus.pipeline_scheduler_mapping["new-conf-id"] = new_scheduler

Removing a subscriber:

del bus.pipeline_scheduler_mapping["new-conf-id"]

As implemented in astrbot/core/event_bus.py, the dispatch loop checks the mapping for every event, so changes take effect immediately for subsequent events.

Key Source Files and Implementation Details

File	Purpose	Location
astrbot/core/event_bus.py	Core EventBus class with dispatch() loop and scheduler lookup logic	View on GitHub
astrbot/core/pipeline/scheduler.py	Base PipelineScheduler class defining the execute(event) interface	View on GitHub
astrbot/core/astrbot_config_mgr.py	Configuration manager handling get_conf_info(unified_msg_origin) resolution	View on GitHub
tests/unit/test_event_bus.py	Unit tests demonstrating subscription, unsubscription, and error handling	View on GitHub

Summary

• AstrBot routes events through EventBus, which pulls AstrMessageEvent objects from an asyncio.Queue and dispatches them to registered schedulers.
• To subscribe to custom events in AstrBot, extend PipelineScheduler, implement async execute(event), and register the instance in EventBus.pipeline_scheduler_mapping under the configuration ID that matches your event's unified_msg_origin.
• Configuration resolution happens via AstrBotConfigManager.get_conf_info(), which maps message origins to config IDs used for scheduler lookup.
• Dynamic subscription management is supported by modifying the pipeline_scheduler_mapping dictionary at runtime; additions and removals take effect immediately.

Frequently Asked Questions

How does AstrBot determine which scheduler receives a custom event?

AstrBot determines the target scheduler by resolving the event's unified_msg_origin string through AstrBotConfigManager.get_conf_info(). This method returns a configuration record containing an id field. The EventBus then uses this ID as a key to look up the corresponding PipelineScheduler in the pipeline_scheduler_mapping dictionary. If no matching entry exists, the event is logged as an error and dropped.

Can I have multiple schedulers subscribed to the same configuration ID?

No, the current implementation in astrbot/core/event_bus.py uses a one-to-one mapping where each configuration ID maps to a single PipelineScheduler instance. If you need multiple processing stages for the same event type, you should implement a single scheduler that chains multiple handlers internally, or subclass PipelineScheduler to compose multiple operations within its execute() method.

Is it possible to subscribe or unsubscribe from events while the bot is running?

Yes, because pipeline_scheduler_mapping is a standard Python dictionary attached to the EventBus instance, you can modify it at runtime. To subscribe a new handler, assign your scheduler to a configuration ID: bus.pipeline_scheduler_mapping["new-id"] = scheduler. To unsubscribe, delete the key: del bus.pipeline_scheduler_mapping["new-id"]. The dispatch loop checks this mapping for every event, so changes apply immediately to subsequent events without requiring a restart.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}