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

1""" 

2AgentOS Channels — QQ 机器人适配器。 

3 

4Webhook 规范: https://bot.q.qq.com/wiki/develop/api-v2/ 

5 

6特性: 

7 - WebSocket 长连接（QQ 官方推荐） + HTTP webhook 双模式 

8 - JSON 报文解析 

9 - Bot Token 管理 

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 QQAdapter(BaseChannelAdapter): 

25 """QQ 机器人适配器。""" 

26 

27 channel_type = ChannelType.QQ 

28 

29 def __init__(self, config: ChannelConfig): 

30 super().__init__(config) 

31 self._token: str = config.bot_token or "" 

32 self._token_expires: float = float("inf") # QQ Bot Token 由配置直接提供 

33 

34 # ── Webhook ── 

35 

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

37 """QQ Bot 暂不强验证签名。""" 

38 return True 

39 

40 def parse_webhook( 

41 self, raw_body: bytes, headers: dict 

42 ) -> ChannelMessage | list[ChannelMessage]: 

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

44 op = data.get("op", 0) 

45 t = data.get("t", "") 

46 event_data = data.get("d", {}) 

47 

48 # 处理不同类型的 QQ 事件 

49 if op == 10: 

50 # Hello 事件 

51 return ChannelMessage( 

52 msg_id="hello", 

53 channel=ChannelType.QQ, 

54 msg_type=MessageType.EVENT, 

55 content="hello", 

56 extra={"op": 10, "heartbeat_interval": event_data.get("heartbeat_interval", 0)}, 

57 ) 

58 

59 if op == 11: 

60 # Heartbeat ACK 

61 return ChannelMessage( 

62 msg_id="heartbeat_ack", 

63 channel=ChannelType.QQ, 

64 msg_type=MessageType.EVENT, 

65 content="heartbeat_ack", 

66 extra={"op": 11}, 

67 ) 

68 

69 # op == 0: Dispatch 事件 

70 msg_map = { 

71 "AT_MESSAGE_CREATE": "text", 

72 "MESSAGE_CREATE": "text", 

73 "DIRECT_MESSAGE_CREATE": "text", 

74 "C2C_MESSAGE_CREATE": "text", 

75 } 

76 msg_type_str = msg_map.get(t, "text") 

77 msg_type_map = {"text": MessageType.TEXT} 

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

79 

80 author = event_data.get("author", {}) 

81 content = event_data.get("content", "").strip() 

82 

83 # 去掉 @机器人 前缀 

84 if content.startswith("<@"): 

85 end = content.find(">") 

86 if end > 0: 

87 content = content[end + 1 :].strip() 

88 

89 return ChannelMessage( 

90 msg_id=event_data.get("id", ""), 

91 channel=ChannelType.QQ, 

92 msg_type=msg_type, 

93 content=content, 

94 sender_id=author.get("id", ""), 

95 sender_name=author.get("username", ""), 

96 timestamp=float(time.time()), 

97 conversation_id=event_data.get("channel_id", event_data.get("guild_id", "")), 

98 reply_token="", 

99 extra={ 

100 "guild_id": event_data.get("guild_id"), 

101 "channel_id": event_data.get("channel_id"), 

102 "member": event_data.get("member"), 

103 "event_type": t, 

104 "is_group": bool(event_data.get("guild_id")), 

105 }, 

106 ) 

107 

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

109 return json.dumps( 

110 { 

111 "msg_type": 0, 

112 "content": reply_text, 

113 "msg_id": msg.msg_id, 

114 "message_reference": {"message_id": msg.msg_id}, 

115 } 

116 ) 

117 

118 # ── 主动推送 ── 

119 

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

121 token = await self.get_access_token() 

122 # QQ Bot 发消息需要知道 channel_id 

123 channel_id = self.config.extra.get("channel_id", "") 

124 if not channel_id: 

125 return ReplyResult(success=False, error="no channel_id in config") 

126 

127 url = f"https://api.sgroup.qq.com/channels/{channel_id}/messages" 

128 headers = {"Authorization": f"Bot {self.config.app_id}.{token}"} 

129 payload = {"content": content, "msg_type": 0} 

130 

131 async with httpx.AsyncClient() as client: 

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

133 data = resp.json() 

134 msg_id = data.get("id", "") 

135 if msg_id: 

136 return ReplyResult(success=True, msg_id=msg_id) 

137 return ReplyResult(success=False, error=f"qq error: {data}") 

138 

139 async def send_c2c_message(self, user_id: str, content: str) -> ReplyResult: 

140 """发送私聊消息。""" 

141 token = await self.get_access_token() 

142 url = f"https://api.sgroup.qq.com/v2/users/{user_id}/messages" 

143 headers = {"Authorization": f"Bot {self.config.app_id}.{token}"} 

144 payload = {"content": content, "msg_type": 0} 

145 

146 async with httpx.AsyncClient() as client: 

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

148 data = resp.json() 

149 return ReplyResult(success="id" in data, msg_id=data.get("id", "")) 

150 

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

152 return await self.send_message(user_id, f"[图片] {image_url}") 

153 

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

155 return await self.send_message(user_id, f"文件: {filename}\n{file_url}") 

156 

157 # ── Token ── 

158 

159 async def get_access_token(self) -> str: 

160 return self._token or self.config.bot_token