diff --git a/src/langbot/libs/official_account_api/api.py b/src/langbot/libs/official_account_api/api.py index 10508b097..881ce99ae 100644 --- a/src/langbot/libs/official_account_api/api.py +++ b/src/langbot/libs/official_account_api/api.py @@ -5,6 +5,7 @@ from langbot.libs.wecom_api.WXBizMsgCrypt3 import WXBizMsgCrypt import xml.etree.ElementTree as ET from quart import Quart, request import hashlib +import hmac from typing import Callable from langbot.libs.official_account_api.oaevent import OAEvent @@ -24,6 +25,28 @@ xml_template = """ _MAX_CALLBACK_BODY_BYTES = 1024 * 1024 +async def _read_callback_xml(client, req): + body = await req.data + if len(body) > _MAX_CALLBACK_BODY_BYTES: + raise ValueError('Official Account callback body exceeds the size limit') + root = ET.fromstring(body) + if root.find('Encrypt') is not None: + crypt = WXBizMsgCrypt(client.token, client.aes, client.appid) + code, xml = await asyncio.to_thread( + crypt.DecryptMsg, body, req.args.get('msg_signature', ''), + req.args.get('timestamp', ''), req.args.get('nonce', ''), + ) + if code != 0: + raise ValueError('Official Account callback decryption failed') + return xml.decode('utf-8') if isinstance(xml, bytes) else xml + digest = hashlib.sha1(''.join(sorted([ + client.token, req.args.get('timestamp', ''), req.args.get('nonce', ''), + ])).encode('utf-8')).hexdigest() + if not hmac.compare_digest(digest, req.args.get('signature', '')): + raise ValueError('Official Account callback signature verification failed') + return body.decode('utf-8') + + class OAClient: _STATE_TTL_SECONDS = 600 _STATE_MAX = 4096 @@ -63,6 +86,7 @@ class OAClient: } self.access_token_expiry_time = None self.msg_id_map = {} + self._processing_tasks: set[asyncio.Task] = set() self.generated_content = {} self._msg_seen_at = {} self._generated_at = {} @@ -91,6 +115,9 @@ class OAClient: self._generated_at.pop(message_id, None) def clear(self) -> None: + for task in self._processing_tasks: + task.cancel() + self._processing_tasks.clear() self.msg_id_map.clear() self.generated_content.clear() self._msg_seen_at.clear() @@ -155,28 +182,17 @@ class OAClient: ) return 'signature verification failed', 403 elif req.method == 'POST': - encryt_msg = await req.data - if len(encryt_msg) > _MAX_CALLBACK_BODY_BYTES: - raise ValueError('Official Account callback body exceeds the size limit') - wxcpt = WXBizMsgCrypt(self.token, self.aes, self.appid) - ret, xml_msg = await asyncio.to_thread( - wxcpt.DecryptMsg, - encryt_msg, - msg_signature, - timestamp, - nonce, - ) - xml_msg = xml_msg.decode('utf-8') - - if ret != 0: - await self.logger.error('消息解密失败') - raise Exception('消息解密失败') + xml_msg = await _read_callback_xml(self, req) message_data = await self.get_message(xml_msg) if message_data: event = OAEvent.from_payload(message_data) if event: - await self._handle_message(event) + from langbot.pkg.core.task_boundary import create_detached_task + + task = create_detached_task(self._process_message(event)) + self._processing_tasks.add(task) + task.add_done_callback(self._processing_tasks.discard) root = await asyncio.to_thread(ET.fromstring, xml_msg) from_user = root.find('FromUserName').text # 发送者 @@ -192,7 +208,7 @@ class OAClient: to_user=from_user, from_user=to_user, create_time=int(time.time()), - content=content, + content=content.replace(']]>', ']]]]>'), ) return response_xml @@ -202,15 +218,7 @@ class OAClient: await asyncio.sleep(interval) - if self.msg_id_map.get(message_data['MsgId'], 1) == 3: - # response_xml = xml_template.format( - # to_user=from_user, - # from_user=to_user, - # create_time=int(time.time()), - # content = "请求失效：暂不支持公众号超过15秒的请求，如有需求，请联系 LangBot 团队。" - # ) - print('请求失效：暂不支持公众号超过15秒的请求，如有需求，请联系 LangBot 团队。') - return '' + return 'success' except Exception: await self.logger.error(f'handle_callback_request失败: {traceback.format_exc()}') @@ -219,14 +227,14 @@ class OAClient: async def get_message(self, xml_msg: str): root = await asyncio.to_thread(ET.fromstring, xml_msg) - message_data = { - 'ToUserName': root.find('ToUserName').text, - 'FromUserName': root.find('FromUserName').text, - 'CreateTime': int(root.find('CreateTime').text), - 'MsgType': root.find('MsgType').text, - 'Content': root.find('Content').text if root.find('Content') is not None else None, - 'MsgId': int(root.find('MsgId').text) if root.find('MsgId') is not None else None, - } + message_data = {element.tag: element.text for element in root} + message_data['CreateTime'] = int(message_data.get('CreateTime') or 0) + if message_data.get('MsgId'): + message_data['MsgId'] = int(message_data['MsgId']) + else: + message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in ( + 'FromUserName', 'CreateTime', 'Event', 'EventKey', + )) return message_data @@ -249,6 +257,12 @@ class OAClient: return decorator + async def _process_message(self, event: OAEvent): + try: + await self._handle_message(event) + except Exception: + await self.logger.error(f'OfficialAccount processing failed: {traceback.format_exc()}') + async def _handle_message(self, event: OAEvent): """ 处理消息事件。 @@ -288,6 +302,7 @@ class OAClientForLongerResponse: logger: None, unified_mode: bool = False, api_base_url: str = 'https://api.weixin.qq.com', + customer_service: bool = False, ): self.token = token self.aes = EncodingAESKey @@ -313,8 +328,13 @@ class OAClientForLongerResponse: } self.access_token_expiry_time = None self.loading_message = LoadingMessage + from .customer_service import CustomerServiceReplies + + self.customer_service = CustomerServiceReplies(AppID, Appsecret, api_base_url) if customer_service else None + self._callback_responses = {} self.msg_queue = {} self.user_msg_queue = {} + self._processing_tasks: set[asyncio.Task] = set() self._last_queue_cleanup = 0.0 self.logger = logger @@ -336,6 +356,10 @@ class OAClientForLongerResponse: def clear(self) -> None: self.msg_queue.clear() self.user_msg_queue.clear() + self._callback_responses.clear() + for task in self._processing_tasks: + task.cancel() + self._processing_tasks.clear() async def handle_callback_request(self): """处理回调请求（独立端口模式，使用全局 request）。""" @@ -393,27 +417,24 @@ class OAClientForLongerResponse: return 'signature verification failed', 403 elif req.method == 'POST': - encryt_msg = await req.data - if len(encryt_msg) > _MAX_CALLBACK_BODY_BYTES: - raise ValueError('Official Account callback body exceeds the size limit') - wxcpt = WXBizMsgCrypt(self.token, self.aes, self.appid) - ret, xml_msg = await asyncio.to_thread( - wxcpt.DecryptMsg, - encryt_msg, - msg_signature, - timestamp, - nonce, - ) - xml_msg = xml_msg.decode('utf-8') - - if ret != 0: - await self.logger.error('消息解密失败') - raise Exception('消息解密失败') + xml_msg = await _read_callback_xml(self, req) # 解析 XML root = await asyncio.to_thread(ET.fromstring, xml_msg) from_user = root.find('FromUserName').text to_user = root.find('ToUserName').text + message_key = (from_user, root.findtext('MsgId') or ( + root.findtext('CreateTime'), root.findtext('Event'), root.findtext('EventKey') + )) + cached = self._callback_responses.get(message_key) + if cached and time.monotonic() - cached[0] < 600: + return cached[1] + + def remember(response): + self._callback_responses[message_key] = (time.monotonic(), response) + while len(self._callback_responses) > self._MAX_USERS: + self._callback_responses.pop(next(iter(self._callback_responses))) + return response if self.msg_queue.get(from_user) and self.msg_queue[from_user][0]['content']: queue_top = self.msg_queue[from_user].pop(0) @@ -428,20 +449,22 @@ class OAClientForLongerResponse: to_user=from_user, from_user=to_user, create_time=int(time.time()), - content=queue_content, + content=queue_content.replace(']]>', ']]]]>'), ) - return response_xml + return remember(response_xml) else: response_xml = xml_template.format( to_user=from_user, from_user=to_user, create_time=int(time.time()), - content=self.loading_message, + content=self.loading_message.replace(']]>', ']]]]>'), ) + if not self.loading_message: + response_xml = 'success' - if self.user_msg_queue.get(from_user) and self.user_msg_queue[from_user][0]['content']: - return response_xml + if self.user_msg_queue.get(from_user): + return remember(response_xml) else: message_data = await self.get_message(xml_msg) @@ -457,9 +480,15 @@ class OAClientForLongerResponse: -self._MAX_MESSAGES_PER_USER : ] self._prune_queues() - await self._handle_message(event) + # The callback must return the loading reply immediately. + # Native listeners establish their own Workspace scope. + from langbot.pkg.core.task_boundary import create_detached_task - return response_xml + task = create_detached_task(self._process_pending_message(event)) + self._processing_tasks.add(task) + task.add_done_callback(self._processing_tasks.discard) + + return remember(response_xml) except Exception: await self.logger.error(f'handle_callback_request失败: {traceback.format_exc()}') @@ -468,14 +497,14 @@ class OAClientForLongerResponse: async def get_message(self, xml_msg: str): root = await asyncio.to_thread(ET.fromstring, xml_msg) - message_data = { - 'ToUserName': root.find('ToUserName').text, - 'FromUserName': root.find('FromUserName').text, - 'CreateTime': int(root.find('CreateTime').text), - 'MsgType': root.find('MsgType').text, - 'Content': root.find('Content').text if root.find('Content') is not None else None, - 'MsgId': int(root.find('MsgId').text) if root.find('MsgId') is not None else None, - } + message_data = {element.tag: element.text for element in root} + message_data['CreateTime'] = int(message_data.get('CreateTime') or 0) + if message_data.get('MsgId'): + message_data['MsgId'] = int(message_data['MsgId']) + else: + message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in ( + 'FromUserName', 'CreateTime', 'Event', 'EventKey', + )) return message_data @@ -508,7 +537,32 @@ class OAClientForLongerResponse: for handler in self._message_handlers[msg_type]: await handler(event) + async def _process_pending_message(self, event: OAEvent): + try: + await self._handle_message(event) + except asyncio.CancelledError: + raise + except Exception: + self.user_msg_queue.pop(event.user_id, None) + await self.logger.error(f'OfficialAccount passive processing failed: {traceback.format_exc()}') + async def set_message(self, from_user: int, message_id: int, content: str): + if self.customer_service is not None: + try: + code = await self.customer_service.send(str(from_user), content) + except Exception as exc: + # Do not expose request URLs containing access tokens or AppSecret. + self.user_msg_queue.pop(from_user, None) + raise RuntimeError(f'OfficialAccount customer message failed ({type(exc).__name__})') from None + if code == 0: + self.user_msg_queue.pop(from_user, None) + return False + if code not in {48001, 45015, 45047, 43004}: + self.user_msg_queue.pop(from_user, None) + raise RuntimeError(f'OfficialAccount customer message rejected: errcode={code}') + await self.logger.info( + f'OfficialAccount customer message unavailable (errcode={code}); reply queued for next message' + ) if from_user not in self.msg_queue: self.msg_queue[from_user] = [] @@ -520,3 +574,4 @@ class OAClientForLongerResponse: ) self.msg_queue[from_user] = self.msg_queue[from_user][-self._MAX_MESSAGES_PER_USER :] self._prune_queues() + return True diff --git a/src/langbot/libs/official_account_api/customer_service.py b/src/langbot/libs/official_account_api/customer_service.py new file mode 100644 index 000000000..e5fe23f75 --- /dev/null +++ b/src/langbot/libs/official_account_api/customer_service.py @@ -0,0 +1,46 @@ +"""Optional asynchronous replies through WeChat's customer service API.""" + +import asyncio +import time + +import httpx + + +class CustomerServiceReplies: + def __init__(self, appid: str, secret: str, base_url: str): + self.appid = appid + self.secret = secret + self.base_url = base_url.rstrip('/') + self._token = '' + self._expires = 0.0 + self._lock = asyncio.Lock() + + async def send(self, user_id: str, content: str) -> int: + # Retry only explicit token rejection, never an ambiguous network failure. + async with self._lock, httpx.AsyncClient(timeout=20) as client: + for attempt in range(2): + if time.monotonic() >= self._expires: + response = await client.get( + f'{self.base_url}/cgi-bin/token', + params={'grant_type': 'client_credential', 'appid': self.appid, 'secret': self.secret}, + ) + if response.status_code != 200: + raise RuntimeError(f'OfficialAccount token request HTTP {response.status_code}') + data = response.json() + if data.get('errcode'): + return int(data['errcode']) + self._token = data['access_token'] + self._expires = time.monotonic() + max(0, int(data['expires_in']) - 120) + response = await client.post( + f'{self.base_url}/cgi-bin/message/custom/send', + params={'access_token': self._token}, + json={'touser': user_id, 'msgtype': 'text', 'text': {'content': content}}, + ) + if response.status_code != 200: + raise RuntimeError(f'OfficialAccount customer message HTTP {response.status_code}') + code = int(response.json()['errcode']) + if code in {40001, 40014, 42001} and attempt == 0: + self._expires = 0.0 + continue + return code + raise RuntimeError('OfficialAccount token refresh failed') diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 6591542ed..27c4945d0 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -465,7 +465,11 @@ def build_platform_tool_resources( reason = None if definition is None: reason = 'unknown_tool' - elif definition.api not in supported_apis: + elif definition.api not in supported_apis and not ( + definition.name == 'event_reply' + and {'reply_message', 'get_message'} <= supported_apis + and (event.delivery.reply_target or {}).get('message_id') + ): reason = 'adapter_api_unsupported' elif definition.scope == 'event' and not _event_matches(event.event_type, definition.event_patterns): reason = 'event_incompatible' @@ -727,9 +731,24 @@ async def execute_platform_tool( bot = await ap.platform_mgr.get_bot_by_uuid(execution_context, bot_id) if bot is None: raise ValueError(f'Bot {bot_id} is not running') - if definition.api not in set(bot.adapter.get_supported_apis() or []): + supported_apis = set(bot.adapter.get_supported_apis() or []) + reply_via_source = ( + definition.name == 'event_reply' + and 'send_message' not in supported_apis + and {'reply_message', 'get_message'} <= supported_apis + and (delivery.get('reply_target') or {}).get('message_id') + ) + if definition.api not in supported_apis and not reply_via_source: raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}') api_func = getattr(bot.adapter, definition.api, None) + if reply_via_source: + async def api_func(target_type, target_id, message): + source = await bot.adapter.get_message( + chat_type=target_type, + chat_id=target_id, + message_id=delivery['reply_target']['message_id'], + ) + return await bot.adapter.reply_message(source, message) if not callable(api_func): raise ValueError(f'Platform API {definition.api} is declared but not implemented') if definition.api == 'send_message': diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index ef0c680ed..3d2ef6b13 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -704,7 +704,8 @@ class BotService: bot = await self.get_bot(context, bot_data['uuid'], include_secret=True) try: - await self.ap.platform_mgr.load_bot(context, bot) + if bot.get('enable'): + await self.ap.platform_mgr.load_bot(context, bot) except Exception as exc: raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc @@ -740,8 +741,8 @@ class BotService: bot = await self.get_bot(context, bot_uuid, include_secret=True) try: await self.ap.platform_mgr.remove_bot(context, bot_uuid) - runtime_bot = await self.ap.platform_mgr.load_bot(context, bot) - if runtime_bot.enable: + if bot.get('enable'): + runtime_bot = await self.ap.platform_mgr.load_bot(context, bot) await runtime_bot.run() except Exception as exc: raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc diff --git a/src/langbot/pkg/platform/adapters/officialaccount/adapter.py b/src/langbot/pkg/platform/adapters/officialaccount/adapter.py index 124230aad..f4f362b4a 100644 --- a/src/langbot/pkg/platform/adapters/officialaccount/adapter.py +++ b/src/langbot/pkg/platform/adapters/officialaccount/adapter.py @@ -56,13 +56,14 @@ class OfficialAccountAdapter(OfficialAccountAPIMixin, abstract_platform_adapter. } if mode == 'drop': bot = OAClient(**common_kwargs) - elif mode == 'passive': + elif mode in ('passive', 'auto'): bot = OAClientForLongerResponse( **common_kwargs, LoadingMessage=config.get('LoadingMessage', ''), + customer_service=mode == 'auto', ) else: - raise KeyError('OfficialAccount Mode must be "drop" or "passive"') + raise KeyError('OfficialAccount Mode must be "drop", "passive" or "auto"') super().__init__( config=config, @@ -112,11 +113,12 @@ class OfficialAccountAdapter(OfficialAccountAPIMixin, abstract_platform_adapter. if not isinstance(source, OAEvent): raise ValueError('OfficialAccount reply_message requires an OAEvent source object') content = await OfficialAccountMessageConverter.yiri2target(message) - if self.config.get('Mode') == 'passive': - await self.bot.set_message(source.user_id, source.message_id, content) + if self.config.get('Mode') in ('passive', 'auto'): + queued = await self.bot.set_message(source.user_id, source.message_id, content) else: await self.bot.set_message(source.message_id, content) - return platform_events.MessageResult(message_id=source.message_id, raw={'queued': True}) + queued = True + return platform_events.MessageResult(message_id=source.message_id, raw={'queued': queued}) async def call_platform_api(self, action: str, params: dict = {}) -> dict: handler = PLATFORM_API_MAP.get(action) diff --git a/src/langbot/pkg/platform/adapters/officialaccount/manifest.yaml b/src/langbot/pkg/platform/adapters/officialaccount/manifest.yaml index 033c00811..2751023fd 100644 --- a/src/langbot/pkg/platform/adapters/officialaccount/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/officialaccount/manifest.yaml @@ -34,6 +34,22 @@ spec: type: array[string] required: false default: [] + - name: AppID + label: + en_US: App ID + zh_Hans: 应用 ID + zh_Hant: 應用 ID + type: string + required: true + default: "" + - name: AppSecret + label: + en_US: App Secret + zh_Hans: 应用密钥 + zh_Hant: 應用密鑰 + type: string + required: true + default: "" - name: webhook_url label: en_US: Webhook Callback URL @@ -62,32 +78,32 @@ spec: type: string required: true default: "" - - name: AppID - label: - en_US: App ID - zh_Hans: 应用 ID - zh_Hant: 應用 ID - type: string - required: true - default: "" - - name: AppSecret - label: - en_US: App Secret - zh_Hans: 应用密钥 - zh_Hant: 應用密鑰 - type: string - required: true - default: "" - name: Mode label: en_US: Mode zh_Hans: 接入模式 zh_Hant: 接入模式 description: - en_US: "drop replies within the current callback; passive returns a loading message first and queues the real reply for the user's next message." - zh_Hans: "drop 会在当前回调内等待回复；passive 会先返回加载提示，并将真实回复排队到用户下一条消息。" - zh_Hant: "drop 會在目前回調內等待回覆；passive 會先回傳載入提示，並將真實回覆排隊到使用者下一則訊息。" - type: string + en_US: "Auto uses customer service messaging when available, falling back to next-message retrieval for permission, window or quota limits. Passive always waits for the next message. Drop requires a reply within 5 seconds." + zh_Hans: "自动模式在生成完成后通过客服接口回复；无权限、超出回复窗口或额度不足时，改为下一条消息领取。passive 始终由下一条消息领取；drop 需要在 5 秒内完成回复。客服接口通常需企业主体认证，以微信实际权限为准。" + zh_Hant: "自動模式在生成完成後透過客服介面回覆；無權限、超出回覆時限或額度不足時，改為下一則訊息領取。passive 始終由下一則訊息領取；drop 需要在 5 秒內完成回覆。客服介面通常需企業主體認證，以微信實際權限為準。" + type: select + options: + - name: auto + label: + en_US: Automatic reply with fallback (auto) + zh_Hans: 自动回复，受限时转为领取（auto） + zh_Hant: 自動回覆，受限時轉為領取（auto） + - name: drop + label: + en_US: Reply in current callback (drop) + zh_Hans: 当前回调内回复（drop） + zh_Hant: 目前回調內回覆（drop） + - name: passive + label: + en_US: Queue reply for next message (passive) + zh_Hans: 下一条消息时返回回复（passive） + zh_Hant: 下一則訊息時傳回回覆（passive） required: true default: "drop" - name: LoadingMessage diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index fa80a5c6b..7b0bc8e08 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -450,10 +450,13 @@ class RuntimeBot: return {} if hasattr(model, 'model_dump'): try: - return model.model_dump(mode='json') + # Native adapter objects are for replying, not event logging. + # Exclude them before JSON serialization, since some (OAEvent, + # for example) cannot be serialized by Pydantic. + return model.model_dump(mode='json', exclude={'source_platform_object'}) except TypeError: try: - return model.model_dump() + return model.model_dump(exclude={'source_platform_object'}) except Exception: return {} except Exception: @@ -2231,6 +2234,10 @@ class PlatformManager: ) binding = _binding for bot in result.all(): + # Disabled drafts may not have the credentials needed to construct + # an adapter. They become runtime bots when explicitly enabled. + if not bot.enable: + continue try: if binding is None: binding = await self.ap.workspace_service.get_execution_binding(workspace_uuid) diff --git a/web/src/app/home/agents/PluginProcessorDetailContent.tsx b/web/src/app/home/agents/PluginProcessorDetailContent.tsx index 0d848f097..7deefa1fa 100644 --- a/web/src/app/home/agents/PluginProcessorDetailContent.tsx +++ b/web/src/app/home/agents/PluginProcessorDetailContent.tsx @@ -605,7 +605,6 @@ export default function PluginProcessorDetailContent({ hasUnsavedChanges={dirty} beforeRun={save} onRunFinished={() => { - setActiveTab('logs'); void refreshLatestRun(); }} supportedEventPatterns={component.supported_event_patterns} diff --git a/web/src/app/home/components/guided-tour/GuidedTour.tsx b/web/src/app/home/components/guided-tour/GuidedTour.tsx index 7ffdc6ab0..436d61aed 100644 --- a/web/src/app/home/components/guided-tour/GuidedTour.tsx +++ b/web/src/app/home/components/guided-tour/GuidedTour.tsx @@ -57,6 +57,7 @@ export default function GuidedTour({ useState(null); const popoverRef = useRef(null); const previousStorageKeyRef = useRef(storageKey); + const enteredStepRef = useRef(null); const activeIndex = steps.findIndex((step) => step.id === activeStepId); const activeStep = activeIndex >= 0 ? steps[activeIndex] : undefined; @@ -86,8 +87,17 @@ export default function GuidedTour({ }, [activeStepId, enabled, finished, steps, storageKey]); useEffect(() => { - if (enabled && !finished) activeStep?.onEnter?.(); - }, [activeStep, enabled, finished]); + if (!enabled || finished || !activeStep) { + enteredStepRef.current = null; + return; + } + // Callback identities can change during navigation. Enter each step once, + // so a URL update cannot replay its navigation and override the user's tab. + const stepKey = JSON.stringify([storageKey, activeStep.id]); + if (enteredStepRef.current === stepKey) return; + enteredStepRef.current = stepKey; + activeStep.onEnter?.(); + }, [activeStep, enabled, finished, storageKey]); const measure = useCallback(() => { if (!activeStep) return;