LangBot Plugin Development & Debugging

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

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

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

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")

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

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

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)

# Returns list[str] of model UUIDs
models = await self.get_llm_models()
model_uuid = models[0]  # First available 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)

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

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

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

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

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

@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()

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

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

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

{
  "group-respond-rules": {
    "at": true,          // Respond when @bot
    "prefix": ["ai"],    // Respond to messages starting with "ai"
    "random": 0.0,       // Probability of responding to any message (0.0-1.0)
    "regexp": []         // Regex patterns
  }
}

docker restart <runtime-container>
# Wait ~5 seconds for plugin to re-mount

# Initialize admin account (first time only)
curl -X POST $BASE/api/v1/user/init \
  -H "Content-Type: application/json" \
  -d '{"user":"admin@test.com","password":"test123"}'

# Login
curl -X POST $BASE/api/v1/user/auth \
  -H "Content-Type: application/json" \
  -d '{"user":"admin@test.com","password":"test123"}'
# Returns: {"data":{"token":"eyJ..."}}

# Create provider
curl -X POST $BASE/api/v1/provider/providers \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"MyProvider","requester":"new-api-chat-completions","base_url":"https://api.example.com/v1","api_keys":["sk-xxx"]}'

# Create LLM model
curl -X POST $BASE/api/v1/provider/models/llm \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"gpt-4o-mini","provider_uuid":"<uuid>","abilities":["chat","tool-use"]}'

# List models
curl $BASE/api/v1/provider/models/llm -H "Authorization: Bearer $TOKEN"

# Get pipeline
curl $BASE/api/v1/pipelines -H "Authorization: Bearer $TOKEN"

# Update pipeline (e.g., set model, modify trigger rules)
curl -X PUT $BASE/api/v1/pipelines/<uuid> \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '<full pipeline JSON>'

spec:
  config:
    - name: model
      type: llm-model-selector
      label:
        en_US: 'LLM Model'
        zh_Hans: 'LLM 模型'
      description:
        en_US: 'Select the LLM model. Falls back to first available if not set.'
        zh_Hans: '选择 LLM 模型。未设置时使用第一个可用模型。'
      required: false

model_uuid = self.get_config().get("model")

docker restart langbot-test-runtime
# Wait ~15 seconds before testing

docker restart langbot-test-runtime
sleep 8
docker restart langbot-test
sleep 8

cd /path/to/MyPlugin
lbp publish

MyPlugin/
├── manifest.yaml       # Only EventListener in spec.components
├── main.py             # BasePlugin with shared logic (fetch, LLM calls)
├── components/
│   └── event_listener/
│       ├── detector.yaml
│       └── detector.py
└── requirements.txt

spec:
  components:
    EventListener:
      fromDirs:
      - path: components/event_listener/

@self.handler(events.PersonMessageReceived)
async def on_msg(event_context: context.EventContext):
    event = event_context.event
    text_parts = []
    for component in event.message_chain:
        if isinstance(component, platform_message.Plain):
            text_parts.append(component.text)
    text = "".join(text_parts).strip()
    
    if should_handle(text):
        event_context.prevent_default()
        event_context.prevent_postorder()
        result = await self.plugin.process(text)
        await event_context.reply(platform_message.MessageChain([
            platform_message.Plain(text=result)
        ]))