--- title: "LangBot プラグインシステム詳解：プロセス分離・イベント駆動・コンポーネント設計" description: "A source-level deep dive into LangBot 4.0's plugin architecture — process-isolated communication, event-driven hooks, a component-based extension model, and comparisons with Dify and MCP." --- この記事は LangBot Blog から同期されています。 [Blog の原文を読む](https://langbot.app/en/blog/langbot-plugin-system-deep-dive). 公開日：2026-02-23 · 著者：LangBot Team > この記事の日本語版はまだ公開されていないため、英語版を掲載しています。 ![LangBot Plugin System](/images/articles/source/images/covers/plugin-system.png) Most chatbot frameworks call their "plugin system" a glorified dynamic import of Python modules. LangBot 4.0 takes a harder but more principled approach — **every plugin runs in its own process**, communicating with the host through a structured JSON-RPC-style protocol. This article dissects the system from source code, end to end. ## Overall Architecture: A Three-Layer Process Model LangBot's plugin system consists of three cooperating process layers: ![LangBot Plugin System Architecture](/images/articles/source/images/plugin-system/architecture.png) **Each layer has a distinct responsibility:** 1. **LangBot Main Process**: Runs business logic (message pipelines, platform adapters, model invocations), connects to Runtime via `PluginRuntimeConnector`. 2. **Plugin Runtime**: The orchestration layer — discovers, launches, and manages all plugin subprocesses, routes requests from the main process to the appropriate plugin. 3. **Plugin Subprocesses**: Each plugin runs in its own Python process, communicating with Runtime via stdio pipes. ### Why Three Layers Instead of Two? The intuitive design would have the main process manage plugin processes directly. LangBot adds the Runtime layer for **deployment flexibility**: - **Local development**: Main process spawns Runtime as a child via stdio (zero config) - **Docker production**: Runtime runs as a separate container, connected via WebSocket - **Windows compatibility**: Since Windows asyncio has incomplete stdio subprocess support, it automatically falls back to WebSocket The same codebase — no config changes — adapts from development to production. ## Communication Protocol: JSON-RPC-Style Request/Response All cross-process communication runs on a unified protocol layer. The core data structures are minimal: ```python # Request class ActionRequest(pydantic.BaseModel): seq_id: int # Sequence number for matching request/response action: str # Action name data: dict # Payload # Response class ActionResponse(pydantic.BaseModel): seq_id: int code: int # 0 = success message: str data: dict chunk_status: str # "continue" | "end" (streaming support) ``` The `Handler` class is the system's core abstraction, acting as both RPC client and server: ```python class Handler: async def call_action(self, action, data, timeout=15.0) -> dict: """Actively call an action provided by the peer, wait for response""" self.seq_id_index += 1 request = ActionRequest.make_request(self.seq_id_index, action.value, data) future = asyncio.Future() self.resp_waiters[self.seq_id_index] = future await self.conn.send(json.dumps(request.model_dump())) response = await asyncio.wait_for(future, timeout) return response.data @action(SomeAction.DO_SOMETHING) async def handle_something(data: dict) -> ActionResponse: """Register an action for the peer to call""" return ActionResponse.success({"result": "ok"}) ``` **Key design points:** - `seq_id`-based request/response matching enables full-duplex concurrent calls - Streaming responses via `chunk_status` for long-running operations like command execution - Large messages auto-chunk (stdio: 16KB / WebSocket: 64KB per chunk) - File transfer uses a separate base64 chunking mechanism ### Action Enums: Clear API Contracts The system defines all cross-process calls through four enum groups: ```python # Plugin → Runtime (plugin-initiated requests) class PluginToRuntimeAction: REGISTER_PLUGIN = "register_plugin" SEND_MESSAGE = "send_message" # Send message to a platform INVOKE_LLM = "invoke_llm" # Call an LLM SET_PLUGIN_STORAGE = "set_plugin_storage" # Persistent storage # ... # Runtime → Plugin (runtime-dispatched commands) class RuntimeToPluginAction: INITIALIZE_PLUGIN = "initialize_plugin" EMIT_EVENT = "emit_event" CALL_TOOL = "call_tool" EXECUTE_COMMAND = "execute_command" SHUTDOWN = "shutdown" # ... # LangBot Main → Runtime class LangBotToRuntimeAction: INSTALL_PLUGIN = "install_plugin" EMIT_EVENT = "emit_event" LIST_TOOLS = "list_tools" # ... # Runtime → LangBot Main class RuntimeToLangBotAction: GET_PLUGIN_SETTINGS = "get_plugin_settings" SET_BINARY_STORAGE = "set_binary_storage" # ... ``` This makes API boundaries crystal clear — what a plugin can and cannot do is defined entirely by these enums. ## Plugin Lifecycle A plugin goes through these stages from installation to execution: ### 1. Discovery On startup, Runtime scans the `data/plugins/` directory: ```python async def launch_all_plugins(self): for plugin_path in glob.glob("data/plugins/*"): if not os.path.isdir(plugin_path): continue task = self.launch_plugin(plugin_path) self.plugin_run_tasks.append(task) ``` Directory names follow the `{author}__{name}` convention, each containing a `manifest.yaml` and plugin code. ### 2. Launch Runtime spawns an independent subprocess for each plugin: ```python async def launch_plugin(self, plugin_path: str): python_path = sys.executable args = ["-m", "langbot_plugin.cli.__init__", "run", "-s", "--prod"] ctrl = StdioClientController( command=python_path, args=args, working_dir=plugin_path, # Each plugin runs in its own directory ) await ctrl.run(new_plugin_connection_callback) ``` **Key detail**: The subprocess working directory is set to the plugin's own directory — natural filesystem isolation. ### 3. Registration After starting, the plugin process actively registers itself with Runtime: ```python # Runtime-side registration handler async def register_plugin(self, handler, container_data, debug_plugin=False): plugin_container = PluginContainer.from_dict(container_data) # Fetch plugin settings from the main process plugin_settings = await self.context.control_handler.call_action( RuntimeToLangBotAction.GET_PLUGIN_SETTINGS, {...} ) # Initialize the plugin (send config) await handler.initialize_plugin(plugin_settings) # Store the plugin container self.plugins.append(plugin_container) ``` ### 4. Running Once in `INITIALIZED` state, the plugin can receive events, tool calls, and command executions. ### 5. Shutdown ```python async def shutdown_plugin(self, plugin_container): # 1. Notify the plugin to shut down gracefully await plugin_container._runtime_plugin_handler.shutdown_plugin() # 2. Close the communication connection await plugin_container._runtime_plugin_handler.conn.close() # 3. Kill the subprocess if handler.stdio_process is not None: handler.stdio_process.kill() await asyncio.wait_for(handler.stdio_process.wait(), timeout=2) ``` ## Component System: Four Extension Types A LangBot plugin isn't a single hook function — it's a **component container**. A single plugin can provide multiple component types simultaneously: ### EventListener The most fundamental extension — listen for events in the message pipeline: ```python from langbot_plugin.api.definition.components.common.event_listener import EventListener from langbot_plugin.api.entities.events import PersonNormalMessageReceived from langbot_plugin.api.entities.context import EventContext class MyListener(EventListener): @EventListener.handler(PersonNormalMessageReceived) async def on_person_message(self, ctx: EventContext): event = ctx.event # Modify the user message before it reaches the LLM event.user_message_alter = "Answer in poetry: " + event.text_message # Or block further processing # ctx.prevent_default() # ctx.prevent_postorder() ``` Supported events cover the full message lifecycle: | Event | Trigger | |-------|---------| | `PersonMessageReceived` | Any private message received | | `GroupMessageReceived` | Any group message received | | `PersonNormalMessageReceived` | Private message deemed processable | | `GroupNormalMessageReceived` | Group message deemed processable | | `NormalMessageResponded` | LLM response completed | | `PromptPreProcessing` | Prompt preprocessing stage | Event propagation supports two interruption modes: - `prevent_default()`: Skip default behavior (e.g., skip the LLM call) - `prevent_postorder()`: Stop subsequent plugins from running ### Tool Tools for LLM Function Calling: ```python from langbot_plugin.api.definition.components.tool.tool import Tool class WeatherTool(Tool): async def call(self, params: dict, session, query_id: int) -> str: city = params.get("city", "Beijing") # Call weather API... return f"{city}: Sunny, 25°C" ``` Tool metadata (name, description, parameter schema) is defined in a companion YAML manifest file. LangBot automatically converts this into the Function definition that LLMs understand. ### Command User-triggered commands via `!command`, with subcommand support: ```python from langbot_plugin.api.definition.components.command.command import Command class MyCommand(Command): def __init__(self): super().__init__() @self.subcommand("hello", help="Say hello") async def hello(self, ctx): yield CommandReturn(text="Hello from plugin!") @self.subcommand("status", help="Show status") async def status(self, ctx): yield CommandReturn(text="All systems operational.") ``` Command results are returned via `AsyncGenerator`, providing natural streaming output. ### KnowledgeRetriever A multi-instance component for connecting external knowledge bases: ```python from langbot_plugin.api.definition.components.knowledge_retriever.retriever import KnowledgeRetriever class MyRetriever(KnowledgeRetriever): async def retrieve(self, context) -> list: results = await self.search_external_db(context.query) return [RetrievalResultEntry(content=r) for r in results] ``` KnowledgeRetriever is a **polymorphic component** — a single retriever class can spawn multiple instances, each with independent configuration. This allows users to connect multiple different external knowledge bases. ## SDK API: What Plugins Can Do Plugins gain rich capabilities through the `LangBotAPIProxy` inherited by `BasePlugin`: ```python class LangBotAPIProxy: # Message operations async def send_message(self, bot_uuid, target_type, target_id, message_chain) # Model invocation async def get_llm_models(self) -> list[str] async def invoke_llm(self, model_uuid, messages, funcs=[], extra_args={}) # Persistent storage (plugin-level isolation) async def set_plugin_storage(self, key, value: bytes) async def get_plugin_storage(self, key) -> bytes # Workspace storage (cross-plugin shared) async def set_workspace_storage(self, key, value: bytes) async def get_workspace_storage(self, key) -> bytes # System info async def get_langbot_version(self) -> str async def get_bots(self) -> list[str] async def list_plugins_manifest(self) -> list ``` **The storage API design is worth noting**: Two levels of KV storage — `plugin_storage` (plugin-private) and `workspace_storage` (globally shared), storing data as bytes (base64-serialized in transit). Simple but flexible enough. ## Event Dispatch Mechanism The complete path from main process to plugin: ![Event Dispatch Flow](/images/articles/source/images/plugin-system/event-dispatch.png) Key source code: ```python async def emit_event(self, event_context, include_plugins=None): for plugin in self.plugins: if plugin.status != RuntimeContainerStatus.INITIALIZED: continue if not plugin.enabled: continue # Pipeline-level plugin filtering if include_plugins is not None: plugin_id = f"{plugin.manifest.metadata.author}/{plugin.manifest.metadata.name}" if plugin_id not in include_plugins: continue resp = await plugin._runtime_plugin_handler.emit_event( event_context.model_dump() ) event_context = EventContext.model_validate(resp["event_context"]) # Plugin requested propagation stop if event_context.is_prevented_postorder(): break return emitted_plugins, event_context ``` The `include_plugins` parameter enables **pipeline-level plugin binding** — different message processing pipelines can use different subsets of plugins. ## Installation & Distribution Plugins support three installation sources: 1. **Local upload**: `.lbpkg` files (actually zip archives containing manifest.yaml and code) 2. **Marketplace**: Install from LangBot Space online 3. **GitHub Release**: Download from a GitHub repository's Release assets The installation flow: ```python async def install_plugin(self, source, install_info): yield {"current_action": "downloading plugin package"} # 1. Fetch and extract the plugin package (unzip) plugin_path, author, name, version = await self.install_plugin_from_file(plugin_file) yield {"current_action": "installing dependencies"} # 2. Install dependencies (pip install -r requirements.txt) pkgmgr_helper.install_requirements(requirements_file) yield {"current_action": "initializing plugin settings"} # 3. Initialize configuration await self.context.control_handler.call_action( RuntimeToLangBotAction.INITIALIZE_PLUGIN_SETTINGS, {...} ) yield {"current_action": "launching plugin"} # 4. Launch the plugin process task = self.launch_plugin(plugin_path) ``` The entire process reports progress via `AsyncGenerator`, enabling real-time installation status in the frontend. ## Developer Experience The SDK provides a complete developer toolchain: ```bash # Initialize a new plugin lbp init # Add a component lbp component add # Run locally for debugging lbp run # Package for publication lbp publish ``` **Debug mode** has a particularly clever design: the developer's plugin connects to the running Runtime via WebSocket (instead of stdio), meaning you can hot-reload plugin code without restarting LangBot. Debug plugins are specially marked in the UI and protected from accidental deletion. ## Comparisons with Other Systems ### vs Dify Plugins Dify's plugin system (`dify-plugin-daemon`) shares the process isolation philosophy with LangBot, but the focus differs: - **Dify**: Plugins extend workflow node types (Tool, Model, Extension) — designed for AI application orchestration - **LangBot**: Plugins extend the message processing pipeline (Event, Tool, Command, KnowledgeRetriever) — designed for instant messaging scenarios LangBot's `EventListener` component provides a capability Dify lacks — injecting logic at any stage of message processing. ### vs MCP (Model Context Protocol) MCP is a standardized protocol for AI tool invocation. LangBot's Tool component and MCP services overlap functionally, but serve different purposes: - **MCP**: A universal "AI calls external capabilities" protocol, usable by any LLM application - **LangBot Tool**: Deeply integrated with message processing context, with access to session info, user identity, etc. In practice, LangBot natively supports MCP — users can configure MCP servers directly in LangBot without writing plugins. LangBot's Tool component is for scenarios requiring access to LangBot's internal context. ## Design Decisions Explained **Why process isolation instead of threads/coroutines?** - Plugin code quality is unpredictable; a segfault shouldn't crash the entire service - Dependency isolation: different plugins may depend on different versions of the same library - Resource control: you can set per-plugin process resource limits **Why JSON instead of Protobuf/MessagePack?** - Debug-friendly: developers can directly read communication logs - Natively supported in Python, no extra dependencies - The performance bottleneck isn't serialization (plugin call frequency is far below database queries) **Why stdio over WebSocket by default?** - stdio requires no network stack — lower latency - Simpler process lifecycle management (child processes auto-cleanup when parent exits) - WebSocket is only used where stdio isn't supported (Docker, Windows) ## Conclusion LangBot's plugin system is a **production-grade, process-isolated, event-driven component framework for extensibility**. Its core design principles: 1. **Safety first**: Process isolation ensures plugins can't destabilize the main service 2. **Deployment flexibility**: Dual stdio/WebSocket modes adapt to all environments 3. **Developer-friendly**: Complete SDK, CLI, and debug support 4. **Component-based**: Four component types cover the major extension needs If you're interested in developing LangBot plugins, start with the [plugin development docs](https://langbot.app/docs/en/plugin/dev/tutor), or browse existing plugins on the [marketplace](https://space.langbot.app/market) for inspiration.