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

1""" 

2AgentOS Channels — 企业微信适配器。 

3 

4Webhook 规范: https://developer.work.weixin.qq.com/document/path/90238 

5 

6特性: 

7 - XML/JSON 双报文解析 

8 - SHA1 签名验证 

9 - 被动回复 + 主动群机器人 webhook 推送 

10 - access_token 自动续期 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16import time 

17import xml.etree.ElementTree as ET 

18 

19import httpx 

20 

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

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

23 

24 

25class WeComAdapter(BaseChannelAdapter): 

26 """企业微信适配器。""" 

27 

28 channel_type = ChannelType.WECOM 

29 

30 def __init__(self, config: ChannelConfig): 

31 super().__init__(config) 

32 self._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 params = headers.get("x-wx-params", {}) 

40 msg_signature = params.get("msg_signature", "") 

41 timestamp = str(params.get("timestamp", "")) 

42 nonce = str(params.get("nonce", "")) 

43 signature = self.make_signature(self.config.verify_token, timestamp, nonce, "") 

44 return msg_signature == signature 

45 

46 def parse_webhook( 

47 self, raw_body: bytes, headers: dict 

48 ) -> ChannelMessage | list[ChannelMessage]: 

49 text = raw_body.decode("utf-8") 

50 data = json.loads(text) if text.strip().startswith("{") else self._parse_xml(text) 

51 msg_type_str = data.get("MsgType", data.get("msgtype", "text")) 

52 msg_type_map = { 

53 "text": MessageType.TEXT, 

54 "image": MessageType.IMAGE, 

55 "voice": MessageType.VOICE, 

56 "video": MessageType.VIDEO, 

57 "file": MessageType.FILE, 

58 "event": MessageType.EVENT, 

59 } 

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

61 content = "" 

62 if msg_type_str == "text": 

63 content = data.get("Content", data.get("text", {}).get("content", "")) 

64 elif msg_type_str == "image": 

65 content = "[图片]" 

66 

67 return ChannelMessage( 

68 msg_id=data.get("MsgId", "") or "", 

69 channel=ChannelType.WECOM, 

70 msg_type=msg_type, 

71 content=content, 

72 sender_id=data.get("FromUserName", data.get("UserID", "")), 

73 sender_name=data.get("Name", ""), 

74 timestamp=float(data.get("CreateTime", time.time())), 

75 conversation_id=data.get("ChatId", data.get("FromUserName", "")), 

76 media_url=data.get("PicUrl", ""), 

77 media_id=data.get("MediaId", ""), 

78 extra={ 

79 "to_user": data.get("ToUserName"), 

80 "agent_id": data.get("AgentID"), 

81 "msg_type_raw": msg_type_str, 

82 "webhook_url": data.get("WebhookUrl", ""), 

83 "chat_type": data.get("ChatType", "single"), 

84 }, 

85 ) 

86 

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

88 if msg.extra.get("webhook_url"): 

89 return json.dumps({"msgtype": "text", "text": {"content": reply_text}}) 

90 to_user = msg.extra.get("to_user", msg.sender_id) 

91 create_time = int(time.time()) 

92 return ( 

93 "<xml>" 

94 f"<ToUserName><![CDATA[{to_user}]]></ToUserName>" 

95 f"<FromUserName><![CDATA[{msg.sender_id}]]></FromUserName>" 

96 f"<CreateTime>{create_time}</CreateTime>" 

97 "<MsgType><![CDATA[text]]></MsgType>" 

98 f"<Content><![CDATA[{reply_text}]]></Content>" 

99 "</xml>" 

100 ) 

101 

102 # ── 主动推送（群机器人 webhook 或应用消息）── 

103 

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

105 # 如果有 webhook_url 则走群机器人推送 

106 webhook_url = self.config.extra.get("webhook_url", "") 

107 if webhook_url: 

108 async with httpx.AsyncClient() as client: 

109 resp = await client.post( 

110 webhook_url, 

111 json={ 

112 "msgtype": "text", 

113 "text": {"content": content}, 

114 }, 

115 timeout=10, 

116 ) 

117 return ReplyResult(success=resp.status_code == 200) 

118 

119 token = await self.get_access_token() 

120 url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" 

121 payload = { 

122 "touser": user_id, 

123 "msgtype": "text", 

124 "agentid": int(self.config.agent_id or 0), 

125 "text": {"content": content}, 

126 } 

127 async with httpx.AsyncClient() as client: 

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

129 data = resp.json() 

130 if data.get("errcode") == 0: 

131 return ReplyResult(success=True, msg_id=data.get("msgid", "")) 

132 return ReplyResult( 

133 success=False, error=f"wecom error {data.get('errcode')}: {data.get('errmsg')}" 

134 ) 

135 

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

137 token = await self.get_access_token() 

138 url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" 

139 payload = { 

140 "touser": user_id, 

141 "msgtype": "image", 

142 "agentid": int(self.config.agent_id or 0), 

143 "image": {"media_id": image_url}, 

144 } 

145 async with httpx.AsyncClient() as client: 

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

147 return ReplyResult(success=resp.json().get("errcode") == 0) 

148 

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

150 token = await self.get_access_token() 

151 url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" 

152 payload = { 

153 "touser": user_id, 

154 "msgtype": "file", 

155 "agentid": int(self.config.agent_id or 0), 

156 "file": {"media_id": file_url}, 

157 } 

158 async with httpx.AsyncClient() as client: 

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

160 return ReplyResult(success=resp.json().get("errcode") == 0) 

161 

162 # ── Token ── 

163 

164 async def get_access_token(self) -> str: 

165 if self._token and time.time() < self._token_expires - 300: 

166 return self._token 

167 url = ( 

168 "https://qyapi.weixin.qq.com/cgi-bin/gettoken" 

169 f"?corpid={self.config.corp_id}" 

170 f"&corpsecret={self.config.app_secret}" 

171 ) 

172 async with httpx.AsyncClient() as client: 

173 resp = await client.get(url, timeout=10) 

174 data = resp.json() 

175 self._token = data["access_token"] 

176 self._token_expires = time.time() + data.get("expires_in", 7200) 

177 return self._token 

178 

179 @staticmethod 

180 def _parse_xml(xml_str: str) -> dict: 

181 root = ET.fromstring(xml_str) 

182 return {child.tag: child.text for child in root}