Coverage for agentos/channels/base.py: 0%

1""" 

2AgentOS Channels — 基础适配器协议。 

3 

4所有渠道适配器均实现本协议，确保 MessageGateway 零差异调用。 

5""" 

6 

7from __future__ import annotations 

8 

9import hashlib 

10import hmac 

11from abc import ABC, abstractmethod 

12from collections.abc import Awaitable, Callable 

13from dataclasses import dataclass, field 

14 

15from agentos.channels.message import ChannelMessage, ChannelType 

16 

17 

18@dataclass 

19class ChannelConfig: 

20 """渠道配置。""" 

21 

22 channel: ChannelType 

23 enabled: bool = False 

24 webhook_path: str = "" # 接收 webhook 的 URL 路径 

25 webhook_port: int = 8080 

26 verify_token: str = "" # 签名校验 token 

27 app_id: str = "" 

28 app_secret: str = "" 

29 encoding_aes_key: str = "" # 加解密 key（微信系） 

30 corp_id: str = "" 

31 agent_id: str = "" 

32 bot_app_id: str = "" 

33 bot_token: str = "" 

34 bot_secret: str = "" 

35 extra: dict = field(default_factory=dict) 

36 

37 def to_dict(self) -> dict: 

38 return { 

39 "channel": self.channel.value, 

40 "enabled": self.enabled, 

41 "webhook_path": self.webhook_path, 

42 "webhook_port": self.webhook_port, 

43 "app_id": self.app_id, 

44 } 

45 

46 

47@dataclass 

48class ReplyResult: 

49 """回复结果。""" 

50 

51 success: bool 

52 msg_id: str = "" 

53 error: str = "" 

54 

55 

56CallbackType = Callable[[ChannelMessage], Awaitable[str | None]] 

57"""消息回调: 接收 ChannelMessage，返回可选的同步回复文本。""" 

58 

59 

60class BaseChannelAdapter(ABC): 

61 """渠道适配器基类。 

62 

63 每个渠道适配器负责: 

64 1. 接收 webhook → 验证签名 → 解析为 ChannelMessage 

65 2. 将 Agent Engine 的回复发送回渠道 

66 3. Token 管理与自动续期 

67 """ 

68 

69 channel_type: ChannelType 

70 config: ChannelConfig 

71 _on_message: CallbackType | None = None 

72 

73 def __init__(self, config: ChannelConfig): 

74 self.config = config 

75 

76 def set_callback(self, cb: CallbackType): 

77 """设置收到消息时的回调。""" 

78 self._on_message = cb 

79 

80 # ── Webhook 入口 ── 

81 

82 @abstractmethod 

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

84 """验证 webhook 签名。返回 True 表示验证通过。""" 

85 ... 

86 

87 @abstractmethod 

88 def parse_webhook( 

89 self, raw_body: bytes, headers: dict 

90 ) -> ChannelMessage | list[ChannelMessage]: 

91 """解析 webhook 原始报文为 ChannelMessage(s)。""" 

92 ... 

93 

94 # ── 被动回复（同步，在 webhook 响应中返回）── 

95 

96 @abstractmethod 

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

98 """构建被动回复报文（xml/json）。""" 

99 ... 

100 

101 # ── 主动推送（异步，通过 API 发送）── 

102 

103 @abstractmethod 

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

105 """主动推送消息到用户。""" 

106 ... 

107 

108 @abstractmethod 

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

110 """推送图片。""" 

111 ... 

112 

113 @abstractmethod 

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

115 """推送文件。""" 

116 ... 

117 

118 # ── Token 管理 ── 

119 

120 @abstractmethod 

121 async def get_access_token(self) -> str: 

122 """获取/刷新 access_token。""" 

123 ... 

124 

125 # ── 工具方法 ── 

126 

127 @staticmethod 

128 def make_signature(token: str, timestamp: str, nonce: str, *args: str) -> str: 

129 """通用签名算法（微信/企微/飞书/钉钉均适用）。""" 

130 parts = sorted([token, timestamp, nonce] + list(args)) 

131 return hashlib.sha1("".join(parts).encode()).hexdigest() 

132 

133 @staticmethod 

134 def hmac_sha256(key: str, data: str) -> str: 

135 return hmac.new(key.encode(), data.encode(), hashlib.sha256).hexdigest()