Lifecycle
MaiBot plugins have three lifecycle methods: on_load(), on_unload(), and on_config_update(). The SDK enforces that all plugins implement these three methods; otherwise, the Runner will refuse to load the plugin.
create_plugin() Factory Function
Each plugin's plugin.py must export a top-level create_plugin() function that returns the plugin instance:
from maibot_sdk import MaiBotPlugin
class MyPlugin(MaiBotPlugin):
async def on_load(self) -> None:
...
async def on_unload(self) -> None:
...
async def on_config_update(self, scope: str, config_data: dict, version: str) -> None:
...
def create_plugin():
return MyPlugin()When the Runner loads a plugin:
- Import the
plugin.pymodule - Call
create_plugin()to obtain the plugin instance - Inject
PluginContext(at this pointself.ctxis available) - Call
on_load()
on_load()
Callback after the plugin has finished loading. The Runner calls this method after injecting the PluginContext and completing capability bootstrap, so all capability proxies of self.ctx can be used directly within on_load().
async def on_load(self) -> None:
"""Called after plugin loaded. Initialize resources here.
Runner has already injected PluginContext before calling this,
so self.ctx is available.
"""Typical use cases:
- Initialize internal plugin state
- Call
self.ctx.gateway.update_state()to report the message gateway status - Call
self.register_dynamic_api()to register dynamic APIs andawait self.sync_dynamic_apis() - Read configuration and initialize resources
Example:
from maibot_sdk import MaiBotPlugin, PluginConfigBase, Field
class MyConfig(PluginConfigBase):
greeting: str = Field(default="Hello!", description="Default greeting")
class MyPlugin(MaiBotPlugin):
config_model = MyConfig
async def on_load(self) -> None:
# self.ctx has already been injected and can be used directly
self.ctx.logger.info("Plugin loaded, current greeting: %s", self.config.greeting)
# Dynamic APIs can be registered here
self.register_dynamic_api(
"my_api",
self._handle_api,
description="Example API",
version="1",
public=True,
)
await self.sync_dynamic_apis()
async def _handle_api(self, **kwargs):
return {"status": "ok"}on_unload()
Callback before plugin unload. Release all resources held by the plugin in this method.
async def on_unload(self) -> None:
"""Called before plugin unloaded. Cleanup resources."""Typical use cases:
- Close network connections, file handles
- Report gateway offline status (
self.ctx.gateway.update_state(..., ready=False)) - Unregister dynamic APIs
- Save persistent data
Example:
class MyPlugin(MaiBotPlugin):
async def on_unload(self) -> None:
self.ctx.logger.info("Plugin is unloading")
# Report message gateway offline
await self.ctx.gateway.update_state(
gateway_name="my_gateway",
ready=False,
)
# Clear dynamic APIs
self.clear_dynamic_apis()
await self.sync_dynamic_apis(offline_reason="Plugin has been unloaded")Note
self.ctx can still be used within on_unload(), but cleanup should be completed as quickly as possible. Do not perform time-consuming operations.
on_config_update()
Configuration hot-reload callback. The Runner calls this method when the plugin configuration or any subscribed global configuration changes.
async def on_config_update(
self,
scope: str,
config_data: dict[str, Any],
version: str,
) -> None:
"""Called when config hot-reloads.
Args:
scope: Configuration change scope, with possible values "self", "bot", or "model".
config_data: Latest configuration data corresponding to the current scope.
version: Configuration version number.
"""scope values
"self"→CONFIG_RELOAD_SCOPE_SELF— Plugin's own configuration. Always triggered whenconfig.tomlin the plugin directory changes; no subscription required."bot"→ON_BOT_CONFIG_RELOAD— Global Bot configuration. Requires subscription viaconfig_reload_subscriptions."model"→ON_MODEL_CONFIG_RELOAD— LLM model configuration. Requires subscription viaconfig_reload_subscriptions.
INFO
- The callback for
scope == "self"is always triggered and does not require additional subscription. scope == "bot"andscope == "model"are triggered only if declared inconfig_reload_subscriptions.
Example
from maibot_sdk import MaiBotPlugin, CONFIG_RELOAD_SCOPE_SELF, ON_BOT_CONFIG_RELOAD, ON_MODEL_CONFIG_RELOAD
from typing import ClassVar, Iterable
class MyPlugin(MaiBotPlugin):
# Subscribe to hot-reloads for both bot and model global configurations
config_reload_subscriptions: ClassVar[Iterable[str]] = ("bot", "model")
async def on_load(self) -> None:
self.ctx.logger.info("Plugin loaded")
async def on_unload(self) -> None:
self.ctx.logger.info("Plugin unloaded")
async def on_config_update(self, scope: str, config_data: dict, version: str) -> None:
if scope == CONFIG_RELOAD_SCOPE_SELF:
# Plugin's own configuration changed; self.config is automatically updated
self.ctx.logger.info("Plugin configuration updated: version=%s", version)
elif scope == ON_BOT_CONFIG_RELOAD:
# Global Bot configuration changed
bot_name = config_data.get("bot_name", "Unknown")
self.ctx.logger.info("Bot configuration updated: bot_name=%s, version=%s", bot_name, version)
elif scope == ON_MODEL_CONFIG_RELOAD:
# LLM model configuration changed
model_name = config_data.get("model_name", "Unknown")
self.ctx.logger.info("Model configuration updated: model=%s, version=%s", model_name, version)config_reload_subscriptions
Class variable declaring the global configuration hot-reload scopes that the plugin needs to subscribe to. Only supports the values "bot" and "model":
from typing import ClassVar, Iterable
class MyPlugin(MaiBotPlugin):
# Subscribe to both global configurations
config_reload_subscriptions: ClassVar[Iterable[str]] = ("bot", "model")
# Subscribe only to Bot configuration
# config_reload_subscriptions: ClassVar[Iterable[str]] = ("bot",)
# Subscribe only to Model configuration
# config_reload_subscriptions: ClassVar[Iterable[str]] = ("model",)
# Do not subscribe to any global configuration (default value)
# config_reload_subscriptions: ClassVar[Iterable[str]] = ()Rules:
- The default value is an empty tuple
(), meaning no global configurations are subscribed to. - The
"self"scope always triggers the callback and does not need (nor can it) be declared here. - Only
"bot"and"model"are valid subscription values. - Declaring unsupported values will raise a
ValueErroringet_config_reload_subscriptions(). - You cannot pass a string directly (e.g.,
config_reload_subscriptions = "bot"); an iterable collection must be used.
Since 1.3.2: saving configuration no longer blocks on the broadcast
Config hot-reload broadcasts now run in a background task and are coalesced before delivery: when you change configuration several times in quick succession, the plugin receives only one merged latest snapshot, and the WebUI save request no longer waits for the broadcast to finish, so saving is noticeably faster. Callback semantics are unchanged — on_config_update() still receives the complete latest configuration for that scope.
Complete Lifecycle Example
Below is a complete plugin example that includes all lifecycle methods:
from typing import Any, ClassVar, Iterable
from maibot_sdk import (
CONFIG_RELOAD_SCOPE_SELF,
Command,
MaiBotPlugin,
ON_BOT_CONFIG_RELOAD,
ON_MODEL_CONFIG_RELOAD,
Tool,
)
from maibot_sdk.types import ToolParameterInfo, ToolParamType
class GreeterPlugin(MaiBotPlugin):
"""Greeting Plugin — Demonstrates the complete plugin lifecycle."""
# Subscribe to global configuration hot reload
config_reload_subscriptions: ClassVar[Iterable[str]] = ("bot", "model")
async def on_load(self) -> None:
"""Initialize when the plugin is loaded."""
self.ctx.logger.info("GreeterPlugin loaded")
# self.ctx is already available here; you can directly call capability proxies
raw_config = self.get_plugin_config_data()
self.ctx.logger.info("Current config: %s", raw_config)
async def on_unload(self) -> None:
"""Clean up resources when the plugin is unloaded."""
self.ctx.logger.info("GreeterPlugin is unloading")
async def on_config_update(self, scope: str, config_data: dict[str, Any], version: str) -> None:
"""Handle configuration hot update."""
if scope == CONFIG_RELOAD_SCOPE_SELF:
self.ctx.logger.info("Plugin config updated: version=%s", version)
elif scope == ON_BOT_CONFIG_RELOAD:
self.ctx.logger.info("Bot config updated: version=%s", version)
elif scope == ON_MODEL_CONFIG_RELOAD:
self.ctx.logger.info("Model config updated: version=%s", version)
@Tool(
"greet",
brief_description="Greet the user",
detailed_description="Parameter description:\n- stream_id: string, required. The current chat stream ID.",
parameters=[
ToolParameterInfo(
name="stream_id",
param_type=ToolParamType.STRING,
description="Current chat stream ID",
required=True,
),
],
)
async def handle_greet(self, stream_id: str, **kwargs):
await self.ctx.send.text("Hello!", stream_id)
return {"success": True, "message": "Replied"}
@Command("hello", pattern=r"^/hello")
async def handle_hello(self, **kwargs):
await self.ctx.send.text("Hello!", kwargs["stream_id"])
return True, "Hello!", 2
def create_plugin():
return GreeterPlugin()Lifecycle Sequence
Verify and Troubleshoot
Verification — reload the plugin and read the log first: the line your on_load() prints appears; then change one field on the WebUI plugin config page and save, and the scope="self" record from your on_config_update() shows up right after — all three lifecycle methods, their signatures, and the injected ctx are verified.
- Loading is refused with "must implement on_load()" /
on_unload()/on_config_update()— before registration the Runner checks each method against the base class: a plugin that only inheritsMaiBotPluginwithout overriding one, or that defines it on another class, hits this. Define all three on the plugin class itself. - Saving configuration fails with a
TypeErrorin the Runner log — the Runner always callson_config_update()with the three positional arguments(scope, config_data, version); a signature with fewer parameters, a different order, or one that does not accept them fails. Copyasync def on_config_update(self, scope: str, config_data: dict, version: str) -> None. - Loading fails on the subscription declaration —
config_reload_subscriptionsmust be an iterable collection: assigning a plain string (= "bot") raisesTypeError, while"self"or any unsupported value raisesValueError(only"bot"and"model"are valid). - A global configuration change never reaches your callback — only
scope="self"always fires;"bot"/"model"fire only when declared. Declare the class variable on the plugin class (ClassVar[Iterable[str]]); an instance attribute is ignored. - Reading
self.ctxraises aRuntimeErrorabout the plugin context not being initialized —self.ctxandself.configexist only aftercreate_plugin()returned the instance and the Runner injectedPluginContext; reading them at module level or in__init__()fails. Move that initialization intoon_load().