diff --git a/docs/event-based-agents/adapters/00-index.md b/docs/event-based-agents/adapters/00-index.md
index c9467c03..2ee9ccab 100644
--- a/docs/event-based-agents/adapters/00-index.md
+++ b/docs/event-based-agents/adapters/00-index.md
@@ -23,6 +23,7 @@ Current acceptance report: [EBA Adapter Acceptance Report](./acceptance-report.m
| WeCom | Migrated; private text plugin E2E verified, media/group gaps remain | [WeCom](./wecom.md) |
| WeComBot | Migrated; private text and outbound/API plugin E2E verified, feedback/group gaps remain | [WeComBot](./wecombot.md) |
| Official Account | Migrated; private text plugin E2E verified, proactive outbound not supported | [Official Account](./officialaccount.md) |
+| QQ Official API | Migrated; WebSocket inbound reached LangBot, model config blocked reply | [QQ Official API](./qqofficial.md) |
## Documentation Checklist
diff --git a/docs/event-based-agents/adapters/acceptance-report.md b/docs/event-based-agents/adapters/acceptance-report.md
index b115b322..b9553b9b 100644
--- a/docs/event-based-agents/adapters/acceptance-report.md
+++ b/docs/event-based-agents/adapters/acceptance-report.md
@@ -13,6 +13,7 @@ Scope:
- `wecombot-eba`
- `wecomcs-eba`
- `officialaccount-eba`
+- `qqofficial-eba`
This report follows `acceptance-checklist.md`. Evidence levels are intentionally strict:
@@ -36,6 +37,7 @@ This report follows `acceptance-checklist.md`. Evidence levels are intentionally
| WeComBot | Partial EBA acceptance | WeCom AI Bot is split into the EBA directory with WebSocket long connection mode and optional webhook mode, EBA message/feedback/platform-specific conversion, cache-backed common APIs, platform API map, unit tests, and a direct live probe. Private text, outbound component sweep, safe common APIs, and all declared WeComBot platform APIs reached `EBAEventProbe`; group, real inbound media, and feedback callback evidence remain pending. |
| WeCom Customer Service | Partial EBA acceptance | WeCom Customer Service is split into the EBA directory with manifest, converters, API mixin, platform API map, unit tests, docs, and a direct live probe scaffold. Real WeChat customer-side UI text reached `EBAEventProbe`; plugin outbound text/image and safe cache-backed common APIs passed. Inbound media and platform-specific API live coverage remain pending; later fallback text sends were blocked by WeCom `95001 send msg count limit`. |
| Official Account | Partial EBA acceptance | WeChat Official Account is split into the EBA directory with manifest, converters, cache-backed safe APIs, platform API map, unit tests, and a direct live probe scaffold. Real WeChat Official Account UI private text reached `EBAEventProbe`; safe cache-backed common APIs and declared platform APIs passed. Proactive outbound `send_message` is not supported because replies must be tied to inbound webhook windows; inbound image/voice live UI evidence remains pending. |
+| QQ Official API | Partial EBA acceptance | QQ Official API is split into the EBA directory with manifest, converters, cache-backed safe APIs, platform API map, unit tests, docs, and a direct live probe scaffold. A real WebSocket-mode QQ Official bot reached the LangBot pipeline on `dev.rockchin.top`; reply/outbound evidence is blocked by the test model provider returning `model_not_found` for `deepseek-v3`. |
Telegram and DingTalk now have real user-side UI image/file upload evidence in plugin JSONL. Discord and aiocqhttp do not yet have real UI inbound image/file evidence.
@@ -55,6 +57,7 @@ Telegram and DingTalk now have real user-side UI image/file upload evidence in p
| Lark / Feishu partial live | Feishu Mac, LangBot organization `LangBotDev` private chat | `data/temp/lark-plugin-e2e-ws.jsonl` |
| WeCom Customer Service | WeChat customer-side UI, `客服消息 -> 浪波智能客服` on `dev.rockchin.top` | `/home/wgc/LangBotxg/LangBotEbaTest/data/temp/wecomcs_eba_plugin_probe.jsonl` |
| Official Account | WeChat desktop client, subscribed Official Account on `dev.rockchin.top` | `/home/wgc/LangBotxg/LangBotEbaTest/data/temp/officialaccount_eba_plugin_probe.jsonl` |
+| QQ Official API unit | local mocked QQ Official client paths | `tests/unit_tests/platform/test_qqofficial_eba_adapter.py` |
All plugin runs used SDK standalone runtime ports `5400/5401`, LangBot `--standalone-runtime`, and the real plugin at `langbot-plugin-demo/EBAEventProbe`.
diff --git a/docs/event-based-agents/adapters/qqofficial.md b/docs/event-based-agents/adapters/qqofficial.md
new file mode 100644
index 00000000..ee72efef
--- /dev/null
+++ b/docs/event-based-agents/adapters/qqofficial.md
@@ -0,0 +1,114 @@
+# QQOfficial EBA Adapter
+
+Adapter directory: `src/langbot/pkg/platform/adapters/qqofficial/`
+
+Manifest name: `qqofficial-eba`
+
+Status: partial migration. The EBA adapter structure, manifest, converters, cache-backed safe APIs, platform API map, unit tests, and direct live probe scaffold are in place. A real QQ Official WebSocket bot on `dev.rockchin.top` received an inbound user message and drove LangBot into the normal pipeline path; the response path was blocked by the test environment model service returning `model_not_found` for `deepseek-v3`.
+
+## Config
+
+| Field | Required | Notes |
+| --- | --- | --- |
+| `appid` | yes | QQ Official app ID. |
+| `secret` | yes | QQ Official app secret. |
+| `token` | yes | QQ Official callback token. |
+| `enable-webhook` | yes | Uses LangBot unified webhook when true; otherwise uses the QQ WebSocket gateway. |
+| `enable-stream-reply` | yes | Enables C2C streaming replies when supported by the QQ Official endpoint. |
+| `webhook_url` | no | Generated by LangBot and copied into the QQ Official callback settings in webhook mode. |
+
+## Events
+
+| Event | Evidence | Notes |
+| --- | --- | --- |
+| `message.received` | adapter-live, unit | `C2C_MESSAGE_CREATE`, `DIRECT_MESSAGE_CREATE`, `GROUP_AT_MESSAGE_CREATE`, and `AT_MESSAGE_CREATE` map to common `MessageReceivedEvent`. A real WebSocket-mode QQ Official bot reached the LangBot pipeline on `dev.rockchin.top`; plugin JSONL evidence remains pending. |
+| `platform.specific` | unit, blocked | Unmapped gateway events are emitted as structured `PlatformSpecificEvent`; live evidence is pending. |
+
+## Common APIs
+
+| API | Evidence | Notes |
+| --- | --- | --- |
+| `send_message` | unit, blocked | Sends private C2C, group, and text-only channel messages through the existing QQ Official client. Live outbound UI verification is pending because the test pipeline failed before producing a bot response. |
+| `reply_message` | unit, blocked | Replies using the source `QQOfficialEvent` message ID when available. Live reply was blocked by the test environment model service returning `model_not_found`. |
+| `get_message` | unit | Returns cached inbound `MessageReceivedEvent`. |
+| `get_user_info` | unit | Returns cached inbound sender. |
+| `get_friend_list` | unit | Returns cached private senders. |
+| `get_group_info` | unit | Returns cached group/channel metadata from inbound events. |
+| `get_group_member_info` | unit | Returns cached group sender as a common member. |
+| `get_group_member_list` | unit | Returns cached group members observed by the adapter. |
+| `call_platform_api` | unit, blocked | Safe diagnostic actions are implemented; live calls are pending credentials. |
+
+## Platform APIs
+
+| Action | Evidence | Notes |
+| --- | --- | --- |
+| `check_access_token` | unit, blocked | Calls the existing client token check. |
+| `refresh_access_token` | unit, blocked | Forces token refresh. |
+| `get_gateway_url` | unit, blocked | Fetches the WebSocket gateway URL. |
+| `get_mode` | unit | Returns webhook and stream-reply mode. |
+
+## Components
+
+| Receive Component | Evidence | Notes |
+| --- | --- | --- |
+| `Source` | unit | Uses QQ message/event IDs and timestamp. |
+| `Plain` | unit | Preserves text content. |
+| `At` | unit | Group and channel mention events insert an adapter bot mention marker. |
+| `Image` | unit | QQ image attachment URL is converted to common `Image`; falls back to URL if download fails. |
+| `Unknown` | unit | Unsupported/empty native payloads become `Unknown`. |
+| `Voice`, `File`, `Quote`, `Face`, `Forward`, mixed chain | blocked | Current native parser only exposes text and image attachments; live endpoint behavior still needs verification. |
+
+| Send Component | Evidence | Notes |
+| --- | --- | --- |
+| `Plain` | unit, blocked | Sends through private, group, or channel text APIs. |
+| `At`, `AtAll` | unit, blocked | Converted to readable mention text. |
+| `Image` | unit, blocked | Sends through the QQ Official rich media upload/send path for C2C and group targets. |
+| `Voice` | unit, blocked | Sends through the QQ Official rich media upload/send path for C2C and group targets. |
+| `File` | unit, blocked | Sends through the QQ Official rich media upload/send path for C2C and group targets. |
+| `Quote`, `Forward`, mixed chain | unit, blocked | Flattened to ordered send payloads where possible. |
+| `Face` | not-supported | No common QQ Official face mapping is implemented. |
+
+## Verification Record
+
+Test date: 2026-06-02
+
+Endpoint/simulator: `dev.rockchin.top` with a real QQ Official WebSocket bot (`qqofficial-eba`, bot UUID `80a5560b-52b1-40e7-b7d6-4a2341eb4780`) and LangBot running from `/home/wgc/LangBotxg/LangBotEbaTest`.
+
+Observed evidence:
+
+- The QQ Official WebSocket bot was enabled with `enable-webhook=false`.
+- A real user message reached LangBot and entered the standard pipeline path.
+- The response path stopped at the model layer with `model_not_found` for `deepseek-v3`; this is a model/provider configuration issue, not an adapter conversion failure.
+- `qq-webhook.langbot.dev` was temporarily routed through Caddy to `127.0.0.1:5301` for webhook checks, but the observed EBA bot used WebSocket mode.
+
+Standalone runtime command:
+
+```bash
+cd langbot-plugin-sdk
+uv run python -m langbot_plugin.cli.__init__ rt --debug-only --ws-control-port 5400 --ws-debug-port 5401 --skip-deps-check
+```
+
+Probe plugin: `data/plugins/LangBot__EBAEventProbe` when live credentials are available.
+
+Adapter live probe:
+
+```bash
+uv run python -m py_compile tests/e2e/live_qqofficial_eba_probe.py
+QQOFFICIAL_APPID=... QQOFFICIAL_SECRET=... QQOFFICIAL_TOKEN=... uv run python tests/e2e/live_qqofficial_eba_probe.py
+```
+
+Webhook-mode probe:
+
+```bash
+QQOFFICIAL_APPID=... QQOFFICIAL_SECRET=... QQOFFICIAL_TOKEN=... uv run python tests/e2e/live_qqofficial_eba_probe.py --webhook --host 0.0.0.0 --port 5312
+```
+
+Evidence JSONL path: `data/temp/qqofficial_eba_probe.jsonl` for direct adapter live probe; plugin E2E evidence should use `data/temp/qqofficial_eba_plugin_probe.jsonl`.
+
+Destructive operations: none implemented.
+
+Blocked items:
+
+- `plugin-e2e-ui`: standalone probe plugin JSONL evidence is still pending; the observed live run reached LangBot core/pipeline but was not recorded by the EBA probe plugin.
+- `plugin-e2e-outbound`: waiting for visible QQ client verification of plugin `send_message`/`reply_message` output after a working model/provider is configured.
+- Inbound non-text media and platform lifecycle events require endpoint evidence before they can be marked complete.
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/__init__.py b/src/langbot/pkg/platform/adapters/qqofficial/__init__.py
new file mode 100644
index 00000000..d87bfc1d
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/__init__.py
@@ -0,0 +1,6 @@
+"""QQ Official API EBA platform adapter."""
+
+from langbot.pkg.platform.adapters.qqofficial.adapter import QQOfficialAdapter
+
+__all__ = ['QQOfficialAdapter']
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py
new file mode 100644
index 00000000..15b2ed0d
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py
@@ -0,0 +1,400 @@
+from __future__ import annotations
+
+import asyncio
+import time
+import traceback
+import typing
+
+import pydantic
+
+from langbot.libs.qq_official_api.api import QQOfficialClient
+from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
+from langbot.pkg.platform.adapters.qqofficial.api_impl import QQOfficialAPIMixin
+from langbot.pkg.platform.adapters.qqofficial.errors import NotSupportedError
+from langbot.pkg.platform.adapters.qqofficial.event_converter import QQOfficialEventConverter
+from langbot.pkg.platform.adapters.qqofficial.message_converter import QQOfficialMessageConverter
+from langbot.pkg.platform.adapters.qqofficial.platform_api import PLATFORM_API_MAP
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter):
+ bot: typing.Any = pydantic.Field(exclude=True)
+
+ message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter()
+ event_converter: QQOfficialEventConverter = QQOfficialEventConverter()
+
+ config: dict
+ bot_uuid: str | None = None
+ enable_webhook: bool = False
+ listeners: dict[
+ typing.Type[platform_events.Event],
+ typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None],
+ ] = {}
+ _message_cache: dict[str, platform_events.MessageReceivedEvent] = {}
+ _user_cache: dict[str, platform_entities.User] = {}
+ _group_cache: dict[str, platform_entities.UserGroup] = {}
+ _member_cache: dict[tuple[str, str], platform_entities.UserGroupMember] = {}
+ _stream_ctx: dict[str, dict] = {}
+ _stream_ctx_ts: dict[str, float] = {}
+ _fallback_text: dict[str, str] = {}
+ _fallback_text_ts: dict[str, float] = {}
+ _ws_task: asyncio.Task | None = None
+
+ _STREAM_CTX_TTL = 300
+
+ class Config:
+ arbitrary_types_allowed = True
+
+ def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
+ required_keys = ['appid', 'secret', 'token']
+ missing_keys = [key for key in required_keys if not config.get(key)]
+ if missing_keys:
+ raise Exception(f'QQOfficial EBA adapter missing config: {missing_keys}')
+
+ enable_webhook = config.get('enable-webhook', config.get('enable_webhook', False))
+ bot = QQOfficialClient(
+ app_id=config['appid'],
+ secret=config['secret'],
+ token=config['token'],
+ logger=logger,
+ unified_mode=enable_webhook,
+ )
+ super().__init__(
+ config=config,
+ logger=logger,
+ bot=bot,
+ bot_account_id=config['appid'],
+ bot_uuid=None,
+ enable_webhook=enable_webhook,
+ listeners={},
+ _message_cache={},
+ _user_cache={},
+ _group_cache={},
+ _member_cache={},
+ _stream_ctx={},
+ _stream_ctx_ts={},
+ _fallback_text={},
+ _fallback_text_ts={},
+ _ws_task=None,
+ )
+ self._register_native_handlers()
+
+ def set_bot_uuid(self, bot_uuid: str):
+ self.bot_uuid = bot_uuid
+
+ def get_supported_events(self) -> list[str]:
+ return [
+ 'message.received',
+ 'platform.specific',
+ ]
+
+ def get_supported_apis(self) -> list[str]:
+ return [
+ 'send_message',
+ 'reply_message',
+ 'get_message',
+ 'get_user_info',
+ 'get_friend_list',
+ 'get_group_info',
+ 'get_group_member_list',
+ 'get_group_member_info',
+ 'call_platform_api',
+ ]
+
+ async def send_message(
+ self,
+ target_type: str,
+ target_id: str,
+ message: platform_message.MessageChain,
+ ) -> platform_events.MessageResult:
+ raw = await self._send_content_list(str(target_type), str(target_id), await QQOfficialMessageConverter.yiri2target(message))
+ return platform_events.MessageResult(raw={'results': raw})
+
+ async def reply_message(
+ self,
+ message_source: platform_events.MessageEvent,
+ message: platform_message.MessageChain,
+ quote_origin: bool = False,
+ ) -> platform_events.MessageResult:
+ source = await QQOfficialEventConverter.yiri2target(message_source)
+ if not isinstance(source, QQOfficialEvent):
+ raise ValueError('QQOfficial reply_message requires a QQOfficialEvent source object')
+ target_type, target_id = self._reply_target(source)
+ raw = await self._send_content_list(
+ target_type,
+ target_id,
+ await QQOfficialMessageConverter.yiri2target(message),
+ msg_id=source.d_id,
+ )
+ return platform_events.MessageResult(message_id=source.d_id or source.id, raw={'results': raw})
+
+ async def call_platform_api(self, action: str, params: dict = {}) -> dict:
+ handler = PLATFORM_API_MAP.get(action)
+ if handler is None:
+ raise NotSupportedError(f'call_platform_api:{action}')
+ return await handler(self, dict(params or {}))
+
+ 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
+
+ def unregister_listener(
+ self,
+ event_type: typing.Type[platform_events.Event],
+ callback: typing.Callable[
+ [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
+ ],
+ ):
+ registered = self.listeners.get(event_type)
+ if registered is callback:
+ self.listeners.pop(event_type, None)
+
+ async def handle_unified_webhook(self, bot_uuid: str, path: str, request):
+ return await self.bot.handle_unified_webhook(request)
+
+ async def run_async(self):
+ if self.enable_webhook:
+ await self.logger.info('QQ Official EBA adapter running in unified webhook mode')
+ while True:
+ await asyncio.sleep(1)
+ else:
+ await self._run_websocket()
+
+ async def kill(self) -> bool:
+ if self._ws_task:
+ self._ws_task.cancel()
+ try:
+ await self._ws_task
+ except asyncio.CancelledError:
+ pass
+ self._ws_task = None
+ return True
+
+ async def is_muted(self, group_id: int | None = None) -> bool:
+ return False
+
+ async def is_stream_output_supported(self) -> bool:
+ return bool(self.config.get('enable-stream-reply') or self.config.get('enable_stream_reply'))
+
+ async def create_message_card(self, message_id: str, event: platform_events.MessageEvent) -> bool:
+ source = event.source_platform_object
+ if not isinstance(source, QQOfficialEvent) or source.t != 'C2C_MESSAGE_CREATE':
+ return False
+ self._stream_ctx[message_id] = {
+ '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_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,
+ ):
+ await self._cleanup_stale_streams()
+ chunk_text = '\n\n'.join(component.text for component in message if isinstance(component, platform_message.Plain))
+ 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:
+ await self.reply_message(message_source, platform_message.MessageChain([platform_message.Plain(text=full_text)]), quote_origin)
+ return
+
+ ctx = self._stream_ctx[message_id]
+ if chunk_text:
+ ctx['accumulated_text'] += chunk_text
+ if not ctx['session_started']:
+ if not ctx['accumulated_text']:
+ return
+ ctx['session_started'] = True
+
+ content_to_send = ctx['accumulated_text'][ctx['sent_length'] :]
+ if not content_to_send and not is_final:
+ return
+ now = time.time()
+ if not is_final and (now - ctx['last_update_ts']) < 0.5:
+ return
+ ctx['last_update_ts'] = now
+
+ 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=10 if is_final else 1,
+ )
+ if isinstance(resp, dict) and resp.get('id'):
+ ctx['stream_msg_id'] = resp['id']
+ ctx['sent_length'] = len(ctx['accumulated_text'])
+ ctx['index'] += 1
+ if is_final:
+ self._stream_ctx.pop(message_id, None)
+ self._stream_ctx_ts.pop(message_id, None)
+
+ def _register_native_handlers(self):
+ for event_type in ('C2C_MESSAGE_CREATE', 'DIRECT_MESSAGE_CREATE', 'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'):
+ self.bot.on_message(event_type)(self._handle_native_event)
+
+ async def _handle_native_event(self, event: QQOfficialEvent):
+ self.bot_account_id = self.config.get('appid', self.bot_account_id)
+ try:
+ if platform_events.FriendMessage in self.listeners or platform_events.GroupMessage in self.listeners:
+ legacy_event = await self.event_converter.target2legacy(event)
+ if legacy_event and type(legacy_event) in self.listeners:
+ await self.listeners[type(legacy_event)](legacy_event, self)
+
+ eba_event = await self.event_converter.target2yiri(event)
+ if eba_event:
+ self._cache_event(eba_event)
+ await self._dispatch_eba_event(eba_event)
+ except Exception:
+ await self.logger.error(f'Error in qqofficial native event: {traceback.format_exc()}')
+
+ async def _dispatch_eba_event(self, event: platform_events.EBAEvent):
+ for event_type in (type(event), platform_events.EBAEvent, platform_events.Event):
+ callback = self.listeners.get(event_type)
+ if callback:
+ await callback(event, self)
+ return
+
+ def _cache_event(self, event: platform_events.Event):
+ if not isinstance(event, platform_events.MessageReceivedEvent):
+ return
+ self._message_cache[str(event.message_id)] = event
+ self._user_cache[str(event.sender.id)] = event.sender
+ if event.group:
+ self._group_cache[str(event.group.id)] = event.group
+ self._member_cache[(str(event.group.id), str(event.sender.id))] = platform_entities.UserGroupMember(
+ user=event.sender,
+ group_id=event.group.id,
+ role=platform_entities.MemberRole.MEMBER,
+ display_name=event.sender.nickname,
+ )
+
+ async def _run_websocket(self):
+ await self.logger.info('QQ Official EBA 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):
+ if event_type not in {'C2C_MESSAGE_CREATE', 'DIRECT_MESSAGE_CREATE', 'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'}:
+ await self._dispatch_eba_event(QQOfficialEventConverter.platform_specific(QQOfficialEvent({'t': event_type, **(event_data or {})}), f'qqofficial.{event_type}'))
+ return
+ if not isinstance(event_data, dict):
+ await self.logger.warning(f'Event data is not dict, skipping: {event_type} -> {type(event_data)}')
+ return
+ payload = {'t': event_type, 'd': event_data}
+ message_data = await self.bot.get_message(payload)
+ if message_data:
+ await self.bot._handle_message(QQOfficialEvent.from_payload(message_data))
+
+ async def on_error(error: Exception):
+ 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
+
+ @staticmethod
+ def _reply_target(event: QQOfficialEvent) -> tuple[str, str]:
+ if event.t == 'C2C_MESSAGE_CREATE':
+ return 'person', event.user_openid
+ if event.t == 'GROUP_AT_MESSAGE_CREATE':
+ return 'group', event.group_openid
+ if event.t == 'AT_MESSAGE_CREATE':
+ return 'channel', event.channel_id
+ if event.t == 'DIRECT_MESSAGE_CREATE':
+ return 'channel_private', event.guild_id
+ raise NotSupportedError(f'reply_message:{event.t or "unknown_event"}')
+
+ async def _send_content_list(self, target_type: str, target_id: str, content_list: list[dict], msg_id: str | None = None) -> list[dict]:
+ target_type = self._normalize_target_type(target_type)
+ results: list[dict] = []
+ for content in content_list:
+ content_type = content.get('type', 'text')
+ if target_type == 'channel':
+ if content_type == 'text':
+ raw = await self.bot.send_channle_group_text_msg(target_id, content.get('content', ''), msg_id)
+ results.append({'type': content_type, 'raw': raw})
+ continue
+ if target_type == 'channel_private':
+ if content_type == 'text':
+ raw = await self.bot.send_channle_private_text_msg(target_id, content.get('content', ''), msg_id)
+ results.append({'type': content_type, 'raw': raw})
+ continue
+ if content_type == 'text':
+ if target_type == 'c2c':
+ raw = await self.bot.send_private_text_msg(target_id, content.get('content', ''), msg_id)
+ elif target_type == 'group':
+ raw = await self.bot.send_group_text_msg(target_id, content.get('content', ''), msg_id)
+ else:
+ raise NotSupportedError(f'send_message:{target_type}')
+ results.append({'type': content_type, 'raw': raw})
+ elif content_type == 'image':
+ raw = await self.bot.send_image_msg(target_type, target_id, file_url=content.get('url'), file_data=content.get('base64'), msg_id=msg_id)
+ results.append({'type': content_type, 'raw': raw})
+ elif content_type == 'voice':
+ raw = await self.bot.send_voice_msg(target_type, target_id, file_url=content.get('url'), file_data=content.get('base64'), msg_id=msg_id)
+ results.append({'type': content_type, 'raw': raw})
+ elif content_type == 'file':
+ raw = await self.bot.send_file_msg(
+ target_type,
+ target_id,
+ file_url=content.get('url'),
+ file_data=content.get('base64'),
+ file_name=content.get('name', 'file'),
+ msg_id=msg_id,
+ )
+ results.append({'type': content_type, 'raw': raw})
+ return results
+
+ @staticmethod
+ def _normalize_target_type(target_type: str) -> str:
+ if target_type in {'person', 'private', 'friend', 'c2c'}:
+ return 'c2c'
+ if target_type in {'group', 'group_openid'}:
+ return 'group'
+ if target_type in {'channel', 'guild'}:
+ return 'channel'
+ if target_type in {'channel_private', 'direct', 'dm'}:
+ return 'channel_private'
+ return target_type
+
+ async def _cleanup_stale_streams(self):
+ now = time.time()
+ for message_id in [key for key, ts in self._stream_ctx_ts.items() if now - ts > self._STREAM_CTX_TTL]:
+ self._stream_ctx.pop(message_id, None)
+ self._stream_ctx_ts.pop(message_id, None)
+ for message_id in [key for key, ts in self._fallback_text_ts.items() if now - ts > self._STREAM_CTX_TTL]:
+ self._fallback_text.pop(message_id, None)
+ self._fallback_text_ts.pop(message_id, None)
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/api_impl.py b/src/langbot/pkg/platform/adapters/qqofficial/api_impl.py
new file mode 100644
index 00000000..d7184767
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/api_impl.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import typing
+
+from langbot.pkg.platform.adapters.qqofficial.errors import NotSupportedError
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+class QQOfficialAPIMixin:
+ _message_cache: dict[str, platform_events.MessageReceivedEvent]
+ _user_cache: dict[str, platform_entities.User]
+ _group_cache: dict[str, platform_entities.UserGroup]
+ _member_cache: dict[tuple[str, str], platform_entities.UserGroupMember]
+
+ async def get_message(
+ self,
+ chat_type: str,
+ chat_id: typing.Union[int, str],
+ message_id: typing.Union[int, str],
+ ) -> platform_events.MessageReceivedEvent:
+ event = self._message_cache.get(str(message_id))
+ if event is None:
+ raise NotSupportedError('get_message:message_not_cached')
+ return event
+
+ async def get_user_info(self, user_id: typing.Union[int, str]) -> platform_entities.User:
+ user = self._user_cache.get(str(user_id))
+ if user is None:
+ raise NotSupportedError('get_user_info:not_cached')
+ return user
+
+ async def get_friend_list(self) -> list[platform_entities.User]:
+ return list(self._user_cache.values())
+
+ async def get_group_info(self, group_id: typing.Union[int, str]) -> platform_entities.UserGroup:
+ group = self._group_cache.get(str(group_id))
+ if group is None:
+ raise NotSupportedError('get_group_info:not_cached')
+ return group
+
+ async def get_group_member_info(
+ self,
+ group_id: typing.Union[int, str],
+ user_id: typing.Union[int, str],
+ ) -> platform_entities.UserGroupMember:
+ member = self._member_cache.get((str(group_id), str(user_id)))
+ if member is None:
+ raise NotSupportedError('get_group_member_info:not_cached')
+ return member
+
+ async def get_group_member_list(
+ self,
+ group_id: typing.Union[int, str],
+ ) -> list[platform_entities.UserGroupMember]:
+ return [member for (cached_group_id, _), member in self._member_cache.items() if cached_group_id == str(group_id)]
+
+ async def edit_message(
+ self,
+ chat_type: str,
+ chat_id: typing.Union[int, str],
+ message_id: typing.Union[int, str],
+ new_content: platform_message.MessageChain,
+ ) -> None:
+ raise NotSupportedError('edit_message')
+
+ async def delete_message(
+ self,
+ chat_type: str,
+ chat_id: typing.Union[int, str],
+ message_id: typing.Union[int, str],
+ ) -> None:
+ raise NotSupportedError('delete_message')
+
+ async def forward_message(
+ self,
+ from_chat_type: str,
+ from_chat_id: typing.Union[int, str],
+ message_id: typing.Union[int, str],
+ to_chat_type: str,
+ to_chat_id: typing.Union[int, str],
+ ) -> platform_events.MessageResult:
+ raise NotSupportedError('forward_message')
+
+ async def upload_file(self, file_data: bytes, filename: str) -> str:
+ raise NotSupportedError('upload_file')
+
+ async def get_file_url(self, file_id: str) -> str:
+ raise NotSupportedError('get_file_url')
+
+ async def mute_member(self, group_id: typing.Union[int, str], user_id: typing.Union[int, str], duration: int = 0):
+ raise NotSupportedError('mute_member')
+
+ async def unmute_member(self, group_id: typing.Union[int, str], user_id: typing.Union[int, str]):
+ raise NotSupportedError('unmute_member')
+
+ async def kick_member(self, group_id: typing.Union[int, str], user_id: typing.Union[int, str]):
+ raise NotSupportedError('kick_member')
+
+ async def leave_group(self, group_id: typing.Union[int, str]):
+ raise NotSupportedError('leave_group')
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/errors.py b/src/langbot/pkg/platform/adapters/qqofficial/errors.py
new file mode 100644
index 00000000..72483b09
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/errors.py
@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+try:
+ from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedError
+except ModuleNotFoundError:
+
+ class NotSupportedError(Exception):
+ def __init__(self, api_name: str, *args):
+ super().__init__(f"API '{api_name}' is not supported by this adapter", *args)
+ self.api_name = api_name
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/event_converter.py b/src/langbot/pkg/platform/adapters/qqofficial/event_converter.py
new file mode 100644
index 00000000..051c068e
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/event_converter.py
@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import datetime
+import time
+import typing
+
+from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
+from langbot.pkg.platform.adapters.qqofficial.message_converter import QQOfficialMessageConverter
+from langbot.pkg.platform.adapters.qqofficial.types import ADAPTER_NAME
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+
+
+class QQOfficialEventConverter(abstract_platform_adapter.AbstractEventConverter):
+ @staticmethod
+ async def yiri2target(event: platform_events.Event) -> typing.Any:
+ return getattr(event, 'source_platform_object', None)
+
+ async def target2legacy(self, event: QQOfficialEvent) -> platform_events.FriendMessage | platform_events.GroupMessage | None:
+ eba_event = await self.target2yiri(event)
+ if not isinstance(eba_event, platform_events.MessageReceivedEvent):
+ return None
+ if eba_event.chat_type == platform_entities.ChatType.PRIVATE:
+ return platform_events.FriendMessage(
+ sender=platform_entities.Friend(
+ id=eba_event.sender.id,
+ nickname=eba_event.sender.nickname,
+ remark='',
+ ),
+ message_chain=eba_event.message_chain,
+ time=eba_event.timestamp,
+ source_platform_object=event,
+ )
+ return platform_events.GroupMessage(
+ sender=platform_entities.GroupMember(
+ id=eba_event.sender.id,
+ member_name=eba_event.sender.nickname,
+ permission='MEMBER',
+ group=platform_entities.Group(
+ id=eba_event.group.id if eba_event.group else eba_event.chat_id,
+ name=eba_event.group.name if eba_event.group else '',
+ permission=platform_entities.Permission.Member,
+ ),
+ special_title='',
+ ),
+ message_chain=eba_event.message_chain,
+ time=eba_event.timestamp,
+ source_platform_object=event,
+ )
+
+ async def target2yiri(self, event: QQOfficialEvent) -> platform_events.Event:
+ if event.t in {'C2C_MESSAGE_CREATE', 'DIRECT_MESSAGE_CREATE', 'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'}:
+ return await self.message_to_eba(event)
+ return self.platform_specific(event, f'qqofficial.{event.t or "unknown"}')
+
+ async def message_to_eba(self, event: QQOfficialEvent) -> platform_events.MessageReceivedEvent:
+ timestamp = _timestamp_value(event.timestamp)
+ sender = platform_entities.User(
+ id=self._sender_id(event),
+ nickname=event.username or self._sender_id(event),
+ )
+ chat_type = platform_entities.ChatType.PRIVATE
+ chat_id = self._private_chat_id(event)
+ group = None
+ if event.t in {'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'}:
+ chat_type = platform_entities.ChatType.GROUP
+ chat_id = event.channel_id if event.t == 'AT_MESSAGE_CREATE' else event.group_openid
+ chat_id = chat_id or event.group_openid or event.channel_id or ''
+ group = platform_entities.UserGroup(id=str(chat_id), name=str(chat_id))
+
+ return platform_events.MessageReceivedEvent(
+ type='message.received',
+ adapter_name=ADAPTER_NAME,
+ message_id=event.d_id or event.id or '',
+ message_chain=await QQOfficialMessageConverter.target2yiri(event),
+ sender=sender,
+ chat_type=chat_type,
+ chat_id=chat_id or '',
+ group=group,
+ timestamp=timestamp,
+ source_platform_object=event,
+ )
+
+ @staticmethod
+ def _sender_id(event: QQOfficialEvent) -> str:
+ member_openid = event.member_openid or event.get('member_openid', '')
+ if event.t in {'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'}:
+ return member_openid or event.user_openid or event.d_author_id or ''
+ return event.user_openid or member_openid or event.d_author_id or event.guild_id or event.group_openid or ''
+
+ @staticmethod
+ def _private_chat_id(event: QQOfficialEvent) -> str:
+ if event.t == 'DIRECT_MESSAGE_CREATE':
+ return event.guild_id or event.user_openid or ''
+ return event.user_openid or event.guild_id or ''
+
+ @staticmethod
+ def platform_specific(event: QQOfficialEvent, action: str) -> platform_events.PlatformSpecificEvent:
+ return platform_events.PlatformSpecificEvent(
+ type='platform.specific',
+ adapter_name=ADAPTER_NAME,
+ action=action,
+ data=dict(event),
+ timestamp=_timestamp_value(event.timestamp),
+ source_platform_object=event,
+ )
+
+
+def _timestamp_value(value: str) -> float:
+ if not value:
+ return time.time()
+ try:
+ return float(datetime.datetime.strptime(value, '%Y-%m-%dT%H:%M:%S%z').timestamp())
+ except (TypeError, ValueError):
+ return time.time()
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml b/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml
new file mode 100644
index 00000000..09239f00
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/manifest.yaml
@@ -0,0 +1,120 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+
+metadata:
+ name: qqofficial-eba
+ label:
+ en_US: QQ Official API (EBA)
+ zh_Hans: QQ 官方 API (EBA)
+ zh_Hant: QQ 官方 API (EBA)
+ description:
+ en_US: QQ Official API adapter with Event-Based Agents support, using Webhook or WebSocket mode.
+ zh_Hans: QQ 官方 API 适配器（EBA 架构版本），支持 Webhook 和 WebSocket 两种连接模式。
+ zh_Hant: QQ 官方 API 適配器（EBA 架構版本），支援 Webhook 和 WebSocket 兩種連線模式。
+ icon: qqofficial.svg
+
+spec:
+ categories:
+ - china
+ help_links:
+ zh: https://link.langbot.app/zh/platforms/qqofficial
+ en: https://link.langbot.app/en/platforms/qqofficial
+ ja: https://link.langbot.app/ja/platforms/qqofficial
+ config:
+ - name: appid
+ label:
+ en_US: App ID
+ zh_Hans: 应用 ID
+ zh_Hant: 應用 ID
+ type: string
+ required: true
+ default: ""
+ - name: secret
+ label:
+ en_US: Secret
+ zh_Hans: 密钥
+ zh_Hant: 密鑰
+ type: string
+ required: true
+ default: ""
+ - name: token
+ label:
+ en_US: Token
+ zh_Hans: 令牌
+ zh_Hant: 令牌
+ 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 receives messages through LangBot's unified webhook endpoint. Otherwise it uses the QQ WebSocket gateway.
+ zh_Hans: 启用后，机器人通过 LangBot 统一 Webhook 接收消息；否则使用 QQ WebSocket 网关。
+ zh_Hant: 啟用後，機器人透過 LangBot 統一 Webhook 接收訊息；否則使用 QQ 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 adapter uses QQ Official streaming replies for C2C private messages.
+ zh_Hans: 启用后，适配器会对 C2C 私聊使用 QQ 官方流式回复。
+ zh_Hant: 啟用後，適配器會對 C2C 私聊使用 QQ 官方串流回覆。
+ 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
+
+ supported_events:
+ - message.received
+ - platform.specific
+
+ supported_apis:
+ required:
+ - send_message
+ - reply_message
+ optional:
+ - get_message
+ - get_user_info
+ - get_friend_list
+ - get_group_info
+ - get_group_member_list
+ - get_group_member_info
+ - call_platform_api
+
+ platform_specific_apis:
+ - action: check_access_token
+ description: { en_US: "Check whether the cached QQ Official access token is usable", zh_Hans: "检查当前缓存的 QQ 官方 access token 是否可用" }
+ - action: refresh_access_token
+ description: { en_US: "Force refresh the QQ Official access token", zh_Hans: "强制刷新 QQ 官方 access token" }
+ - action: get_gateway_url
+ description: { en_US: "Return the QQ Official WebSocket gateway URL", zh_Hans: "获取 QQ 官方 WebSocket 网关地址" }
+ - action: get_mode
+ description: { en_US: "Return adapter receive and stream-reply mode", zh_Hans: "返回适配器接收模式和流式回复模式" }
+
+execution:
+ python:
+ path: ./adapter.py
+ attr: QQOfficialAdapter
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/message_converter.py b/src/langbot/pkg/platform/adapters/qqofficial/message_converter.py
new file mode 100644
index 00000000..16b43ab9
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/message_converter.py
@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import datetime
+import re
+
+from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
+from langbot.pkg.utils import image
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+def _is_base64_data(value: str) -> bool:
+ if not value:
+ return False
+ if value.startswith('data:'):
+ return True
+ if value.startswith(('http://', 'https://', '/', './', '../')):
+ return False
+ 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) -> list[dict]:
+ content_list: list[dict] = []
+ for component in message_chain:
+ if isinstance(component, platform_message.Source):
+ continue
+ if isinstance(component, platform_message.Plain):
+ content_list.append({'type': 'text', 'content': component.text})
+ elif isinstance(component, platform_message.At):
+ content_list.append({'type': 'text', 'content': f'@{component.display or component.target}'})
+ elif isinstance(component, platform_message.AtAll):
+ content_list.append({'type': 'text', 'content': '@all'})
+ elif isinstance(component, platform_message.Image):
+ content_list.append(QQOfficialMessageConverter._media_payload(component, 'image'))
+ elif isinstance(component, platform_message.Voice):
+ content_list.append(QQOfficialMessageConverter._media_payload(component, 'voice'))
+ elif isinstance(component, platform_message.File):
+ payload = QQOfficialMessageConverter._media_payload(component, 'file')
+ payload['name'] = component.name or component.id or 'file'
+ content_list.append(payload)
+ elif isinstance(component, platform_message.Quote):
+ if component.id is not None:
+ content_list.append({'type': 'text', 'content': f'[Quote {component.id}]'})
+ if component.origin:
+ content_list.extend(await QQOfficialMessageConverter.yiri2target(component.origin))
+ elif isinstance(component, platform_message.Forward):
+ for node in component.node_list:
+ if node.message_chain:
+ content_list.extend(await QQOfficialMessageConverter.yiri2target(node.message_chain))
+ else:
+ text = str(component)
+ if text:
+ content_list.append({'type': 'text', 'content': text})
+ return content_list
+
+ @staticmethod
+ def _media_payload(component, content_type: str) -> dict:
+ url = getattr(component, 'url', '') or getattr(component, 'path', '') or None
+ b64 = getattr(component, 'base64', '') or None
+ if url and not b64 and _is_base64_data(url):
+ b64 = url
+ url = None
+ return {'type': content_type, 'url': url, 'base64': b64}
+
+ @staticmethod
+ async def target2yiri(event: QQOfficialEvent) -> platform_message.MessageChain:
+ components: list[platform_message.MessageComponent] = [
+ platform_message.Source(id=event.d_id or event.id or '', time=_parse_timestamp(event.timestamp)),
+ ]
+
+ if event.t in {'GROUP_AT_MESSAGE_CREATE', 'AT_MESSAGE_CREATE'}:
+ components.append(platform_message.At(target='justbot'))
+
+ if event.attachments:
+ try:
+ base64_url = await image.get_qq_official_image_base64(
+ pic_url=event.attachments,
+ content_type=event.content_type,
+ )
+ components.append(platform_message.Image(base64=base64_url))
+ except Exception:
+ components.append(platform_message.Image(url=event.attachments))
+
+ if event.content:
+ components.append(platform_message.Plain(text=event.content))
+
+ if len(components) == 1 or (
+ len(components) == 2 and isinstance(components[1], platform_message.At)
+ ):
+ components.append(platform_message.Unknown(text=f'[unsupported qqofficial event: {event.t or "unknown"}]'))
+
+ return platform_message.MessageChain(components)
+
+
+def _parse_timestamp(value: str) -> datetime.datetime:
+ if not value:
+ return datetime.datetime.now()
+ try:
+ return datetime.datetime.strptime(value, '%Y-%m-%dT%H:%M:%S%z')
+ except (TypeError, ValueError):
+ return datetime.datetime.now()
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/platform_api.py b/src/langbot/pkg/platform/adapters/qqofficial/platform_api.py
new file mode 100644
index 00000000..b205cc77
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/platform_api.py
@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import typing
+
+
+async def check_access_token(adapter, params: dict) -> dict:
+ ok = await adapter.bot.check_access_token()
+ return {'ok': bool(ok), 'expires_at': getattr(adapter.bot, 'access_token_expiry_time', None)}
+
+
+async def refresh_access_token(adapter, params: dict) -> dict:
+ adapter.bot.access_token = ''
+ adapter.bot.access_token_expiry_time = None
+ await adapter.bot.get_access_token()
+ return {'ok': bool(adapter.bot.access_token), 'expires_at': adapter.bot.access_token_expiry_time}
+
+
+async def get_gateway_url(adapter, params: dict) -> dict:
+ url = await adapter.bot.get_gateway_url()
+ return {'url': url}
+
+
+async def get_mode(adapter, params: dict) -> dict:
+ return {
+ 'webhook': bool(adapter.enable_webhook),
+ 'stream_reply': bool(adapter.config.get('enable-stream-reply') or adapter.config.get('enable_stream_reply')),
+ 'bot_account_id': adapter.bot_account_id,
+ }
+
+
+PLATFORM_API_MAP: dict[str, typing.Callable[[typing.Any, dict], typing.Awaitable[dict]]] = {
+ 'check_access_token': check_access_token,
+ 'refresh_access_token': refresh_access_token,
+ 'get_gateway_url': get_gateway_url,
+ 'get_mode': get_mode,
+}
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/qqofficial.svg b/src/langbot/pkg/platform/adapters/qqofficial/qqofficial.svg
new file mode 100644
index 00000000..8b956b9e
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/qqofficial.svg
@@ -0,0 +1,2 @@
+
+
diff --git a/src/langbot/pkg/platform/adapters/qqofficial/types.py b/src/langbot/pkg/platform/adapters/qqofficial/types.py
new file mode 100644
index 00000000..e5fb8f89
--- /dev/null
+++ b/src/langbot/pkg/platform/adapters/qqofficial/types.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import pydantic
+
+ADAPTER_NAME = 'qqofficial-eba'
+
+
+class QQOfficialAdapterConfig(pydantic.BaseModel):
+ appid: str
+ secret: str
+ token: str
+ enable_webhook: bool = False
+ enable_stream_reply: bool = False
+
diff --git a/tests/e2e/live_qqofficial_eba_probe.py b/tests/e2e/live_qqofficial_eba_probe.py
new file mode 100644
index 00000000..5c45a82c
--- /dev/null
+++ b/tests/e2e/live_qqofficial_eba_probe.py
@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+from quart import Quart, request
+
+from langbot.pkg.platform.adapters.qqofficial.adapter import QQOfficialAdapter
+from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+class ProbeLogger(AbstractEventLogger):
+ async def info(self, text, images=None, message_session_id=None, no_throw=True):
+ print(f'[info] {text}')
+
+ async def debug(self, text, images=None, message_session_id=None, no_throw=True):
+ print(f'[debug] {text}')
+
+ async def warning(self, text, images=None, message_session_id=None, no_throw=True):
+ print(f'[warning] {text}')
+
+ async def error(self, text, images=None, message_session_id=None, no_throw=True):
+ print(f'[error] {text}')
+
+
+def redact(value: Any) -> Any:
+ if isinstance(value, dict):
+ secret_keys = {'secret', 'token', 'authorization', 'access_token', 'clientsecret'}
+ return {key: '' if key.lower() in secret_keys else redact(item) for key, item in value.items()}
+ if isinstance(value, list):
+ return [redact(item) for item in value]
+ return value
+
+
+def summarize_event(event: platform_events.EBAEvent) -> dict:
+ data = {
+ 'type': event.type,
+ 'adapter_name': event.adapter_name,
+ 'timestamp': event.timestamp,
+ }
+ for field in ('message_id', 'chat_id', 'chat_type', 'action', 'data'):
+ if hasattr(event, field):
+ value = getattr(event, field)
+ if hasattr(value, 'value'):
+ value = value.value
+ data[field] = redact(value)
+ if hasattr(event, 'sender') and event.sender is not None:
+ data['sender'] = event.sender.model_dump()
+ if hasattr(event, 'group') and event.group is not None:
+ data['group'] = event.group.model_dump()
+ if hasattr(event, 'message_chain') and event.message_chain is not None:
+ data['message_chain'] = event.message_chain.model_dump()
+ return data
+
+
+def record_api(results: list[dict[str, Any]], name: str, ok: bool, result: Any = None, error: Exception | None = None):
+ entry = {'name': name, 'ok': ok}
+ if result is not None:
+ entry['result'] = redact(result)
+ if error is not None:
+ entry['error'] = repr(error)
+ results.append(entry)
+ print('QQOFFICIAL_EBA_API', json.dumps(entry, ensure_ascii=False, default=str))
+
+
+async def run_api(results: list[dict[str, Any]], name: str, func):
+ try:
+ result = await func()
+ record_api(results, name, True, result)
+ return result
+ except Exception as exc:
+ record_api(results, name, False, error=exc)
+ return None
+
+
+def config_from_env(enable_webhook: bool) -> dict:
+ config = {
+ 'appid': os.getenv('QQOFFICIAL_APPID', ''),
+ 'secret': os.getenv('QQOFFICIAL_SECRET', ''),
+ 'token': os.getenv('QQOFFICIAL_TOKEN', ''),
+ 'enable-webhook': enable_webhook,
+ 'enable-stream-reply': os.getenv('QQOFFICIAL_ENABLE_STREAM_REPLY', '').lower() in {'1', 'true', 'yes'},
+ }
+ missing = [key for key in ('appid', 'secret', 'token') if not config.get(key)]
+ if missing:
+ raise RuntimeError(f'Missing required QQOfficial env vars for fields: {missing}')
+ return config
+
+
+async def run_probe(args: argparse.Namespace):
+ adapter = QQOfficialAdapter(config_from_env(args.webhook), ProbeLogger())
+ events: list[platform_events.EBAEvent] = []
+ api_results: list[dict[str, Any]] = []
+ first_message = asyncio.Event()
+ log_path = Path(args.log)
+ log_path.parent.mkdir(parents=True, exist_ok=True)
+
+ async def listener(event, adapter):
+ events.append(event)
+ summary = summarize_event(event)
+ with log_path.open('a', encoding='utf-8') as fp:
+ fp.write(json.dumps(summary, ensure_ascii=False, default=str) + '\n')
+ print('QQOFFICIAL_EBA_EVENT', json.dumps(summary, ensure_ascii=False, default=str))
+ if isinstance(event, platform_events.MessageReceivedEvent):
+ first_message.set()
+
+ adapter.register_listener(platform_events.EBAEvent, listener)
+
+ server_task = None
+ if args.webhook:
+ app = Quart(__name__)
+
+ @app.route('/callback', methods=['GET', 'POST'])
+ async def callback():
+ return await adapter.handle_unified_webhook('probe', '', request)
+
+ server_task = asyncio.create_task(app.run_task(host=args.host, port=args.port))
+ else:
+ server_task = asyncio.create_task(adapter.run_async())
+
+ try:
+ await asyncio.wait_for(first_message.wait(), timeout=args.timeout)
+ first = next(event for event in events if isinstance(event, platform_events.MessageReceivedEvent))
+ await run_api(api_results, 'reply_message', lambda: adapter.reply_message(first, platform_message.MessageChain([platform_message.Plain(text=args.reply_text)])))
+ await run_api(api_results, 'get_message', lambda: adapter.get_message(first.chat_type.value, first.chat_id, first.message_id))
+ await run_api(api_results, 'get_user_info', lambda: adapter.get_user_info(first.sender.id))
+ await run_api(api_results, 'get_friend_list', adapter.get_friend_list)
+ if getattr(first, 'group', None):
+ await run_api(api_results, 'get_group_info', lambda: adapter.get_group_info(first.group.id))
+ await run_api(api_results, 'get_group_member_info', lambda: adapter.get_group_member_info(first.group.id, first.sender.id))
+ await run_api(api_results, 'get_group_member_list', lambda: adapter.get_group_member_list(first.group.id))
+ await run_api(api_results, 'call_platform_api.get_mode', lambda: adapter.call_platform_api('get_mode', {}))
+ await run_api(api_results, 'call_platform_api.check_access_token', lambda: adapter.call_platform_api('check_access_token', {}))
+ finally:
+ if server_task:
+ server_task.cancel()
+ try:
+ await server_task
+ except asyncio.CancelledError:
+ pass
+ await adapter.kill()
+
+ summary = {
+ 'events': [event.type for event in events],
+ 'api_results': api_results,
+ 'log': str(log_path),
+ }
+ print('QQOFFICIAL_EBA_SUMMARY', json.dumps(summary, ensure_ascii=False, default=str))
+
+
+def main():
+ parser = argparse.ArgumentParser(description='Live QQOfficial EBA adapter probe')
+ parser.add_argument('--host', default='127.0.0.1')
+ parser.add_argument('--port', type=int, default=5312)
+ parser.add_argument('--timeout', type=float, default=300)
+ parser.add_argument('--log', default='data/temp/qqofficial_eba_probe.jsonl')
+ parser.add_argument('--reply-text', default='QQOfficial EBA probe reply')
+ parser.add_argument('--webhook', action='store_true', help='Run a local webhook callback server instead of the WebSocket gateway')
+ args = parser.parse_args()
+ asyncio.run(run_probe(args))
+
+
+if __name__ == '__main__':
+ main()
+
diff --git a/tests/unit_tests/platform/test_qqofficial_eba_adapter.py b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py
new file mode 100644
index 00000000..b423c921
--- /dev/null
+++ b/tests/unit_tests/platform/test_qqofficial_eba_adapter.py
@@ -0,0 +1,267 @@
+from __future__ import annotations
+
+import pathlib
+import datetime
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import yaml
+
+from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
+from langbot.pkg.platform.adapters.qqofficial.adapter import QQOfficialAdapter
+from langbot.pkg.platform.adapters.qqofficial.errors import NotSupportedError
+from langbot.pkg.platform.adapters.qqofficial.event_converter import QQOfficialEventConverter
+from langbot.pkg.platform.adapters.qqofficial.message_converter import QQOfficialMessageConverter
+from langbot.pkg.platform.adapters.qqofficial.platform_api import PLATFORM_API_MAP
+from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+class DummyLogger(AbstractEventLogger):
+ async def info(self, text, images=None, message_session_id=None, no_throw=True):
+ pass
+
+ async def debug(self, text, images=None, message_session_id=None, no_throw=True):
+ pass
+
+ async def warning(self, text, images=None, message_session_id=None, no_throw=True):
+ pass
+
+ async def error(self, text, images=None, message_session_id=None, no_throw=True):
+ pass
+
+
+class DummyQQOfficialClient:
+ MEDIA_TYPE_IMAGE = 1
+ MEDIA_TYPE_VOICE = 3
+ MEDIA_TYPE_FILE = 4
+
+ def __init__(self, *args, **kwargs):
+ self.app_id = kwargs['app_id']
+ self.secret = kwargs['secret']
+ self.token = kwargs['token']
+ self.unified_mode = kwargs['unified_mode']
+ self._message_handlers = {}
+ self.sent = []
+ self.access_token = ''
+ self.access_token_expiry_time = None
+ self.handle_unified_webhook = AsyncMock(return_value='success')
+ self.connect_gateway_loop = AsyncMock()
+
+ def on_message(self, msg_type: str):
+ def decorator(func):
+ self._message_handlers.setdefault(msg_type, []).append(func)
+ return func
+
+ return decorator
+
+ async def check_access_token(self):
+ return bool(self.access_token)
+
+ async def get_access_token(self):
+ self.access_token = 'token'
+ self.access_token_expiry_time = 1710003600
+
+ async def get_gateway_url(self):
+ return 'wss://gateway.example.test'
+
+ async def send_private_text_msg(self, user_openid, content, msg_id=None):
+ self.sent.append(('private_text', user_openid, content, msg_id))
+ return {'id': 'sent-private'}
+
+ async def send_group_text_msg(self, group_openid, content, msg_id=None):
+ self.sent.append(('group_text', group_openid, content, msg_id))
+ return {'id': 'sent-group'}
+
+ async def send_channle_group_text_msg(self, channel_id, content, msg_id=None):
+ self.sent.append(('channel_text', channel_id, content, msg_id))
+ return {'id': 'sent-channel'}
+
+ async def send_channle_private_text_msg(self, guild_id, content, msg_id=None):
+ self.sent.append(('dm_text', guild_id, content, msg_id))
+ return {'id': 'sent-dm'}
+
+ async def send_image_msg(self, target_type, target_id, file_url=None, file_data=None, msg_id=None, content=None):
+ self.sent.append(('image', target_type, target_id, file_url, file_data, msg_id))
+ return {'id': 'sent-image'}
+
+ async def send_voice_msg(self, target_type, target_id, file_url=None, file_data=None, msg_id=None):
+ self.sent.append(('voice', target_type, target_id, file_url, file_data, msg_id))
+ return {'id': 'sent-voice'}
+
+ async def send_file_msg(self, target_type, target_id, file_url=None, file_data=None, file_name=None, msg_id=None):
+ self.sent.append(('file', target_type, target_id, file_url, file_data, file_name, msg_id))
+ return {'id': 'sent-file'}
+
+ async def send_stream_msg(self, **kwargs):
+ self.sent.append(('stream', kwargs))
+ return {'id': 'stream-1'}
+
+
+def manifest() -> dict:
+ manifest_path = (
+ pathlib.Path(__file__).parents[3]
+ / 'src'
+ / 'langbot'
+ / 'pkg'
+ / 'platform'
+ / 'adapters'
+ / 'qqofficial'
+ / 'manifest.yaml'
+ )
+ return yaml.safe_load(manifest_path.read_text())
+
+
+def make_adapter(enable_webhook: bool = True) -> QQOfficialAdapter:
+ config = {
+ 'appid': 'app-id',
+ 'secret': 'secret',
+ 'token': 'token',
+ 'enable-webhook': enable_webhook,
+ 'enable-stream-reply': True,
+ }
+ with patch('langbot.pkg.platform.adapters.qqofficial.adapter.QQOfficialClient', DummyQQOfficialClient):
+ return QQOfficialAdapter(config, DummyLogger())
+
+
+def qq_event(event_type: str = 'C2C_MESSAGE_CREATE', **overrides) -> QQOfficialEvent:
+ payload = {
+ 't': event_type,
+ 'user_openid': overrides.get('user_openid', 'user-openid'),
+ 'timestamp': overrides.get('timestamp', '2026-06-01T10:00:00+0800'),
+ 'd_author_id': overrides.get('author_id', 'author-id'),
+ 'content': overrides.get('content', 'hello'),
+ 'd_id': overrides.get('message_id', 'msg-1'),
+ 'id': overrides.get('event_id', 'event-1'),
+ 'channel_id': overrides.get('channel_id', 'channel-1'),
+ 'username': overrides.get('username', 'alice'),
+ 'guild_id': overrides.get('guild_id', 'guild-1'),
+ 'member_openid': overrides.get('member_openid', 'member-openid'),
+ 'group_openid': overrides.get('group_openid', 'group-openid'),
+ 'image_attachments': overrides.get('image_attachments'),
+ 'content_type': overrides.get('content_type', 'image/png'),
+ }
+ return QQOfficialEvent(payload)
+
+
+def test_qqofficial_supported_events_match_manifest():
+ assert make_adapter().get_supported_events() == manifest()['spec']['supported_events']
+
+
+def test_qqofficial_supported_apis_match_manifest():
+ supported_apis = make_adapter().get_supported_apis()
+ manifest_apis = manifest()['spec']['supported_apis']
+
+ assert supported_apis == manifest_apis['required'] + manifest_apis['optional']
+
+
+def test_qqofficial_platform_api_map_matches_manifest():
+ manifest_actions = {item['action'] for item in manifest()['spec']['platform_specific_apis']}
+
+ assert set(PLATFORM_API_MAP) == manifest_actions
+
+
+@pytest.mark.asyncio
+async def test_qqofficial_message_converter_maps_common_components_to_send_payloads():
+ payload = await QQOfficialMessageConverter.yiri2target(
+ platform_message.MessageChain(
+ [
+ platform_message.Source(id='msg-0', time=datetime.datetime.now()),
+ platform_message.Plain(text='hi'),
+ platform_message.At(target='user-1', display='Alice'),
+ platform_message.AtAll(),
+ platform_message.Image(url='https://example.test/a.png'),
+ platform_message.Voice(base64='data:audio/mpeg;base64,AAAA'),
+ platform_message.File(name='a.txt', url='https://example.test/a.txt'),
+ platform_message.Quote(origin=platform_message.MessageChain([platform_message.Plain(text='quoted')])),
+ ]
+ )
+ )
+
+ assert {'type': 'text', 'content': 'hi'} in payload
+ assert {'type': 'text', 'content': '@Alice'} in payload
+ assert {'type': 'text', 'content': '@all'} in payload
+ assert any(item['type'] == 'image' and item['url'] == 'https://example.test/a.png' for item in payload)
+ assert any(item['type'] == 'voice' and item['base64'].startswith('data:audio') for item in payload)
+ assert any(item['type'] == 'file' and item['name'] == 'a.txt' for item in payload)
+ assert {'type': 'text', 'content': 'quoted'} in payload
+
+
+@pytest.mark.asyncio
+async def test_qqofficial_event_converter_maps_private_group_and_platform_specific():
+ private_event = await QQOfficialEventConverter().target2yiri(qq_event('C2C_MESSAGE_CREATE'))
+ group_event = await QQOfficialEventConverter().target2yiri(qq_event('GROUP_AT_MESSAGE_CREATE'))
+ channel_event = await QQOfficialEventConverter().target2yiri(qq_event('AT_MESSAGE_CREATE'))
+ platform_event = await QQOfficialEventConverter().target2yiri(qq_event('GUILD_CREATE'))
+
+ assert isinstance(private_event, platform_events.MessageReceivedEvent)
+ assert private_event.adapter_name == 'qqofficial-eba'
+ assert private_event.chat_type == platform_entities.ChatType.PRIVATE
+ assert private_event.chat_id == 'user-openid'
+ assert str(private_event.message_chain) == 'hello'
+
+ assert isinstance(group_event, platform_events.MessageReceivedEvent)
+ assert group_event.chat_type == platform_entities.ChatType.GROUP
+ assert group_event.chat_id == 'group-openid'
+ assert isinstance(group_event.message_chain[1], platform_message.At)
+
+ assert channel_event.chat_id == 'channel-1'
+ assert isinstance(platform_event, platform_events.PlatformSpecificEvent)
+ assert platform_event.action == 'qqofficial.GUILD_CREATE'
+
+
+@pytest.mark.asyncio
+async def test_qqofficial_adapter_dispatches_eba_and_legacy_and_caches_group_event():
+ adapter = make_adapter()
+ eba_calls: list[platform_events.Event] = []
+ legacy_calls: list[platform_events.Event] = []
+
+ async def eba_listener(event, adapter):
+ eba_calls.append(event)
+
+ async def legacy_listener(event, adapter):
+ legacy_calls.append(event)
+
+ adapter.register_listener(platform_events.MessageReceivedEvent, eba_listener)
+ adapter.register_listener(platform_events.GroupMessage, legacy_listener)
+ await adapter._handle_native_event(qq_event('GROUP_AT_MESSAGE_CREATE'))
+
+ assert len(eba_calls) == 1
+ assert len(legacy_calls) == 1
+ received = eba_calls[0]
+ assert await adapter.get_message('group', 'group-openid', 'msg-1') == received
+ assert (await adapter.get_group_info('group-openid')).id == 'group-openid'
+ assert (await adapter.get_group_member_info('group-openid', 'member-openid')).user.id == 'member-openid'
+
+
+@pytest.mark.asyncio
+async def test_qqofficial_send_reply_stream_platform_api_and_unsupported():
+ adapter = make_adapter()
+ message = platform_message.MessageChain(
+ [
+ platform_message.Plain(text='reply'),
+ platform_message.Image(url='https://example.test/a.png'),
+ ]
+ )
+ source_event = await QQOfficialEventConverter().target2yiri(qq_event('C2C_MESSAGE_CREATE'))
+
+ reply_result = await adapter.reply_message(source_event, message)
+ assert reply_result.message_id == 'msg-1'
+ assert ('private_text', 'user-openid', 'reply', 'msg-1') in adapter.bot.sent
+ assert any(call[0] == 'image' and call[1] == 'c2c' for call in adapter.bot.sent)
+
+ await adapter.send_message('group', 'group-openid', platform_message.MessageChain([platform_message.Plain(text='hello group')]))
+ assert ('group_text', 'group-openid', 'hello group', None) in adapter.bot.sent
+
+ assert await adapter.call_platform_api('get_mode', {}) == {
+ 'webhook': True,
+ 'stream_reply': True,
+ 'bot_account_id': 'app-id',
+ }
+ await adapter.call_platform_api('refresh_access_token', {})
+ assert adapter.bot.access_token == 'token'
+
+ with pytest.raises(NotSupportedError):
+ await adapter.call_platform_api('missing', {})