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 dataclasses import dataclass, field 

13from typing import Optional, Callable, Awaitable 

14 

15from agentos.channels.message import ChannelMessage, ChannelType 

16 

17 

18@dataclass 

19class ChannelConfig: 

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

21 channel: ChannelType 

22 enabled: bool = False 

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

24 webhook_port: int = 8080 

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

26 app_id: str = "" 

27 app_secret: str = "" 

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

29 corp_id: str = "" 

30 agent_id: str = "" 

31 bot_app_id: str = "" 

32 bot_token: str = "" 

33 bot_secret: str = "" 

34 extra: dict = field(default_factory=dict) 

35 

36 def to_dict(self) -> dict: 

37 return { 

38 "channel": self.channel.value, 

39 "enabled": self.enabled, 

40 "webhook_path": self.webhook_path, 

41 "webhook_port": self.webhook_port, 

42 "app_id": self.app_id, 

43 } 

44 

45 

46@dataclass 

47class ReplyResult: 

48 """回复结果。""" 

49 success: bool 

50 msg_id: str = "" 

51 error: str = "" 

52 

53 

54CallbackType = Callable[[ChannelMessage], Awaitable[Optional[str]]] 

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

56 

57 

58class BaseChannelAdapter(ABC): 

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

60 

61 每个渠道适配器负责: 

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

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

64 3. Token 管理与自动续期 

65 """ 

66 

67 channel_type: ChannelType 

68 config: ChannelConfig 

69 _on_message: Optional[CallbackType] = None 

70 

71 def __init__(self, config: ChannelConfig): 

72 self.config = config 

73 

74 def set_callback(self, cb: CallbackType): 

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

76 self._on_message = cb 

77 

78 # ── Webhook 入口 ── 

79 

80 @abstractmethod 

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

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

83 ... 

84 

85 @abstractmethod 

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

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

88 ... 

89 

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

91 

92 @abstractmethod 

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

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

95 ... 

96 

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

98 

99 @abstractmethod 

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

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

102 ... 

103 

104 @abstractmethod 

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

106 """推送图片。""" 

107 ... 

108 

109 @abstractmethod 

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

111 """推送文件。""" 

112 ... 

113 

114 # ── Token 管理 ── 

115 

116 @abstractmethod 

117 async def get_access_token(self) -> str: 

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

119 ... 

120 

121 # ── 工具方法 ── 

122 

123 @staticmethod 

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

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

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

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

128 

129 @staticmethod 

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

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