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

> Learn how to subscribe to custom events in AstrBot. Create a PipelineScheduler subclass and register it in EventBus to handle your custom events effectively.

- Repository: [AstrBot AI/AstrBot](https://github.com/AstrBotDevs/AstrBot)
- Tags: how-to-guide
- Published: 2026-03-12

---

**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.

```python
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.

```python
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`](https://github.com/AstrBotDevs/AstrBot/blob/main/astrbot/core/event_bus.py), the dispatch logic specifically looks up the scheduler using this ID:

```python

# 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:

```python
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:**

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

```

**Removing a subscriber:**

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

```

As implemented in [`astrbot/core/event_bus.py`](https://github.com/AstrBotDevs/AstrBot/blob/main/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`](https://github.com/AstrBotDevs/AstrBot/blob/main/astrbot/core/event_bus.py) | Core `EventBus` class with `dispatch()` loop and scheduler lookup logic | [View on GitHub](https://github.com/AstrBotDevs/AstrBot/blob/master/astrbot/core/event_bus.py) |
| [`astrbot/core/pipeline/scheduler.py`](https://github.com/AstrBotDevs/AstrBot/blob/main/astrbot/core/pipeline/scheduler.py) | Base `PipelineScheduler` class defining the `execute(event)` interface | [View on GitHub](https://github.com/AstrBotDevs/AstrBot/blob/master/astrbot/core/pipeline/scheduler.py) |
| [`astrbot/core/astrbot_config_mgr.py`](https://github.com/AstrBotDevs/AstrBot/blob/main/astrbot/core/astrbot_config_mgr.py) | Configuration manager handling `get_conf_info(unified_msg_origin)` resolution | [View on GitHub](https://github.com/AstrBotDevs/AstrBot/blob/master/astrbot/core/astrbot_config_mgr.py) |
| [`tests/unit/test_event_bus.py`](https://github.com/AstrBotDevs/AstrBot/blob/main/tests/unit/test_event_bus.py) | Unit tests demonstrating subscription, unsubscription, and error handling | [View on GitHub](https://github.com/AstrBotDevs/AstrBot/blob/master/tests/unit/test_event_bus.py) |

## 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`](https://github.com/AstrBotDevs/AstrBot/blob/main/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.