Coverage for agentos/channels/adapters/feishu.py: 0%

1""" 

2AgentOS Channels — 飞书适配器。 

3 

4Webhook 规范: https://open.feishu.cn/document/server-docs/im-v1/message-content-description 

5 

6特性: 

7 - JSON 报文解析 

8 - 应用 Token + tenant access token 双 token 管理 

9 - 卡片消息支持 

10 - 消息回复（被动 + 主动） 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16import time 

17 

18import httpx 

19 

20from agentos.channels.base import BaseChannelAdapter, ChannelConfig, ReplyResult 

21from agentos.channels.message import ChannelMessage, ChannelType, MessageType 

22 

23 

24class FeishuAdapter(BaseChannelAdapter): 

25 """飞书适配器。""" 

26 

27 channel_type = ChannelType.FEISHU 

28 

29 def __init__(self, config: ChannelConfig): 

30 super().__init__(config) 

31 self._app_token: str = "" 

32 self._tenant_token: str = "" 

33 self._token_expires: float = 0 

34 

35 # ── Webhook ── 

36 

37 def verify_signature(self, raw_body: bytes, headers: dict) -> bool: 

38 """验证飞书事件订阅签名。 

39 

40 签名算法: Base64Encode(SHA256(timestamp + nonce + encrypt_key)) 

41 文档: https://open.feishu.cn/document/server-docs/event-subscription-guide/event-subscription-configure-/encrypt-key-encryption-configuration- 

42 """ 

43 import base64 

44 import hashlib 

45 

46 timestamp = headers.get("X-Lark-Request-Timestamp", "") 

47 nonce = headers.get("X-Lark-Request-Nonce", "") 

48 signature = headers.get("X-Lark-Signature", "") 

49 

50 encrypt_key = self.config.encoding_aes_key or self.config.verify_token 

51 if not all([timestamp, nonce, signature, encrypt_key]): 

52 return False 

53 

54 raw = f"{timestamp}{nonce}{encrypt_key}" 

55 computed = base64.b64encode(hashlib.sha256(raw.encode()).digest()).decode() 

56 return signature == computed 

57 

58 def parse_webhook(self, raw_body: bytes, headers: dict) -> ChannelMessage | list[ChannelMessage]: 

59 data = json.loads(raw_body.decode("utf-8")) 

60 # 飞书事件格式: {"schema": "2.0", "header": {...}, "event": {...}} 

61 event = data.get("event", data) 

62 header = data.get("header", {}) 

63 

64 # 处理 URL 验证 

65 if data.get("type") == "url_verification": 

66 return ChannelMessage( 

67 msg_id="url_verify", 

68 channel=ChannelType.FEISHU, 

69 msg_type=MessageType.EVENT, 

70 content=data.get("challenge", ""), 

71 reply_token=data.get("token", ""), 

72 extra={"is_challenge": True, "challenge": data.get("challenge", "")}, 

73 ) 

74 

75 msg_type_str = event.get("message", {}).get("message_type", "text") 

76 msg_type_map = { 

77 "text": MessageType.TEXT, "image": MessageType.IMAGE, 

78 "audio": MessageType.VOICE, "media": MessageType.FILE, 

79 "file": MessageType.FILE, "post": MessageType.TEXT, 

80 } 

81 msg_type = msg_type_map.get(msg_type_str, MessageType.TEXT) 

82 

83 message = event.get("message", {}) 

84 content = "" 

85 if msg_type_str == "text": 

86 content = json.loads(message.get("content", "{}")).get("text", "") 

87 elif msg_type_str == "post": 

88 content = str(message.get("content", ""))[:200] 

89 

90 sender = event.get("sender", {}) 

91 sender_id = sender.get("sender_id", {}).get("open_id", "") 

92 

93 return ChannelMessage( 

94 msg_id=header.get("event_id", event.get("message", {}).get("message_id", "")), 

95 channel=ChannelType.FEISHU, 

96 msg_type=msg_type, 

97 content=content, 

98 sender_id=sender_id, 

99 sender_name="", 

100 timestamp=float(header.get("create_time", str(int(time.time() * 1000)))) / 1000, 

101 conversation_id=event.get("message", {}).get("chat_id", ""), 

102 reply_token=event.get("message", {}).get("message_id", ""), 

103 media_url=message.get("image_key", ""), 

104 extra={ 

105 "tenant_key": header.get("tenant_key"), 

106 "event_type": header.get("event_type"), 

107 "chat_type": event.get("message", {}).get("chat_type", "p2p"), 

108 "root_id": event.get("message", {}).get("root_id"), 

109 "parent_id": event.get("message", {}).get("parent_id"), 

110 }, 

111 ) 

112 

113 def build_reply(self, msg: ChannelMessage, reply_text: str) -> str: 

114 return json.dumps({ 

115 "msg_type": "text", 

116 "content": json.dumps({"text": reply_text}), 

117 }) 

118 

119 # ── 主动推送 ── 

120 

121 async def send_message(self, user_id: str, content: str, msg_type: str = "text") -> ReplyResult: 

122 token = await self.get_access_token() 

123 url = "https://open.feishu.cn/open-apis/im/v1/messages" 

124 payload = { 

125 "receive_id": user_id, 

126 "msg_type": "text", 

127 "content": json.dumps({"text": content}), 

128 } 

129 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} 

130 async with httpx.AsyncClient() as client: 

131 resp = await client.post(url, params={"receive_id_type": "open_id"}, json=payload, headers=headers, timeout=10) 

132 data = resp.json() 

133 if data.get("code") == 0: 

134 return ReplyResult(success=True, msg_id=data.get("data", {}).get("message_id", "")) 

135 return ReplyResult(success=False, error=f"feishu error {data.get('code')}: {data.get('msg')}") 

136 

137 async def send_image(self, user_id: str, image_url: str) -> ReplyResult: 

138 token = await self.get_access_token() 

139 url = "https://open.feishu.cn/open-apis/im/v1/messages" 

140 payload = { 

141 "receive_id": user_id, 

142 "msg_type": "image", 

143 "content": json.dumps({"image_key": image_url}), 

144 } 

145 headers = {"Authorization": f"Bearer {token}"} 

146 async with httpx.AsyncClient() as client: 

147 resp = await client.post(url, params={"receive_id_type": "open_id"}, json=payload, headers=headers, timeout=10) 

148 return ReplyResult(success=resp.json().get("code") == 0) 

149 

150 async def send_file(self, user_id: str, file_url: str, filename: str) -> ReplyResult: 

151 token = await self.get_access_token() 

152 url = "https://open.feishu.cn/open-apis/im/v1/messages" 

153 payload = { 

154 "receive_id": user_id, 

155 "msg_type": "file", 

156 "content": json.dumps({"file_key": file_url}), 

157 } 

158 headers = {"Authorization": f"Bearer {token}"} 

159 async with httpx.AsyncClient() as client: 

160 resp = await client.post(url, params={"receive_id_type": "open_id"}, json=payload, headers=headers, timeout=10) 

161 return ReplyResult(success=resp.json().get("code") == 0) 

162 

163 # ── Token ── 

164 

165 async def get_access_token(self) -> str: 

166 if self._tenant_token and time.time() < self._token_expires - 300: 

167 return self._tenant_token 

168 url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" 

169 payload = {"app_id": self.config.app_id, "app_secret": self.config.app_secret} 

170 async with httpx.AsyncClient() as client: 

171 resp = await client.post(url, json=payload, timeout=10) 

172 data = resp.json() 

173 self._tenant_token = data["tenant_access_token"] 

174 self._token_expires = time.time() + data.get("expire", 7200) 

175 return self._tenant_token