Merge remote-tracking branch 'origin/master' into feat/workflow

src/langbot/pkg/platform/botmgr.py

@@ -487,7 +487,7 @@ class PlatformManager:
         return None
 
     async def remove_bot(self, bot_uuid: str):
-        for bot in self.bots:
+        for bot in self.bots[:]:
             if bot.bot_entity.uuid == bot_uuid:
                 if bot.enable:
                     await bot.shutdown()

src/langbot/pkg/platform/sources/matrix.py

@@ -0,0 +1,693 @@
+from __future__ import annotations
+
+import typing
+import asyncio
+import traceback
+import base64
+import json
+
+import nio
+
+from langbot.pkg.utils import httpclient
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+
+
+class MatrixMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
+    @staticmethod
+    async def yiri2target(message_chain: platform_message.MessageChain, client: nio.AsyncClient) -> list[dict]:
+        components = []
+        for component in message_chain:
+            if isinstance(component, platform_message.Plain):
+                components.append({'type': 'text', 'text': component.text})
+            elif isinstance(component, platform_message.Image):
+                image_bytes = None
+                if component.base64:
+                    b64_data = component.base64
+                    if ';base64,' in b64_data:
+                        b64_data = b64_data.split(';base64,', 1)[1]
+                    image_bytes = base64.b64decode(b64_data)
+                elif component.url:
+                    session = httpclient.get_session()
+                    async with session.get(component.url) as response:
+                        image_bytes = await response.read()
+                elif component.path:
+                    with open(component.path, 'rb') as f:
+                        image_bytes = f.read()
+                if image_bytes:
+                    resp = await client.upload(image_bytes, content_type='image/png')
+                    if isinstance(resp, nio.UploadResponse):
+                        components.append({'type': 'image', 'mxc_url': resp.content_uri})
+            elif isinstance(component, platform_message.File):
+                file_bytes = None
+                if component.base64:
+                    b64_data = component.base64
+                    if ';base64,' in b64_data:
+                        b64_data = b64_data.split(';base64,', 1)[1]
+                    file_bytes = base64.b64decode(b64_data)
+                elif component.url:
+                    session = httpclient.get_session()
+                    async with session.get(component.url) as response:
+                        file_bytes = await response.read()
+                elif component.path:
+                    with open(component.path, 'rb') as f:
+                        file_bytes = f.read()
+                if file_bytes:
+                    file_name = getattr(component, 'name', None) or 'file'
+                    resp = await client.upload(file_bytes, content_type='application/octet-stream', filename=file_name)
+                    if isinstance(resp, nio.UploadResponse):
+                        components.append(
+                            {
+                                'type': 'file',
+                                'mxc_url': resp.content_uri,
+                                'filename': file_name,
+                                'size': len(file_bytes),
+                            }
+                        )
+            elif isinstance(component, platform_message.Forward):
+                for node in component.node_list:
+                    components.extend(await MatrixMessageConverter.yiri2target(node.message_chain, client))
+        return components
+
+    @staticmethod
+    async def target2yiri(event: nio.RoomMessageText | nio.RoomMessageImage, client: nio.AsyncClient, bot_user_id: str):
+        message_components = []
+
+        if isinstance(event, nio.RoomMessageText):
+            text = event.body
+            if bot_user_id and bot_user_id in text:
+                message_components.append(platform_message.At(target=bot_user_id))
+                text = text.replace(bot_user_id, '').strip()
+            message_components.append(platform_message.Plain(text=text))
+
+        elif isinstance(event, nio.RoomMessageImage):
+            mxc_url = event.url
+            if mxc_url:
+                resp = await client.download(mxc_url)
+                if isinstance(resp, nio.DownloadResponse):
+                    b64 = base64.b64encode(resp.body).decode('utf-8')
+                    content_type = resp.content_type or 'image/png'
+                    message_components.append(platform_message.Image(base64=f'data:{content_type};base64,{b64}'))
+            if event.body:
+                message_components.append(platform_message.Plain(text=event.body))
+
+        return platform_message.MessageChain(message_components)
+
+
+class MatrixEventConverter(abstract_platform_adapter.AbstractEventConverter):
+    @staticmethod
+    async def yiri2target(event: platform_events.MessageEvent):
+        return event.source_platform_object
+
+    @staticmethod
+    async def target2yiri(
+        event: nio.RoomMessageText | nio.RoomMessageImage,
+        room: nio.MatrixRoom,
+        client: nio.AsyncClient,
+        bot_user_id: str,
+        bridge_user_ids: list[str] | None = None,
+    ):
+        lb_message = await MatrixMessageConverter.target2yiri(event, client, bot_user_id)
+
+        # Determine if this is a direct/private chat or a group chat.
+        # Exclude bot itself and bridge bots, count remaining real users.
+        exclude_ids = {bot_user_id}
+        if bridge_user_ids:
+            exclude_ids.update(bridge_user_ids)
+        real_users = [uid for uid in room.users if uid not in exclude_ids]
+        is_direct = len(real_users) <= 1
+
+        if is_direct:
+            return platform_events.FriendMessage(
+                sender=platform_entities.Friend(
+                    id=event.sender,
+                    nickname=room.user_name(event.sender) or event.sender,
+                    remark='',
+                ),
+                message_chain=lb_message,
+                time=event.server_timestamp / 1000.0,
+                source_platform_object={'event': event, 'room': room},
+            )
+        else:
+            return platform_events.GroupMessage(
+                sender=platform_entities.GroupMember(
+                    id=event.sender,
+                    member_name=room.user_name(event.sender) or event.sender,
+                    permission=platform_entities.Permission.Member,
+                    group=platform_entities.Group(
+                        id=room.room_id,
+                        name=room.display_name or room.room_id,
+                        permission=platform_entities.Permission.Member,
+                    ),
+                    special_title='',
+                ),
+                message_chain=lb_message,
+                time=event.server_timestamp / 1000.0,
+                source_platform_object={'event': event, 'room': room},
+            )
+
+
+class BridgeState:
+    """Per-bridge runtime state."""
+
+    def __init__(self, user_id: str, login_command: str, logout_command: str, success_keyword: str, check_command: str):
+        self.user_id = user_id
+        self.login_command = login_command
+        self.logout_command = logout_command
+        self.success_keyword = success_keyword
+        self.check_command = check_command or login_command
+        self.logged_in = False
+        self.dm_room_id: str | None = None
+        self.login_task: asyncio.Task | None = None
+        self.check_task: asyncio.Task | None = None
+        self.check_responded = False
+
+
+class MatrixAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
+    client: typing.Any = None
+    message_converter: MatrixMessageConverter = MatrixMessageConverter()
+    event_converter: MatrixEventConverter = MatrixEventConverter()
+    config: dict
+    listeners: typing.Dict[typing.Type[platform_events.Event], typing.Callable] = {}
+    _running: bool = False
+    _initial_sync_done: bool = False
+    _bridges: list = []
+
+    def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
+        homeserver_url = config.get('homeserver_url', '')
+        access_token = config.get('access_token', '')
+        user_id = config.get('user_id', '')
+
+        if not homeserver_url or not access_token or not user_id:
+            raise ValueError('Matrix 机器人缺少必要配置项 (homeserver_url, user_id, access_token)')
+
+        client = nio.AsyncClient(homeserver_url, user_id)
+        client.access_token = access_token
+        client.user_id = user_id
+
+        super().__init__(
+            config=config,
+            logger=logger,
+            bot_account_id=user_id,
+            client=client,
+            listeners={},
+        )
+
+        # Parse bridges config AFTER super().__init__() to avoid Pydantic resetting _bridges
+        self._bridges = []
+        bridges_raw = config.get('bridges', '')
+        if bridges_raw:
+            if isinstance(bridges_raw, str):
+                try:
+                    bridges_list = json.loads(bridges_raw)
+                except (json.JSONDecodeError, TypeError) as e:
+                    raise ValueError(f'bridges 配置 JSON 解析失败: {e}\n原始值: {bridges_raw}')
+            else:
+                bridges_list = bridges_raw
+            for b in bridges_list:
+                if isinstance(b, dict) and b.get('user_id', '').strip():
+                    self._bridges.append(
+                        BridgeState(
+                            user_id=b['user_id'].strip(),
+                            login_command=b.get('login_command', '').strip(),
+                            logout_command=b.get('logout_command', '').strip(),
+                            success_keyword=b.get('success_keyword', 'Successfully logged in').strip(),
+                            check_command=b.get('check_command', '').strip(),
+                        )
+                    )
+        # Backward compatibility: old single-bridge config
+        if not self._bridges:
+            old_user_id = config.get('bridge_user_id', '').strip()
+            old_command = config.get('bridge_login_command', '').strip()
+            old_keyword = config.get('bridge_login_success_keyword', 'Successfully logged in').strip()
+            old_check = config.get('bridge_check_command', '').strip()
+            old_logout = config.get('bridge_logout_command', '').strip()
+            if old_user_id:
+                self._bridges.append(
+                    BridgeState(
+                        user_id=old_user_id,
+                        login_command=old_command,
+                        logout_command=old_logout,
+                        success_keyword=old_keyword,
+                        check_command=old_check,
+                    )
+                )
+
+    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
+        components = await self.message_converter.yiri2target(message, self.client)
+        for component in components:
+            await self._send_component(target_id, component)
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ):
+        source_obj = message_source.source_platform_object
+        room_id = source_obj['room'].room_id
+        components = await self.message_converter.yiri2target(message, self.client)
+
+        for component in components:
+            if quote_origin:
+                original_event = source_obj['event']
+                await self._send_component(room_id, component, reply_to=original_event.event_id)
+            else:
+                await self._send_component(room_id, component)
+
+    async def _send_component(self, room_id: str, component: dict, reply_to: str | None = None):
+        content = {}
+        if component['type'] == 'text':
+            content = {
+                'msgtype': 'm.text',
+                'body': component['text'],
+            }
+        elif component['type'] == 'image':
+            content = {
+                'msgtype': 'm.image',
+                'body': 'image.png',
+                'url': component['mxc_url'],
+            }
+        elif component['type'] == 'file':
+            content = {
+                'msgtype': 'm.file',
+                'body': component.get('filename', 'file'),
+                'url': component['mxc_url'],
+                'info': {'size': component.get('size', 0)},
+            }
+
+        if reply_to and content:
+            content['m.relates_to'] = {
+                'm.in_reply_to': {'event_id': reply_to},
+            }
+
+        if content:
+            await self.client.room_send(
+                room_id=room_id,
+                message_type='m.room.message',
+                content=content,
+            )
+
+    def register_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
+        ],
+    ):
+        self.listeners[event_type] = callback
+
+    async def run_async(self):
+        self._running = True
+        await self.logger.info('Matrix adapter starting...')
+
+        # Debug: log bridge parsing result
+        bridges_raw = self.config.get('bridges', '')
+        await self.logger.debug(f'bridges config raw: type={type(bridges_raw).__name__}, repr={repr(bridges_raw)}')
+        await self.logger.debug(
+            f'parsed _bridges count: {len(self._bridges)}, ids: {[b.user_id for b in self._bridges]}'
+        )
+
+        # Collect all bridge bot user IDs for filtering
+        _bridge_user_ids = [b.user_id for b in self._bridges]
+        _bridge_user_id_set = set(_bridge_user_ids)
+
+        # Auto-join invited rooms
+        async def on_invite(room: nio.MatrixRoom, event: nio.InviteMemberEvent):
+            if event.membership == 'invite' and event.state_key == self.client.user_id:
+                await self.client.join(room.room_id)
+                await self.logger.debug(f'Auto-joined room: {room.display_name or room.room_id}')
+
+        self.client.add_event_callback(on_invite, nio.InviteMemberEvent)
+
+        # Handle text messages
+        async def on_message(room: nio.MatrixRoom, event: nio.RoomMessageText):
+            if not self._initial_sync_done:
+                return
+            if event.sender == self.client.user_id:
+                return
+
+            # Admin commands (from any non-bridge user)
+            if event.sender not in _bridge_user_id_set:
+                body = (event.body or '').strip()
+                if body == '!relogin':
+                    await self._handle_relogin_command(room.room_id)
+                    return
+                if body == '!status':
+                    await self._handle_status_command(room.room_id)
+                    return
+
+            if event.sender in _bridge_user_id_set:
+                return
+            try:
+                lb_event = await self.event_converter.target2yiri(
+                    event, room, self.client, self.bot_account_id, _bridge_user_ids
+                )
+                if type(lb_event) in self.listeners:
+                    result = self.listeners[type(lb_event)](lb_event, self)
+                    if asyncio.iscoroutine(result):
+                        await result
+            except Exception:
+                await self.logger.error(f'Error handling Matrix message: {traceback.format_exc()}')
+
+        self.client.add_event_callback(on_message, nio.RoomMessageText)
+
+        # Handle image messages
+        async def on_image(room: nio.MatrixRoom, event: nio.RoomMessageImage):
+            if not self._initial_sync_done:
+                return
+            if event.sender == self.client.user_id:
+                return
+            if event.sender in _bridge_user_id_set:
+                return
+            try:
+                lb_event = await self.event_converter.target2yiri(
+                    event, room, self.client, self.bot_account_id, _bridge_user_ids
+                )
+                if type(lb_event) in self.listeners:
+                    result = self.listeners[type(lb_event)](lb_event, self)
+                    if asyncio.iscoroutine(result):
+                        await result
+            except Exception:
+                await self.logger.error(f'Error handling Matrix image: {traceback.format_exc()}')
+
+        self.client.add_event_callback(on_image, nio.RoomMessageImage)
+
+        # Set up bridge-specific callbacks for each bridge
+        _disconnect_keywords = ['disconnected', 'logged out', 'connection lost', 'session expired', 'token expired']
+
+        for bridge in self._bridges:
+            # Login success detection (notice)
+            async def on_bridge_notice(room: nio.MatrixRoom, event: nio.RoomMessageNotice, _b=bridge):
+                if not self._initial_sync_done:
+                    return
+                if event.sender != _b.user_id:
+                    return
+                _b.check_responded = True
+                if _b.success_keyword in (event.body or ''):
+                    _b.logged_in = True
+                    await self.logger.info(f'[{_b.user_id}] Bridge login succeeded.')
+                # Disconnect detection
+                body_lower = (event.body or '').lower()
+                for kw in _disconnect_keywords:
+                    if kw in body_lower and _b.logged_in:
+                        _b.logged_in = False
+                        await self.logger.info(f'[{_b.user_id}] Bridge 账号掉线 (检测到: "{kw}"), 将自动重新登录...')
+                        self._restart_bridge_login(_b)
+                        break
+
+            self.client.add_event_callback(on_bridge_notice, nio.RoomMessageNotice)
+
+            # Login success + disconnect detection (text)
+            async def on_bridge_text(room: nio.MatrixRoom, event: nio.RoomMessageText, _b=bridge):
+                if not self._initial_sync_done:
+                    return
+                if event.sender != _b.user_id:
+                    return
+                _b.check_responded = True
+                if _b.success_keyword in (event.body or ''):
+                    _b.logged_in = True
+                    await self.logger.info(f'[{_b.user_id}] Bridge login succeeded.')
+                body_lower = (event.body or '').lower()
+                for kw in _disconnect_keywords:
+                    if kw in body_lower and _b.logged_in:
+                        _b.logged_in = False
+                        await self.logger.info(f'[{_b.user_id}] Bridge 账号掉线 (检测到: "{kw}"), 将自动重新登录...')
+                        self._restart_bridge_login(_b)
+                        break
+
+            self.client.add_event_callback(on_bridge_text, nio.RoomMessageText)
+
+            # QR code image forwarding
+            async def on_bridge_image(room: nio.MatrixRoom, event: nio.RoomMessageImage, _b=bridge):
+                if not self._initial_sync_done:
+                    return
+                if event.sender != _b.user_id:
+                    return
+                mxc_url = event.url
+                if not mxc_url:
+                    return
+                try:
+                    resp = await self.client.download(mxc_url)
+                    if isinstance(resp, nio.DownloadResponse):
+                        b64 = base64.b64encode(resp.body).decode('utf-8')
+                        content_type = resp.content_type or 'image/png'
+                        await self.logger.info(
+                            f'[{_b.user_id}] Bridge 发送了二维码，请扫码登录:',
+                            images=[platform_message.Image(base64=f'data:{content_type};base64,{b64}')],
+                        )
+                except Exception:
+                    await self.logger.error(
+                        f'[{_b.user_id}] Failed to download bridge QR image: {traceback.format_exc()}'
+                    )
+
+            self.client.add_event_callback(on_bridge_image, nio.RoomMessageImage)
+
+        await self.logger.debug('Matrix adapter running, starting sync...')
+
+        # Initial sync to skip old messages
+        resp = await self.client.sync(timeout=10000)
+        if isinstance(resp, nio.SyncResponse):
+            await self.logger.debug(f'Matrix initial sync done, next_batch: {resp.next_batch}')
+        self._initial_sync_done = True
+
+        # Display account info
+        display_name = self.client.user_id
+        try:
+            profile_resp = await self.client.get_displayname(self.client.user_id)
+            if isinstance(profile_resp, nio.ProfileGetDisplayNameResponse) and profile_resp.displayname:
+                display_name = profile_resp.displayname
+        except Exception:
+            pass
+        joined_rooms = len(self.client.rooms)
+        homeserver = self.config.get('homeserver_url', '')
+        bridge_info = ''
+        if self._bridges:
+            bridge_names = ', '.join(b.user_id for b in self._bridges)
+            bridge_info = f' | 桥接: [{bridge_names}]'
+        await self.logger.info(
+            f'Matrix 账号: {display_name} ({self.client.user_id}) | '
+            f'服务器: {homeserver} | 已加入 {joined_rooms} 个房间{bridge_info}'
+        )
+
+        # Start bridge login and status check tasks for each bridge
+        for bridge in self._bridges:
+            if bridge.login_command:
+                await self.logger.info(
+                    f'[{bridge.user_id}] Bridge login enabled (命令: "{bridge.login_command}", '
+                    f'关键词: "{bridge.success_keyword}")'
+                )
+                bridge.login_task = asyncio.create_task(self._periodic_bridge_login(bridge))
+                bridge.check_task = asyncio.create_task(self._periodic_bridge_check(bridge))
+            else:
+                await self.logger.debug(f'[{bridge.user_id}] Bridge login not configured (no login_command)')
+
+        # Main sync loop
+        while self._running:
+            try:
+                await self.client.sync(timeout=30000)
+            except Exception:
+                await self.logger.error(f'Matrix sync error: {traceback.format_exc()}')
+                await asyncio.sleep(5)
+
+    async def _periodic_bridge_login(self, bridge: BridgeState):
+        """Periodically send login command to a bridge bot until login succeeds."""
+        try:
+            await self.logger.info(f'[{bridge.user_id}] Bridge login task started, looking for DM room...')
+            dm_room_id = None
+            for room_id, room in self.client.rooms.items():
+                if room.member_count == 2 and bridge.user_id in [m for m in room.users]:
+                    dm_room_id = room_id
+                    break
+
+            if not dm_room_id:
+                resp = await self.client.room_create(
+                    is_direct=True,
+                    invite=[bridge.user_id],
+                )
+                if isinstance(resp, nio.RoomCreateResponse):
+                    dm_room_id = resp.room_id
+                    await self.logger.debug(f'[{bridge.user_id}] Created DM room: {dm_room_id}')
+                else:
+                    await self.logger.error(f'[{bridge.user_id}] Failed to create DM room: {resp}')
+                    return
+
+            bridge.dm_room_id = dm_room_id
+
+            # Force logout first on every adapter start
+            logout_cmd = bridge.logout_command or bridge.login_command.replace('login', 'logout')
+            await self.logger.info(f'[{bridge.user_id}] 强制登出: "{logout_cmd}"')
+            await self.client.room_send(
+                room_id=dm_room_id,
+                message_type='m.room.message',
+                content={'msgtype': 'm.text', 'body': logout_cmd},
+            )
+            await asyncio.sleep(3)
+
+            while self._running and not bridge.logged_in:
+                await self.logger.debug(f'[{bridge.user_id}] Sending "{bridge.login_command}" in room {dm_room_id}')
+                await self.client.room_send(
+                    room_id=dm_room_id,
+                    message_type='m.room.message',
+                    content={'msgtype': 'm.text', 'body': bridge.login_command},
+                )
+                for _ in range(60):
+                    if not self._running or bridge.logged_in:
+                        break
+                    await asyncio.sleep(1)
+
+            if bridge.logged_in:
+                await self.logger.debug(f'[{bridge.user_id}] Bridge login confirmed, periodic login stopped.')
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            await self.logger.error(f'[{bridge.user_id}] Bridge periodic login error: {traceback.format_exc()}')
+
+    def _restart_bridge_login(self, bridge: BridgeState):
+        """Cancel existing login task and start a new one."""
+        if bridge.login_task and not bridge.login_task.done():
+            bridge.login_task.cancel()
+        bridge.login_task = asyncio.create_task(self._periodic_bridge_login(bridge))
+
+    async def _periodic_bridge_check(self, bridge: BridgeState):
+        """Periodically check a bridge's login status."""
+        try:
+            while self._running and not bridge.logged_in:
+                await asyncio.sleep(5)
+
+            check_interval = 300  # 5 minutes
+            response_timeout = 30
+            await self.logger.debug(f'[{bridge.user_id}] Bridge status check started (interval: {check_interval}s)')
+
+            while self._running:
+                for _ in range(check_interval):
+                    if not self._running:
+                        return
+                    await asyncio.sleep(1)
+
+                if not bridge.logged_in or not bridge.dm_room_id:
+                    continue
+
+                try:
+                    bridge.check_responded = False
+                    await self.client.room_send(
+                        room_id=bridge.dm_room_id,
+                        message_type='m.room.message',
+                        content={'msgtype': 'm.text', 'body': bridge.check_command},
+                    )
+                    await self.logger.debug(f'[{bridge.user_id}] Bridge status check: sent "{bridge.check_command}"')
+
+                    for _ in range(response_timeout):
+                        if bridge.check_responded or not self._running:
+                            break
+                        await asyncio.sleep(1)
+
+                    if bridge.check_responded:
+                        await self.logger.debug(f'[{bridge.user_id}] Bridge status check: OK')
+                    else:
+                        await self.logger.info(
+                            f'[{bridge.user_id}] Bridge status check: 无响应, 可能已掉线, 尝试重新登录...'
+                        )
+                        bridge.logged_in = False
+                        self._restart_bridge_login(bridge)
+                except Exception:
+                    await self.logger.error(f'[{bridge.user_id}] Bridge status check error: {traceback.format_exc()}')
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            await self.logger.error(f'[{bridge.user_id}] Bridge status check fatal error: {traceback.format_exc()}')
+
+    async def _handle_relogin_command(self, room_id: str):
+        """Handle !relogin command: logout then re-login all bridges."""
+        if not self._bridges:
+            await self.client.room_send(
+                room_id=room_id,
+                message_type='m.room.message',
+                content={'msgtype': 'm.text', 'body': '没有配置任何桥。'},
+            )
+            return
+
+        lines = ['开始重新登录所有桥...']
+        for bridge in self._bridges:
+            if not bridge.login_command or not bridge.dm_room_id:
+                lines.append(f'[{bridge.user_id}] 跳过（未配置登录命令或无DM房间）')
+                continue
+
+                # Use configured logout command, fallback to deriving from login command
+                logout_cmd = bridge.logout_command or bridge.login_command.replace('login', 'logout')
+            lines.append(f'[{bridge.user_id}] 发送 "{logout_cmd}"...')
+
+            # Cancel existing tasks
+            if bridge.login_task and not bridge.login_task.done():
+                bridge.login_task.cancel()
+            if bridge.check_task and not bridge.check_task.done():
+                bridge.check_task.cancel()
+
+            # Send logout
+            try:
+                await self.client.room_send(
+                    room_id=bridge.dm_room_id,
+                    message_type='m.room.message',
+                    content={'msgtype': 'm.text', 'body': logout_cmd},
+                )
+            except Exception as e:
+                lines.append(f'[{bridge.user_id}] logout 发送失败: {e}')
+
+            await asyncio.sleep(2)
+
+            # Reset state and restart login
+            bridge.logged_in = False
+            self._restart_bridge_login(bridge)
+            lines.append(f'[{bridge.user_id}] 已触发重新登录')
+
+        await self.client.room_send(
+            room_id=room_id,
+            message_type='m.room.message',
+            content={'msgtype': 'm.text', 'body': '\n'.join(lines)},
+        )
+
+    async def _handle_status_command(self, room_id: str):
+        """Handle !status command: show bridge states."""
+        if not self._bridges:
+            await self.client.room_send(
+                room_id=room_id,
+                message_type='m.room.message',
+                content={'msgtype': 'm.text', 'body': '没有配置任何桥。'},
+            )
+            return
+
+        lines = ['桥状态:']
+        for bridge in self._bridges:
+            status = '已登录 ✓' if bridge.logged_in else '未登录 ✗'
+            dm = bridge.dm_room_id or '无'
+            lines.append(f'• {bridge.user_id}: {status} (DM: {dm})')
+        await self.client.room_send(
+            room_id=room_id,
+            message_type='m.room.message',
+            content={'msgtype': 'm.text', 'body': '\n'.join(lines)},
+        )
+
+    async def kill(self) -> bool:
+        self._running = False
+        for bridge in self._bridges:
+            if bridge.login_task and not bridge.login_task.done():
+                bridge.login_task.cancel()
+            if bridge.check_task and not bridge.check_task.done():
+                bridge.check_task.cancel()
+        if self.client:
+            await self.client.close()
+        await self.logger.debug('Matrix adapter stopped')
+        return True
+
+    async def unregister_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
+        ],
+    ):
+        if event_type in self.listeners:
+            del self.listeners[event_type]

src/langbot/pkg/platform/sources/matrix.yaml

@@ -0,0 +1,123 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+metadata:
+  name: matrix
+  label:
+    en_US: Matrix
+    zh_Hans: Matrix
+    zh_Hant: Matrix
+    ja_JP: Matrix
+    th_TH: Matrix
+    vi_VN: Matrix
+    es_ES: Matrix
+  description:
+    en_US: Matrix protocol adapter, supports self-hosted Synapse servers and any Matrix-compatible homeserver
+    zh_Hans: Matrix 协议适配器，支持自建 Synapse 服务器及任何 Matrix 兼容的 Homeserver
+    zh_Hant: Matrix 協議適配器，支持自建 Synapse 伺服器及任何 Matrix 相容的 Homeserver
+    ja_JP: Matrix プロトコルアダプター、セルフホストの Synapse サーバーおよび Matrix 互換のホームサーバーをサポート
+    th_TH: อะแดปเตอร์โปรโตคอล Matrix รองรับเซิร์ฟเวอร์ Synapse ที่โฮสต์เองและ Homeserver ที่เข้ากันได้กับ Matrix
+    vi_VN: Bộ điều hợp giao thức Matrix, hỗ trợ máy chủ Synapse tự lưu trữ và bất kỳ Homeserver tương thích Matrix nào
+    es_ES: Adaptador del protocolo Matrix, compatible con servidores Synapse autoalojados y cualquier Homeserver compatible con Matrix
+  icon: matrix.png
+spec:
+  categories:
+    - global
+    - protocol
+  config:
+    - name: homeserver_url
+      label:
+        en_US: Homeserver URL
+        zh_Hans: Homeserver 地址
+        zh_Hant: Homeserver 地址
+        ja_JP: Homeserver URL
+        th_TH: URL ของ Homeserver
+        vi_VN: URL Homeserver
+        es_ES: URL del Homeserver
+      description:
+        en_US: "The URL of the Matrix homeserver, e.g. http://localhost:8008"
+        zh_Hans: "Matrix Homeserver 的地址，例如 http://localhost:8008"
+      type: string
+      required: true
+      default: "http://localhost:8008"
+    - name: user_id
+      label:
+        en_US: Bot User ID
+        zh_Hans: 机器人用户 ID
+        zh_Hant: 機器人用戶 ID
+        ja_JP: ボットユーザー ID
+        th_TH: ID ผู้ใช้บอท
+        vi_VN: ID người dùng bot
+        es_ES: ID de usuario del bot
+      description:
+        en_US: "The full Matrix user ID, e.g. @bot:localhost"
+        zh_Hans: "完整的 Matrix 用户 ID，例如 @bot:localhost"
+      type: string
+      required: true
+      default: "@langbot:localhost"
+    - name: access_token
+      label:
+        en_US: Access Token
+        zh_Hans: 访问令牌
+        zh_Hant: 訪問令牌
+        ja_JP: アクセストークン
+        th_TH: โทเค็นการเข้าถึง
+        vi_VN: Mã truy cập
+        es_ES: Token de acceso
+      description:
+        en_US: "Access token obtained by logging in via the Matrix client API"
+        zh_Hans: "通过 Matrix Client API 登录获取的访问令牌"
+      type: string
+      required: true
+      default: ""
+    - name: bridge_user_id
+      label:
+        en_US: Bridge Bot User ID (single bridge, legacy)
+        zh_Hans: 桥机器人用户 ID（单桥兼容）
+      description:
+        en_US: "Single bridge bot user ID (legacy). Prefer 'bridges' for multi-bridge. e.g. @discordbot:localhost"
+        zh_Hans: "单桥机器人用户 ID（旧格式兼容）。推荐使用 bridges 配置多桥。例如 @discordbot:localhost"
+      type: string
+      required: false
+      default: ""
+    - name: bridge_login_command
+      label:
+        en_US: Bridge Login Command (single bridge, legacy)
+        zh_Hans: 桥登录命令（单桥兼容）
+      description:
+        en_US: "Login command for single bridge (legacy). e.g. !discord login"
+        zh_Hans: "单桥登录命令（旧格式兼容）。例如 !discord login"
+      type: string
+      required: false
+      default: ""
+    - name: bridge_login_success_keyword
+      label:
+        en_US: Bridge Login Success Keyword (single bridge, legacy)
+        zh_Hans: 桥登录成功关键词（单桥兼容）
+      description:
+        en_US: "Success keyword for single bridge (legacy). e.g. Successfully logged in"
+        zh_Hans: "单桥登录成功关键词（旧格式兼容）。例如 Successfully logged in"
+      type: string
+      required: false
+      default: "Successfully logged in"
+    - name: bridges
+      label:
+        en_US: Bridges Config (Multi-bridge)
+        zh_Hans: 桥配置（多桥）
+      description:
+        en_US: >
+          JSON array of bridge configs. Each bridge: {"user_id": "@bot:host", "login_command": "!xx login",
+          "success_keyword": "logged in", "check_command": "!xx ping"}.
+          Example: [{"user_id":"@discordbot:localhost","login_command":"!discord login","success_keyword":"logged in"},
+          {"user_id":"@telegrambot:localhost","login_command":"!tg login","success_keyword":"logged in"}]
+        zh_Hans: >
+          JSON 数组格式的多桥配置。每个桥: {"user_id": "@bot:host", "login_command": "!xx login",
+          "success_keyword": "logged in", "check_command": "!xx ping"}。
+          示例: [{"user_id":"@discordbot:localhost","login_command":"!discord login","success_keyword":"logged in"},
+          {"user_id":"@telegrambot:localhost","login_command":"!tg login","success_keyword":"logged in"}]
+      type: string
+      required: false
+      default: ""
+execution:
+  python:
+    path: ./matrix.py
+    attr: MatrixAdapter

src/langbot/pkg/platform/sources/qqofficial.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 import typing
+import re
 import asyncio
 import traceback
 
 import datetime
+import time
 
 import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
@@ -15,11 +17,25 @@ from ...utils import image
 from ..logger import EventLogger
 
 
+def _is_base64_data(value: str) -> bool:
+    """Check if a string contains base64-encoded data rather than a URL."""
+    if not value:
+        return False
+    # data: URI scheme (e.g. data:image/png;base64,xxx)
+    if value.startswith('data:'):
+        return True
+    # Only treat as base64 if it doesn't look like a URL/path and has valid base64 chars
+    if value.startswith(('http://', 'https://', '/', './', '../')):
+        return False
+    # Check if it looks like base64 (only valid chars, reasonable length)
+    return bool(re.fullmatch(r'[A-Za-z0-9+/=\s]{20,}', value))
+
+
 class QQOfficialMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
     @staticmethod
     async def yiri2target(message_chain: platform_message.MessageChain):
+        """将 LangBot 消息链转换为 QQ Official 消息格式列表。"""
         content_list = []
-        # 只实现了发文字
         for msg in message_chain:
             if type(msg) is platform_message.Plain:
                 content_list.append(
@@ -28,6 +44,49 @@ class QQOfficialMessageConverter(abstract_platform_adapter.AbstractMessageConver
                         'content': msg.text,
                     }
                 )
+            elif type(msg) is platform_message.Image:
+                url = msg.url if hasattr(msg, 'url') and msg.url else None
+                b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
+                # Some plugins (e.g. MimoTTS) store base64 data in the url field
+                if url and not b64 and _is_base64_data(url):
+                    b64 = url
+                    url = None
+                content_list.append(
+                    {
+                        'type': 'image',
+                        'url': url,
+                        'base64': b64,
+                    }
+                )
+            elif type(msg) is platform_message.Voice:
+                url = msg.url if hasattr(msg, 'url') and msg.url else None
+                b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
+                # Some plugins (e.g. MimoTTS) store base64 data in the url field
+                if url and not b64 and _is_base64_data(url):
+                    b64 = url
+                    url = None
+                content_list.append(
+                    {
+                        'type': 'voice',
+                        'url': url,
+                        'base64': b64,
+                    }
+                )
+            elif type(msg) is platform_message.File:
+                url = msg.url if hasattr(msg, 'url') and msg.url else None
+                b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
+                # Some plugins store base64 data in the url field
+                if url and not b64 and _is_base64_data(url):
+                    b64 = url
+                    url = None
+                content_list.append(
+                    {
+                        'type': 'file',
+                        'url': url,
+                        'base64': b64,
+                        'name': msg.name if hasattr(msg, 'name') else 'file',
+                    }
+                )
 
         return content_list
 
@@ -129,12 +188,19 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
     config: dict
     bot_account_id: str
     bot_uuid: str = None
+    enable_webhook: bool = False
     message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter()
     event_converter: QQOfficialEventConverter = QQOfficialEventConverter()
 
     def __init__(self, config: dict, logger: EventLogger):
+        enable_webhook = config.get('enable-webhook', False)
+
         bot = QQOfficialClient(
-            app_id=config['appid'], secret=config['secret'], token=config['token'], logger=logger, unified_mode=True
+            app_id=config['appid'],
+            secret=config['secret'],
+            token=config['token'],
+            logger=logger,
+            unified_mode=enable_webhook,
         )
 
         super().__init__(
@@ -144,6 +210,13 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
             bot_account_id=config['appid'],
         )
 
+        self.enable_webhook = enable_webhook
+        self._ws_task: asyncio.Task = None
+        self._stream_ctx: dict = {}
+        self._stream_ctx_ts: dict[str, float] = {}
+        self._fallback_text: dict[str, str] = {}
+        self._fallback_text_ts: dict[str, float] = {}
+
     async def reply_message(
         self,
         message_source: platform_events.MessageEvent,
@@ -156,28 +229,18 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
 
         content_list = await QQOfficialMessageConverter.yiri2target(message)
 
-        # 私聊消息
+        # 确定 target_type 和 target_id
+        target_type = None
+        target_id = None
+
         if qq_official_event.t == 'C2C_MESSAGE_CREATE':
-            for content in content_list:
-                if content['type'] == 'text':
-                    await self.bot.send_private_text_msg(
-                        qq_official_event.user_openid,
-                        content['content'],
-                        qq_official_event.d_id,
-                    )
-
-        # 群聊消息
-        if qq_official_event.t == 'GROUP_AT_MESSAGE_CREATE':
-            for content in content_list:
-                if content['type'] == 'text':
-                    await self.bot.send_group_text_msg(
-                        qq_official_event.group_openid,
-                        content['content'],
-                        qq_official_event.d_id,
-                    )
-
-        # 频道群聊
-        if qq_official_event.t == 'AT_MESSAGE_CREATE':
+            target_type = 'c2c'
+            target_id = qq_official_event.user_openid
+        elif qq_official_event.t == 'GROUP_AT_MESSAGE_CREATE':
+            target_type = 'group'
+            target_id = qq_official_event.group_openid
+        elif qq_official_event.t == 'AT_MESSAGE_CREATE':
+            # 频道群聊使用频道 API，暂不支持富媒体
             for content in content_list:
                 if content['type'] == 'text':
                     await self.bot.send_channle_group_text_msg(
@@ -185,9 +248,9 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
                         content['content'],
                         qq_official_event.d_id,
                     )
-
-        # 频道私聊
-        if qq_official_event.t == 'DIRECT_MESSAGE_CREATE':
+            return
+        elif qq_official_event.t == 'DIRECT_MESSAGE_CREATE':
+            # 频道私聊使用频道 API，暂不支持富媒体
             for content in content_list:
                 if content['type'] == 'text':
                     await self.bot.send_channle_private_text_msg(
@@ -195,6 +258,63 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
                         content['content'],
                         qq_official_event.d_id,
                     )
+            return
+
+        # C2C 和群聊：支持文字 + 富媒体
+        for content in content_list:
+            content_type = content.get('type', 'text')
+
+            if content_type == 'text':
+                if target_type == 'c2c':
+                    await self.bot.send_private_text_msg(
+                        target_id,
+                        content['content'],
+                        qq_official_event.d_id,
+                    )
+                elif target_type == 'group':
+                    await self.bot.send_group_text_msg(
+                        target_id,
+                        content['content'],
+                        qq_official_event.d_id,
+                    )
+
+            elif content_type == 'image':
+                file_url = content.get('url')
+                file_data = content.get('base64')
+                if file_url or file_data:
+                    await self.bot.send_image_msg(
+                        target_type,
+                        target_id,
+                        file_url=file_url,
+                        file_data=file_data,
+                        msg_id=qq_official_event.d_id,
+                    )
+
+            elif content_type == 'voice':
+                file_url = content.get('url')
+                file_data = content.get('base64')
+                if file_url or file_data:
+                    await self.bot.send_voice_msg(
+                        target_type,
+                        target_id,
+                        file_url=file_url,
+                        file_data=file_data,
+                        msg_id=qq_official_event.d_id,
+                    )
+
+            elif content_type == 'file':
+                file_url = content.get('url')
+                file_data = content.get('base64')
+                file_name = content.get('name', 'file')
+                if file_url or file_data:
+                    await self.bot.send_file_msg(
+                        target_type,
+                        target_id,
+                        file_url=file_url,
+                        file_data=file_data,
+                        file_name=file_name,
+                        msg_id=qq_official_event.d_id,
+                    )
 
     async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
         pass
@@ -238,17 +358,196 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
         return await self.bot.handle_unified_webhook(request)
 
     async def run_async(self):
-        # 统一 webhook 模式下，不启动独立的 Quart 应用
-        # 保持运行但不启动独立端口
+        if not self.enable_webhook:
+            await self._run_websocket()
+        else:
+            # 统一 webhook 模式下，不启动独立的 Quart 应用
+            async def keep_alive():
+                while True:
+                    await asyncio.sleep(1)
 
-        async def keep_alive():
-            while True:
-                await asyncio.sleep(1)
+            await keep_alive()
 
-        await keep_alive()
+    async def _run_websocket(self):
+        """以 WebSocket 模式运行网关连接"""
+        await self.logger.info('QQ Official adapter starting in WebSocket mode')
+
+        async def on_ready():
+            await self.logger.info('QQ Official WebSocket connected and ready')
+
+        async def on_event(event_type: str, event_data: dict):
+            # 只处理消息事件，忽略 READY/RESUMED 等系统事件
+            message_event_types = {
+                'C2C_MESSAGE_CREATE',
+                'DIRECT_MESSAGE_CREATE',
+                'GROUP_AT_MESSAGE_CREATE',
+                'AT_MESSAGE_CREATE',
+            }
+            if event_type not in message_event_types:
+                return
+            if not isinstance(event_data, dict):
+                await self.logger.warning(f'Event data is not dict, skipping: {event_type} -> {type(event_data)}')
+                return
+            await self.logger.info(f'Processing message event: {event_type}')
+            # 构造与 webhook 模式相同的 payload 结构
+            payload = {'t': event_type, 'd': event_data}
+            message_data = await self.bot.get_message(payload)
+            if message_data:
+                event = QQOfficialEvent.from_payload(message_data)
+                await self.bot._handle_message(event)
+
+        async def on_error(error: Exception):
+            await self.logger.error(f'WebSocket error: {error}')
+            await self.logger.error(f'QQ Official WebSocket error: {error}')
+
+        self._ws_task = asyncio.create_task(self.bot.connect_gateway_loop(on_event, on_ready, on_error))
+        try:
+            await self._ws_task
+        except asyncio.CancelledError:
+            pass
 
     async def kill(self) -> bool:
-        return False
+        if self._ws_task:
+            self._ws_task.cancel()
+            try:
+                await self._ws_task
+            except asyncio.CancelledError:
+                pass
+            self._ws_task = None
+        return True
+
+    # --------------- 流式输出 ---------------
+
+    _STREAM_CTX_TTL = 300  # seconds
+
+    async def _cleanup_stale_streams(self):
+        """Remove stream contexts that have not been updated for more than _STREAM_CTX_TTL seconds."""
+        now = time.time()
+        stale_ids = [mid for mid, ts in self._stream_ctx_ts.items() if now - ts > self._STREAM_CTX_TTL]
+        for mid in stale_ids:
+            self._stream_ctx.pop(mid, None)
+            self._stream_ctx_ts.pop(mid, None)
+        stale_fb = [mid for mid, ts in self._fallback_text_ts.items() if now - ts > self._STREAM_CTX_TTL]
+        for mid in stale_fb:
+            self._fallback_text.pop(mid, None)
+            self._fallback_text_ts.pop(mid, None)
+        if stale_ids or stale_fb:
+            await self.logger.debug(f'Cleaned up {len(stale_ids)} stream contexts, {len(stale_fb)} fallback texts')
+
+    async def is_stream_output_supported(self) -> bool:
+        return self.config.get('enable-stream-reply', False)
+
+    async def create_message_card(self, message_id: str, event: platform_events.MessageEvent) -> bool:
+        source = event.source_platform_object
+        # Streaming API only supports C2C private chat
+        if source.t != 'C2C_MESSAGE_CREATE':
+            return False
+
+        ctx = {
+            'user_openid': source.user_openid,
+            'msg_id': source.d_id,
+            'stream_msg_id': None,
+            'msg_seq': 1,
+            'index': 0,
+            'last_update_ts': 0,
+            'accumulated_text': '',
+            'sent_length': 0,
+            'session_started': False,
+        }
+
+        self._stream_ctx[message_id] = ctx
+        self._stream_ctx_ts[message_id] = time.time()
+        return True
+
+    async def reply_message_chunk(
+        self,
+        message_source: platform_events.MessageEvent,
+        bot_message: dict,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+        is_final: bool = False,
+    ):
+        # Periodically clean up stale stream contexts
+        await self._cleanup_stale_streams()
+        # 提取纯文本内容（当前 chunk 的文本）
+        text_parts = []
+        for msg in message:
+            if type(msg) is platform_message.Plain:
+                text_parts.append(msg.text)
+        chunk_text = '\n\n'.join(text_parts)
+
+        message_id = (
+            bot_message.get('resp_message_id')
+            if isinstance(bot_message, dict)
+            else getattr(bot_message, 'resp_message_id', None)
+        )
+        if not message_id or message_id not in self._stream_ctx:
+            # 非流式场景（如群聊不支持流式），累积文本后一次性回复
+            if chunk_text:
+                self._fallback_text[message_id] = self._fallback_text.get(message_id, '') + chunk_text
+                self._fallback_text_ts[message_id] = time.time()
+            if is_final:
+                full_text = self._fallback_text.pop(message_id, '')
+                if full_text:
+                    fallback_msg = platform_message.MessageChain([platform_message.Plain(text=full_text)])
+                    await self.reply_message(message_source, fallback_msg, quote_origin)
+            return
+
+        ctx = self._stream_ctx[message_id]
+
+        # 累积文本
+        if chunk_text:
+            ctx['accumulated_text'] += chunk_text
+
+        # 未启动会话时，等第一个有内容的 chunk 来建立会话
+        if not ctx['session_started']:
+            if not ctx['accumulated_text']:
+                return
+            # 用第一个 chunk 的文本建立会话（不发 "..." 避免污染前缀）
+            ctx['session_started'] = True
+
+        # 发送内容 = 全量累积文本
+        # QQ API 的 replace 模式不允许修改已下发前缀，所以：
+        # - 首次：发送全部文本，建立会话
+        # - 后续：只能发送新增部分（append 行为）
+        content_to_send = ctx['accumulated_text'][ctx['sent_length'] :]
+        if not content_to_send and not is_final:
+            return
+
+        input_state = 10 if is_final else 1
+
+        # Rate limiting: skip non-final updates if last update was <0.5s ago
+        now = time.time()
+        if not is_final and (now - ctx['last_update_ts']) < 0.5:
+            return
+        ctx['last_update_ts'] = now
+
+        try:
+            resp = await self.bot.send_stream_msg(
+                user_openid=ctx['user_openid'],
+                content=content_to_send,
+                event_id=ctx['msg_id'],
+                msg_id=ctx['msg_id'],
+                msg_seq=ctx['msg_seq'],
+                index=ctx['index'],
+                stream_msg_id=ctx['stream_msg_id'],
+                input_state=input_state,
+            )
+            if resp and isinstance(resp, dict):
+                new_stream_id = resp.get('id')
+                if new_stream_id:
+                    ctx['stream_msg_id'] = new_stream_id
+            ctx['sent_length'] = len(ctx['accumulated_text'])
+            ctx['index'] += 1
+            await self.logger.debug(
+                f'[QQ Official] 流式 chunk 已发送, index={ctx["index"]}, '
+                f'sent_len={ctx["sent_length"]}, is_final={is_final}'
+            )
+        except Exception as e:
+            await self.logger.error(f'Failed to send stream message: {e}')
+
+        if is_final:
+            self._stream_ctx.pop(message_id, None)
 
     def unregister_listener(
         self,

src/langbot/pkg/platform/sources/qqofficial.yaml

@@ -7,9 +7,9 @@ metadata:
     zh_Hans: QQ 官方 API
     zh_Hant: QQ 官方 API
   description:
-    en_US: QQ Official API (Webhook)
-    zh_Hans: QQ 官方 API (Webhook)，需要公网地址以接收消息推送，请查看文档了解使用方式
-    zh_Hant: QQ 官方 API (Webhook)，需要公網地址以接收訊息推送，請查看文件了解使用方式
+    en_US: QQ Official API (Webhook / WebSocket)
+    zh_Hans: QQ 官方 API，支持 Webhook 和 WebSocket 两种连接模式
+    zh_Hant: QQ 官方 API，支援 Webhook 和 WebSocket 兩種連線模式
   icon: qqofficial.svg
 spec:
   categories:
@@ -19,18 +19,6 @@ spec:
     en: https://link.langbot.app/en/platforms/qqofficial
     ja: https://link.langbot.app/ja/platforms/qqofficial
   config:
-    - name: webhook_url
-      label:
-        en_US: Webhook Callback URL
-        zh_Hans: Webhook 回调地址
-        zh_Hant: Webhook 回調地址
-      description:
-        en_US: Copy this URL and paste it into your QQ Official API webhook configuration
-        zh_Hans: 复制此地址并粘贴到 QQ 官方 API 的 Webhook 配置中
-        zh_Hant: 複製此地址並貼到 QQ 官方 API 的 Webhook 設定中
-      type: webhook-url
-      required: false
-      default: ""
     - name: appid
       label:
         en_US: App ID
@@ -55,6 +43,46 @@ spec:
       type: string
       required: true
       default: ""
+    - name: enable-webhook
+      label:
+        en_US: Enable Webhook Mode
+        zh_Hans: 启用Webhook模式
+        zh_Hant: 啟用 Webhook 模式
+      description:
+        en_US: If enabled, the bot will use webhook mode to receive messages. Otherwise, it will use WebSocket mode
+        zh_Hans: 如果启用，机器人将使用 Webhook 模式接收消息。否则，将使用 WebSocket 模式
+        zh_Hant: 如果啟用，機器人將使用 Webhook 模式接收訊息。否則，將使用 WebSocket 模式
+      type: boolean
+      required: true
+      default: false
+    - name: enable-stream-reply
+      label:
+        en_US: Enable Stream Reply Mode
+        zh_Hans: 启用流式回复模式
+        zh_Hant: 啟用串流回覆模式
+      description:
+        en_US: If enabled, the bot will use streaming mode to reply messages (C2C only)
+        zh_Hans: 如果启用，机器人将使用流式方式回复消息（仅私聊）
+        zh_Hant: 如果啟用，機器人將使用串流方式回覆訊息（僅私聊）
+      type: boolean
+      required: true
+      default: false
+    - name: webhook_url
+      label:
+        en_US: Webhook Callback URL
+        zh_Hans: Webhook 回调地址
+        zh_Hant: Webhook 回調地址
+      description:
+        en_US: Copy this URL and paste it into your QQ Official API webhook configuration
+        zh_Hans: 复制此地址并粘贴到 QQ 官方 API 的 Webhook 配置中
+        zh_Hant: 複製此地址並貼到 QQ 官方 API 的 Webhook 設定中
+      type: webhook-url
+      required: false
+      default: ""
+      show_if:
+        field: enable-webhook
+        operator: eq
+        value: true
 execution:
   python:
     path: ./qqofficial.py

src/langbot/pkg/platform/sources/web_page_bot.yaml

@@ -0,0 +1,177 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+metadata:
+  name: web_page_bot
+  label:
+    en_US: "Page Bot"
+    zh_Hans: "页面机器人"
+    zh_Hant: "頁面機器人"
+    ja_JP: "ページボット"
+    th_TH: "บอทหน้าเว็บ"
+    vi_VN: "Bot trang web"
+    es_ES: "Bot de página"
+  description:
+    en_US: "Embed a chat widget on any website with a simple script tag"
+    zh_Hans: "通过一行脚本标签将聊天组件嵌入到任何网站"
+    zh_Hant: "透過一行腳本標籤將聊天元件嵌入到任何網站"
+    ja_JP: "シンプルなスクリプトタグで任意のウェブサイトにチャットウィジェットを埋め込みます"
+    th_TH: "ฝังวิดเจ็ตแชทในเว็บไซต์ใดก็ได้ด้วยแท็กสคริปต์"
+    vi_VN: "Nhúng widget trò chuyện vào bất kỳ trang web nào bằng thẻ script"
+    es_ES: "Incrusta un widget de chat en cualquier sitio web con una etiqueta de script"
+  icon: "webpage.webp"
+spec:
+  categories:
+    - popular
+  config:
+    - name: title
+      label:
+        en_US: Widget Title
+        zh_Hans: 组件标题
+        zh_Hant: 元件標題
+        ja_JP: ウィジェットタイトル
+        th_TH: ชื่อวิดเจ็ต
+        vi_VN: Tiêu đề widget
+        es_ES: Título del widget
+      description:
+        en_US: The title displayed in the chat widget header
+        zh_Hans: 显示在聊天组件顶部的标题
+        zh_Hant: 顯示在聊天元件頂部的標題
+        ja_JP: チャットウィジェットのヘッダーに表示されるタイトル
+        th_TH: ชื่อที่แสดงในส่วนหัวของวิดเจ็ตแชท
+        vi_VN: Tiêu đề hiển thị trong đầu widget trò chuyện
+        es_ES: El título que se muestra en el encabezado del widget de chat
+      type: string
+      required: false
+      default: "LangBot"
+    - name: bubble_icon
+      label:
+        en_US: Bubble Icon
+        zh_Hans: 气泡图标
+        zh_Hant: 氣泡圖示
+        ja_JP: バブルアイコン
+        th_TH: ไอคอนบับเบิล
+        vi_VN: Biểu tượng bong bóng
+        es_ES: Icono de burbuja
+        ru_RU: Иконка пузырька
+      description:
+        en_US: "Icon displayed on the floating chat bubble"
+        zh_Hans: "浮动聊天气泡上显示的图标"
+      type: select
+      required: false
+      default: "logo"
+      options:
+        - name: "logo"
+          label:
+            en_US: "LangBot Logo"
+            zh_Hans: "LangBot 图标"
+        - name: "chat"
+          label:
+            en_US: "Chat Bubble"
+            zh_Hans: "聊天气泡"
+        - name: "robot"
+          label:
+            en_US: "Robot"
+            zh_Hans: "机器人"
+        - name: "headset"
+          label:
+            en_US: "Headset"
+            zh_Hans: "客服耳机"
+        - name: "sparkle"
+          label:
+            en_US: "Sparkle"
+            zh_Hans: "星光"
+        - name: "message"
+          label:
+            en_US: "Message"
+            zh_Hans: "消息"
+    - name: language
+      label:
+        en_US: Widget Language
+        zh_Hans: 组件语言
+        zh_Hant: 元件語言
+        ja_JP: ウィジェット言語
+        th_TH: ภาษาวิดเจ็ต
+        vi_VN: Ngôn ngữ widget
+        es_ES: Idioma del widget
+        ru_RU: Язык виджета
+      description:
+        en_US: "Display language of the chat widget"
+        zh_Hans: "聊天组件的显示语言"
+        zh_Hant: "聊天元件的顯示語言"
+        ja_JP: "チャットウィジェットの表示言語"
+        th_TH: "ภาษาแสดงผลของวิดเจ็ตแชท"
+        vi_VN: "Ngôn ngữ hiển thị của widget trò chuyện"
+        es_ES: "Idioma de visualización del widget de chat"
+        ru_RU: "Язык отображения виджета чата"
+      type: select
+      required: false
+      default: "en_US"
+      options:
+        - name: "en_US"
+          label:
+            en_US: "English"
+        - name: "zh_Hans"
+          label:
+            en_US: "简体中文"
+        - name: "zh_Hant"
+          label:
+            en_US: "繁體中文"
+        - name: "ja_JP"
+          label:
+            en_US: "日本語"
+        - name: "es_ES"
+          label:
+            en_US: "Español"
+        - name: "ru_RU"
+          label:
+            en_US: "Русский"
+        - name: "th_TH"
+          label:
+            en_US: "ไทย"
+        - name: "vi_VN"
+          label:
+            en_US: "Tiếng Việt"
+    - name: embed_code
+      label:
+        en_US: Embed Code
+        zh_Hans: 嵌入代码
+        zh_Hant: 嵌入代碼
+        ja_JP: 埋め込みコード
+        th_TH: โค้ดฝังตัว
+        vi_VN: Mã nhúng
+        es_ES: Código de incrustación
+      description:
+        en_US: "Copy this code and paste it into your website HTML. The code will be generated after saving."
+        zh_Hans: "复制此代码并粘贴到你的网站 HTML 中。保存后将自动生成。"
+        zh_Hant: "複製此代碼並貼到你的網站 HTML 中。儲存後將自動生成。"
+        ja_JP: "このコードをコピーしてウェブサイトのHTMLに貼り付けてください。保存後に自動生成されます。"
+        th_TH: "คัดลอกโค้ดนี้และวางในHTML ของเว็บไซต์ของคุณ จะสร้างอัตโนมัติหลังจากบันทึก"
+        vi_VN: "Sao chép mã này và dán vào HTML trang web của bạn. Mã sẽ được tạo tự động sau khi lưu."
+        es_ES: "Copia este código y pégalo en el HTML de tu sitio web. El código se generará después de guardar."
+      type: embed-code
+      required: false
+      default: ""
+    - name: turnstile_site_key
+      label:
+        en_US: Turnstile Site Key
+        zh_Hans: Turnstile 站点密钥
+      description:
+        en_US: "Cloudflare Turnstile site key for bot protection. Get it from the Cloudflare dashboard (Turnstile > Add Site). Leave empty to disable."
+        zh_Hans: "Cloudflare Turnstile 站点密钥，用于防止机器人滥用。在 Cloudflare 控制台（Turnstile > 添加站点）中获取。留空则不启用。"
+      type: string
+      required: false
+      default: ""
+    - name: turnstile_secret_key
+      label:
+        en_US: Turnstile Secret Key
+        zh_Hans: Turnstile 服务端密钥
+      description:
+        en_US: "Cloudflare Turnstile secret key for server-side token verification. Found alongside the site key in the Cloudflare dashboard. Required if site key is set."
+        zh_Hans: "Cloudflare Turnstile 服务端密钥，用于服务端验证令牌。与站点密钥一起在 Cloudflare 控制台中获取。设置了站点密钥时必填。"
+      type: string
+      required: false
+      default: ""
+execution:
+  python:
+    path: "web_page_bot_adapter.py"
+    attr: "WebPageBotAdapter"

src/langbot/pkg/platform/sources/web_page_bot_adapter.py

@@ -0,0 +1,97 @@
+"""Web Page Bot adapter - lightweight adapter for embeddable chat widget"""
+
+import typing
+
+import pydantic
+
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+
+
+class WebPageBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
+    """Lightweight adapter for the embeddable page bot.
+
+    This adapter does not handle messages itself. The actual WebSocket
+    communication is handled by the singleton websocket_proxy_bot.
+    This adapter stores event listeners so that RuntimeBot can register
+    its handlers, which are then called by the websocket adapter when
+    a message arrives for this bot's pipeline.
+
+    Message sending/replying is delegated to the websocket_proxy_bot's
+    adapter so that replies are actually delivered over the WebSocket
+    connection while the dashboard correctly shows this adapter's name.
+    """
+
+    listeners: dict = pydantic.Field(default_factory=dict, exclude=True)
+    _ws_adapter: typing.Any = None
+
+    class Config:
+        arbitrary_types_allowed = True
+        # Allow private attributes
+        underscore_attrs_are_private = True
+
+    def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger, **kwargs):
+        super().__init__(config=config, logger=logger, **kwargs)
+
+    def set_ws_adapter(self, ws_adapter) -> None:
+        """Set the underlying WebSocket adapter used for actual message delivery."""
+        object.__setattr__(self, '_ws_adapter', ws_adapter)
+
+    async def send_message(
+        self,
+        target_type: str,
+        target_id: str,
+        message: platform_message.MessageChain,
+    ) -> dict:
+        if self._ws_adapter is not None:
+            return await self._ws_adapter.send_message(target_type, target_id, message)
+        return {}
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ) -> dict:
+        if self._ws_adapter is not None:
+            return await self._ws_adapter.reply_message(message_source, message, quote_origin)
+        return {}
+
+    async def reply_message_chunk(
+        self,
+        message_source: platform_events.MessageEvent,
+        bot_message,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+        is_final: bool = False,
+    ) -> dict:
+        if self._ws_adapter is not None:
+            return await self._ws_adapter.reply_message_chunk(
+                message_source, bot_message, message, quote_origin, is_final
+            )
+        return {}
+
+    def register_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        func: typing.Callable,
+    ):
+        self.listeners[event_type] = func
+
+    def unregister_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        func: typing.Callable,
+    ):
+        self.listeners.pop(event_type, None)
+
+    async def is_muted(self, group_id: int) -> bool:
+        return False
+
+    async def run_async(self):
+        pass
+
+    async def kill(self):
+        pass

src/langbot/pkg/platform/sources/websocket_adapter.py

@@ -312,7 +312,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
 
     async def _process_image_components(self, message_chain_obj: list):
         """
-        处理消息链中的图片组件，将path转换为base64
+        处理消息链中的图片和文件组件，将path转换为base64
 
         Args:
             message_chain_obj: 消息链对象列表
@@ -322,16 +322,18 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
         storage_mgr = self.ap.storage_mgr
 
         for component in message_chain_obj:
-            if component.get('type') == 'Image' and component.get('path'):
-                try:
-                    # 从storage读取文件
-                    file_content = await storage_mgr.storage_provider.load(component['path'])
+            comp_type = component.get('type', '')
+            comp_path = component.get('path', '')
 
-                    # 转换为base64
+            if not comp_path:
+                continue
+
+            if comp_type == 'Image':
+                try:
+                    file_content = await storage_mgr.storage_provider.load(comp_path)
                     base64_str = base64.b64encode(file_content).decode('utf-8')
 
-                    # 添加data URI前缀（根据文件扩展名判断MIME类型）
-                    file_key = component['path']
+                    file_key = comp_path
                     if file_key.lower().endswith(('.jpg', '.jpeg')):
                         mime_type = 'image/jpeg'
                     elif file_key.lower().endswith('.png'):
@@ -341,19 +343,19 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                     elif file_key.lower().endswith('.webp'):
                         mime_type = 'image/webp'
                     else:
-                        mime_type = 'image/png'  # 默认
+                        mime_type = 'image/png'
 
                     component['base64'] = f'data:{mime_type};base64,{base64_str}'
-                    await storage_mgr.storage_provider.delete(component['path'])
+                    await storage_mgr.storage_provider.delete(comp_path)
                     component['path'] = ''
-                    # 保留path字段用于后端处理，前端使用base64显示
                 except Exception as e:
-                    await self.logger.error(f'加载图片文件失败 {component["path"]}: {e}')
+                    await self.logger.error(f'Failed to load image file {comp_path}: {e}')
 
     async def handle_websocket_message(
         self,
         connection: WebSocketConnection,
         message_data: dict,
+        owner_bot=None,
     ):
         """
         处理从WebSocket接收的消息
@@ -366,6 +368,8 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
             message_data: 消息数据，包含:
                 - message: 消息链
                 - stream: 是否启用流式输出 (可选，默认True)
+            owner_bot: Optional RuntimeBot that owns this pipeline (e.g. a web_page_bot).
+                       When provided, its identity is used for logging and session tracking.
         """
         pipeline_uuid = connection.pipeline_uuid
         session_type = connection.session_type
@@ -491,12 +495,26 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                 sender=sender, message_chain=message_chain, time=datetime.now().timestamp()
             )
 
-        # 设置流水线UUID
+        # 设置流水线UUID (proxy bot always needs it for reply_message routing)
         self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
+        if owner_bot is not None:
+            owner_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
 
-        # 异步触发事件处理（不等待结果）
-        if event.__class__ in self.listeners:
-            asyncio.create_task(self.listeners[event.__class__](event, self))
+        # 异步触发事件处理
+        # Use owner_bot's listeners if available, otherwise fall back to proxy bot
+        listeners = (
+            owner_bot.adapter.listeners
+            if (owner_bot and hasattr(owner_bot.adapter, 'listeners') and owner_bot.adapter.listeners)
+            else self.listeners
+        )
+        # Pass owner_bot's adapter so that downstream logging / dashboard
+        # attributes the message to the correct bot adapter name.
+        # Wire the ws adapter into the owner so replies are actually delivered.
+        if owner_bot and hasattr(owner_bot.adapter, 'set_ws_adapter'):
+            owner_bot.adapter.set_ws_adapter(self)
+        callback_adapter = owner_bot.adapter if (owner_bot and hasattr(owner_bot, 'adapter')) else self
+        if event.__class__ in listeners:
+            asyncio.create_task(listeners[event.__class__](event, callback_adapter))
 
     def get_websocket_messages(self, pipeline_uuid: str, session_type: str) -> list[dict]:
         """获取消息历史"""