How to Create Custom Plugins and Skills in AstrBot
Purpose
After setting up AstrBot, I wanted to extend it with my own features. The problem: I couldn’t find clear documentation on how the plugin system actually works. I struggled with understanding the event flow, the filter decorators, and especially how to integrate LLM calls into my plugins.
This post demonstrates how to build AstrBot plugins from scratch, covering the Star plugin system, event handling, message responses, and LLM-powered agent skills.
Environment
- AstrBot 3.5+
- Python 3.10+
- Basic understanding of async/await
The Star Plugin System
AstrBot uses what it calls “Star” plugins - loosely coupled, async-driven components that hook into the message pipeline. When a message arrives, it flows through registered plugins via event filters.
The architecture looks like this:
[Message Event] --> [Event Bus] --> [Filter] --> [Plugin Handler] | [Context] | [DB] [Config] [Event Queue] [LLM]Each plugin registers itself with the @register() decorator and defines handlers using @filter decorators. The Context API gives you access to AstrBot’s core functionality: database, configuration, event queue, and LLM providers.
Creating Your First Plugin
I started with a simple “hello world” plugin to understand the structure:
from astrbot.api.event import filter, AstrMessageEventfrom astrbot.api.star import Context, Star, registerfrom astrbot import logger
@register("helloworld", "cowrie", "A simple Hello World plugin", "1.0.0", "https://github.com/example/helloworld")class HelloWorldPlugin(Star): def __init__(self, context: Context): super().__init__(context) logger.info("HelloWorld plugin initialized")
@filter.command("helloworld") async def helloworld(self, event: AstrMessageEvent): user_name = event.get_sender_name() logger.info(f"Hello world command triggered by {user_name}") yield event.plain_result(f"Hello, {user_name}!")Key parts:
@register()defines plugin metadata: name, author, description, version, repo URL- The class extends
Starand takes aContextin__init__ @filter.command("helloworld")registers a command handleryield event.plain_result()sends a text response back to the user
When I send /helloworld in any connected platform, the bot responds with “Hello, [my name]!”.
Plugin Directory Structure
AstrBot discovers plugins automatically from the addons/ directory. My plugin structure:
addons/└── helloworld/ └── main.pyThat’s it - no complex configuration files. AstrBot scans the directory and loads any file with the @register() decorator.
Building Custom Commands
The @filter.command() decorator handles command registration. I wanted to understand argument parsing, so I built an echo command:
@filter.command("echo")async def echo_command(self, event: AstrMessageEvent): message = event.get_plain_text().replace('/echo', '').strip() if not message: yield event.plain_result("Usage: /echo <message>") return yield event.plain_result(message)This works, but it’s fragile. If someone types /echo with extra spaces, the replace might not work correctly. I needed a better approach.
Admin-Only Commands
Some commands should only work for administrators. AstrBot supports permission checking:
@filter.command("admincmd", permission=True)async def admin_only(self, event: AstrMessageEvent): yield event.plain_result("Admin command executed")When permission=True, only users with admin privileges can trigger this command.
Event Handling Beyond Commands
Commands are just one type of event. I wanted my plugin to react to all messages, not just commands. The @filter.on_message() decorator does this:
@filter.on_message()async def on_any_message(self, event: AstrMessageEvent): text = event.get_plain_text() if "hello" in text.lower(): yield event.plain_result("Hi there!")This triggers on every message. But I realized this could get noisy - I wanted to filter by platform:
@filter.on_message(platform=["qq"])async def on_qq_message(self, event: AstrMessageEvent): # Only triggers for QQ platform messages logger.info(f"QQ message received: {event.get_plain_text()}")LLM Request Hooks
The most powerful feature I discovered: hooks that intercept LLM requests. This lets you modify the prompt before it reaches the LLM:
from astrbot.api.provider import ProviderRequest
@filter.on_llm_request()async def modify_llm_request(self, event: AstrMessageEvent, req: ProviderRequest): # Add custom instructions to system prompt req.system_prompt += "\n\nYou are speaking with users in a casual, friendly tone."
# Can't use yield here - this is a hook, not a response logger.info(f"Modified system prompt for user: {event.get_sender_name()}")Important: You can’t yield responses in LLM hooks. The hook modifies the request in-place, then returns control to the normal LLM flow.
Sending Rich Messages
Simple text responses work for basic plugins, but I wanted to send images, files, and other media. AstrBot uses MessageChain for this:
from astrbot.api.event import MessageChainfrom astrbot.api.message_components import Plain, Image, At
@filter.command("show_image")async def show_image(self, event: AstrMessageEvent): chain = MessageChain([ Plain("Here's the image you requested: "), Image(path="/path/to/image.png"), Plain(" Let me know if you need anything else!") ]) yield event.plain_result(chain)The MessageChain structure:
MessageChain├── Plain(text) # Text content├── At(user_id) # @mention a user├── Image(url/path) # Image attachment├── Record(path) # Voice/audio└── CustomComponent() # Platform-specificplain_result vs send
I initially confused event.plain_result() with event.send(). The difference:
plain_result(): Returns a response that yields back through the event pipelinesend(): Sends a message immediately, bypassing normal flow
Use send() when you need to send multiple messages or send outside the normal response flow:
@filter.command("multi_message")async def multi_message(self, event: AstrMessageEvent): await event.send(MessageChain([Plain("First message")]))
# Do some processing await asyncio.sleep(1)
await event.send(MessageChain([Plain("Second message")]))
# Final response yield event.plain_result("Done!")LLM Integration: Building Agent Skills
This is where AstrBot shines. I wanted to build a plugin that uses the LLM to perform multi-step tasks. The tool_loop_agent method handles this:
from astrbot.core.agent.tool import FunctionToolfrom astrbot.core.tool_set import ToolSet
class WeatherTool(FunctionTool): name = "get_weather" description = "Get current weather for a city"
async def call(self, context, **kwargs): city = kwargs.get("city", "Beijing") # In a real plugin, call a weather API here return f"Weather in {city}: Sunny, 25°C"
@filter.command("weather")async def weather_command(self, event: AstrMessageEvent): city = event.get_plain_text().replace('/weather', '').strip() or "Beijing"
llm_resp = await self.context.tool_loop_agent( event=event, chat_provider_id=await self.context.get_current_chat_provider_id(event.unified_msg_origin), prompt=f"What's the weather like in {city}?", tools=ToolSet([WeatherTool()]), max_steps=5, tool_call_timeout=30 )
yield event.plain_result(llm_resp.completion_text)The agent flow:
User Command --> Plugin Handler --> tool_loop_agent | [LLM Request] --> [Tool Decision] | | [Tool Execution] [More Tools?] | [Result Compilation]The LLM decides when to call tools, executes them, and returns a final response. I can add multiple tools to the ToolSet, and the agent will orchestrate them automatically.
Creating Custom Tools
Tools extend FunctionTool and define:
name: Tool identifierdescription: What the tool does (LLM uses this to decide when to call it)call(): The actual implementation
from astrbot.core.agent.tool import FunctionTool, ToolExecResultfrom pydantic import Field
class SearchTool(FunctionTool): name = "search_web" description = "Search the web for information"
parameters = Field(default_factory=lambda: { "type": "object", "properties": { "query": { "type": "string", "description": "Search query" } }, "required": ["query"] })
async def call(self, context, **kwargs): query = kwargs.get("query")
# Implement actual search logic results = await perform_search(query)
return ToolExecResult.success(results)Context API: Database and Configuration
Plugins often need to store data or read configuration. The Context API provides access:
@filter.command("stats")async def show_stats(self, event: AstrMessageEvent): # Access database db = self.context.get_database() user_id = event.get_sender_id()
# Read user data user_stats = await db.get("user_stats", user_id)
if not user_stats: user_stats = {"messages": 0, "joined": datetime.now().isoformat()} await db.set("user_stats", user_id, user_stats)
# Update stats user_stats["messages"] += 1 await db.set("user_stats", user_id, user_stats)
yield event.plain_result(f"You've sent {user_stats['messages']} messages!")
@filter.command("config")async def show_config(self, event: AstrMessageEvent): # Access plugin configuration config = self.context.get_config()
# Read specific values api_key = config.get("api_key", "not set")
yield event.plain_result(f"API Key configured: {bool(api_key)}")Platform Adapter Development
I wanted to connect AstrBot to a custom platform not officially supported. The solution: create a platform adapter:
from astrbot.api.platform import Platform, PlatformMetadata, register_platform_adapter
class CustomPlatform(Platform): async def send_text(self, session, message: str): # Implement text sending for your platform pass
async def send_image(self, session, image): # Implement image sending pass
async def send_file(self, session, file): # Implement file sending pass
@register_platform_adapter("custom_platform", CustomPlatform)def register_custom_platform(): return PlatformMetadata("Custom Platform", "1.0.0")The platform adapter handles message conversion between AstrBot’s internal format and the external platform’s API.
Common Issues I Encountered
Issue 1: Plugin Not Loading
My plugin didn’t show up in AstrBot. The cause: I forgot the @register() decorator. Without it, AstrBot has no way to know the class is a plugin.
Issue 2: Event Handler Not Triggering
My @filter.on_message() handler never fired. The problem: I was trying to yield a response, but the event had already been handled by another plugin. Using event.send() instead of yield fixed this.
Issue 3: LLM Hook Crashes
My @filter.on_llm_request() hook crashed AstrBot. The issue: I tried to yield a response inside the hook. LLM hooks can only modify the request, not send responses.
Issue 4: Database Errors
Database writes failed silently. I discovered the database is key-value based, and keys must be strings. I was passing integers as keys, which caused type errors.
Summary
In this post, I showed how to build AstrBot plugins using the Star system. The key points:
- Plugin Registration: Use
@register()to define metadata, extendStarclass - Event Handling:
@filter.command()for commands,@filter.on_message()for all messages - Message Responses: Use
MessageChainfor rich content,yield event.plain_result()for responses - LLM Integration: Create
FunctionToolclasses and usetool_loop_agentfor multi-step workflows - Context API: Access database and configuration through
self.context
The event-driven architecture makes plugins loosely coupled and platform-agnostic. Write once, run on QQ, WeChat, Telegram, or any platform AstrBot supports.
The most powerful feature is the agent framework - you can build sophisticated LLM-powered skills without managing the conversation state yourself. The agent handles tool calling, result compilation, and response generation automatically.
Final Words + More Resources
My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me
Here are also the most important links from this article along with some further resources that will help you in this scope:
Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!
Comments