LangBot Plugin Development

LangBot Plugin Development & Debugging

Controlling a running instance via MCP

Beyond writing code, you can drive a live LangBot instance over MCP — no raw HTTP needed. Two MCP servers exist (both reuse existing API keys; see AGENTS.md):

• LangBot instance — http://<host>:5300/mcp (auth: web-UI lbk_ key or the api.global_api_key from config.yaml). Manage bots, pipelines, models, knowledge bases, and skills. See the langbot-mcp-ops skill.
• LangBot Space marketplace — https://space.langbot.app/mcp (auth: Personal Access Token). Search plugins / MCP servers / skills. See the langbot-space-ops skill.

Any change to an agent-accessible HTTP API endpoint must keep the matching MCP tool and these skills in sync.

Plugin Architecture

A LangBot plugin consists of:

MyPlugin/
├── manifest.yaml          # Plugin metadata, config schema
├── main.py                # BasePlugin subclass (entry point, shared state)
├── components/
│   ├── event_listener/    # Hook pipeline events
│   │   ├── collector.yaml
│   │   └── collector.py
│   ├── commands/          # !command handlers
│   │   ├── mycommand.yaml
│   │   └── mycommand.py
│   └── tools/             # LLM function-call tools
│       ├── mytool.yaml
│       └── mytool.py

Each component has a .yaml (metadata) and .py (implementation).

README & i18n convention (enforced on the marketplace)

A plugin published to LangBot Space serves a localized README on its detail page. The resolver (langbot-space PluginService.GetPluginREADME) works like this:

• Root README.md MUST be in English. It is the default and the fallback — when no per-language README matches the viewer's locale, the page serves the root README.md. A non-English root README makes the English/default view show the wrong language.
• All other languages live under readme/README_{lang}.md — e.g. readme/README_zh_Hans.md, readme/README_ja_JP.md. The 8 supported locales: en_US, zh_Hans, zh_Hant, ja_JP, th_TH, vi_VN, es_ES, ru_RU.
• manifest.yaml metadata.label / metadata.description should carry the same 8-locale i18n set (repository must be a real, alive URL).

MyPlugin/
├── manifest.yaml
├── README.md                   # English (default + fallback) — REQUIRED, must be English
└── readme/
    ├── README_zh_Hans.md
    ├── README_zh_Hant.md
    ├── README_ja_JP.md
    ├── README_th_TH.md
    ├── README_vi_VN.md
    ├── README_es_ES.md
    └── README_ru_RU.md

manifest.yaml (incl. repository) is the source of truth — the marketplace syncs from it, so edit the package and re-publish rather than patching live data.

Critical SDK Pitfalls

1. MessageChain is a RootModel — iterate directly

# ❌ WRONG — MessageChain has no .components attribute
for component in event.message_chain.components:

# ✅ CORRECT — MessageChain is a Pydantic RootModel, iterate directly
for component in event.message_chain:

2. Message.content must be list[ContentElement] or str, not a single ContentElement

from langbot_plugin.api.entities.builtin.provider import message as provider_message

# ❌ WRONG — single ContentElement
Message(role="user", content=ContentElement.from_text("hello"))

# ✅ CORRECT — list of ContentElement
Message(role="user", content=[ContentElement.from_text("hello")])

# ✅ ALSO CORRECT — plain string
Message(role="user", content="hello")

3. invoke_llm does NOT accept timeout

# ❌ WRONG
await self.invoke_llm(llm_model_uuid=uuid, messages=msgs, timeout=60)

# ✅ CORRECT
await self.invoke_llm(llm_model_uuid=uuid, messages=msgs)

4. invoke_llm response.content can be str OR list

response = await self.invoke_llm(...)
if response.content:
    if isinstance(response.content, str):
        return response.content
    elif isinstance(response.content, list):
        parts = [e.text for e in response.content if hasattr(e, "text") and e.text]
        return "\n".join(parts)

5. get_llm_models() returns UUIDs

# Returns list[str] of model UUIDs
models = await self.get_llm_models()
model_uuid = models[0]  # First available model UUID

Known bug (v4.9.3): The host handler may return list[dict] instead of list[str]. If you hit TypeError: unhashable type: 'dict' in invoke_llm, the fix is in LangBot/src/langbot/pkg/plugin/handler.py — change 'llm_models': llm_models to 'llm_models': [m['uuid'] for m in llm_models].

6. invoke_llm parameter is llm_model_uuid, NOT model_uuid

# ❌ WRONG — will throw "got an unexpected keyword argument"
await self.invoke_llm(messages=msgs, model_uuid=uuid)

# ✅ CORRECT
await self.invoke_llm(messages=msgs, llm_model_uuid=uuid)

7. prevent_default() alone does NOT block LLM response

To fully prevent the default LLM pipeline from responding when your EventListener handles the message, you must call both:

event_context.prevent_default()    # Block default behavior
event_context.prevent_postorder()  # Block subsequent plugins/pipeline

Using only prevent_default() still allows the LLM to generate a response.

8. get_plugin_storage / set_plugin_storage may throw KeyError: 'owner'

This is a version mismatch between the SDK and host. Wrap storage calls in try/except:

try:
    data = await self.get_plugin_storage("my_key")
except Exception:
    data = None  # Fallback gracefully

9. Component YAML must have full structure, not just name/description

# ❌ WRONG — will silently fail to register the component
name: translator
description:
  en_US: 'Does stuff'

# ✅ CORRECT — full component YAML
apiVersion: v1
kind: EventListener
metadata:
  name: translator
  label:
    en_US: Translator
spec:
execution:
  python:
    path: translator.py
    attr: Translator

10. BasePlugin import path

# ❌ WRONG
from langbot_plugin.api.definition.base_plugin import BasePlugin

# ✅ CORRECT
from langbot_plugin.api.definition.plugin import BasePlugin

Pipeline Events

Events the EventListener can hook (from most general to most specific):

Key insight: *MessageReceived fires for ALL messages regardless of trigger rules. *NormalMessageReceived only fires for messages that match the pipeline's trigger rules (e.g., @bot, prefix, random%). Use *MessageReceived for message collection/logging.

EventContext API

@self.handler(events.GroupMessageReceived)
async def on_msg(event_context: context.EventContext):
    event = event_context.event
    event.launcher_id    # Group ID
    event.sender_id      # Sender ID
    event.message_chain  # MessageChain (iterate directly)

    # Reply to the current conversation
    await event_context.reply(MessageChain([Plain(text="hello")]))

    # Block default pipeline behavior
    event_context.prevent_default()

    # Block subsequent plugins
    event_context.prevent_postorder()

Setting Up a Test Environment

Deploy via Docker (GitOps + Portainer)

See references/test-env-setup.md for full deployment steps.

Quick summary:

1. Create docker-compose.yaml in server-deploy repo
2. Deploy via Portainer git repository method
3. Set up admin account via /api/v1/user/init POST
4. Configure LLM provider and model via API
5. Copy plugin to data/plugins/ directory

WebSocket Testing

LangBot's WebUI chat uses WebSocket. Connect to test message flow:

ws://<host>:<port>/api/v1/pipelines/<pipeline_uuid>/ws/connect?session_type=group

• session_type=group for group chat simulation
• session_type=person for private chat (always triggers pipeline)

Requires Origin header to pass CORS:

const ws = new WebSocket(url, {
  headers: { Origin: 'https://your-langbot-domain' }
});

Send messages:

{"type": "message", "message": [{"type": "Plain", "text": "hello"}]}

Receive:

• {"type": "connected", ...} — connection established
• {"type": "user_message", "data": {...}} — echo of sent message
• {"type": "response", "data": {"content": "...", "is_final": true/false}} — bot reply (streamed)