fix(feishu): require explicit bot mention

agent/tools/builtin/feishu/feishu_client.py

@@ -26,6 +26,7 @@ from enum import Enum
 from typing import Any, Callable, Dict, List, Optional, Union
 
 import lark_oapi as lark
+import requests
 from lark_oapi.api.contact.v3 import GetUserRequest, GetUserResponse
 from lark_oapi.api.im.v1 import (
     CreateMessageRequest, CreateMessageRequestBody,
@@ -842,11 +843,48 @@ class FeishuClient:
             return False
 
         if not self._bot_open_id:
-            # 如果没有缓存机器人 open_id，假设有 mention 就是 @了机器人
-            return len(mentions) > 0
+            try:
+                self._load_bot_open_id()
+            except Exception:
+                logger.exception("获取机器人 open_id 失败，拒绝处理群聊 mention")
+                return False
 
         return any(m.id.open_id == self._bot_open_id for m in mentions)
 
+    def _load_bot_open_id(self) -> str:
+        """从飞书查询当前应用机器人的 open_id。"""
+        token_response = requests.post(
+            f"{self.domain.value}/open-apis/auth/v3/tenant_access_token/internal/",
+            json={"app_id": self.app_id, "app_secret": self.app_secret},
+            timeout=10,
+        )
+        token_response.raise_for_status()
+        token_payload = token_response.json()
+        if int(token_payload.get("code") or 0) != 0:
+            raise RuntimeError(
+                f"获取 tenant_access_token 失败: {token_payload.get('msg')}"
+            )
+        tenant_access_token = str(
+            token_payload.get("tenant_access_token") or ""
+        ).strip()
+        if not tenant_access_token:
+            raise RuntimeError("飞书未返回 tenant_access_token")
+
+        bot_response = requests.get(
+            f"{self.domain.value}/open-apis/bot/v3/info",
+            headers={"Authorization": f"Bearer {tenant_access_token}"},
+            timeout=10,
+        )
+        bot_response.raise_for_status()
+        bot_payload = bot_response.json()
+        if int(bot_payload.get("code") or 0) != 0:
+            raise RuntimeError(f"获取机器人信息失败: {bot_payload.get('msg')}")
+        bot_open_id = str((bot_payload.get("bot") or {}).get("open_id") or "").strip()
+        if not bot_open_id:
+            raise RuntimeError("飞书机器人信息缺少 open_id")
+        self._bot_open_id = bot_open_id
+        return bot_open_id
+
     def _strip_bot_mention(self, text: str, mentions: List) -> str:
         """去除 @机器人 的文本"""
         result = text

examples/tencent_realtime_control/test_feishu_natural_commands.py

@@ -14,7 +14,11 @@ for path in (ROOT, HERE):
     if str(path) not in sys.path:
         sys.path.insert(0, str(path))
 
-from agent.tools.builtin.feishu.feishu_client import ChatType, FeishuMessageEvent
+from agent.tools.builtin.feishu.feishu_client import (
+    ChatType,
+    FeishuClient,
+    FeishuMessageEvent,
+)
 from command_intent_parser import CommandIntentParser, IntentParserConfig
 from feishu_command_service import FeishuCommandService
 from operator_commands import (
@@ -186,6 +190,39 @@ class FeishuAuthorizationTest(unittest.TestCase):
         )
 
 
+class FeishuMentionDetectionTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.client = FeishuClient.__new__(FeishuClient)
+        self.client._bot_open_id = "ou_bot"
+
+    @staticmethod
+    def _mention(open_id: str) -> Mock:
+        mention = Mock()
+        mention.id.open_id = open_id
+        return mention
+
+    def test_mentioning_another_user_is_not_bot_mention(self) -> None:
+        self.assertFalse(
+            self.client._check_bot_mentioned([self._mention("ou_other")])
+        )
+
+    def test_mentioning_bot_is_bot_mention(self) -> None:
+        self.assertTrue(
+            self.client._check_bot_mentioned([self._mention("ou_bot")])
+        )
+
+    def test_missing_bot_identity_fails_closed(self) -> None:
+        self.client._bot_open_id = None
+        with patch.object(
+            self.client,
+            "_load_bot_open_id",
+            side_effect=RuntimeError("unavailable"),
+        ):
+            self.assertFalse(
+                self.client._check_bot_mentioned([self._mention("ou_other")])
+            )
+
+
 class PreviewSnapshotTest(unittest.TestCase):
     class FakeTencent:
         def get_ads(self, account_id: int):