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

1""" 

2AgentOS Channels — 消息网关。 

3 

4提供统一的 FastAPI 应用，一站式接入所有渠道 webhook： 

5 - POST /webhook/wechat 微信公众号 

6 - POST /webhook/wecom 企业微信 

7 - POST /webhook/feishu 飞书 

8 - POST /webhook/dingtalk 钉钉 

9 - POST /webhook/qq QQ 

10 

11同时提供管理接口： 

12 - GET /channels 列出所有已注册渠道 

13 - GET /health 健康检查 

14 

15依赖: 

16 pip install fastapi uvicorn 

17""" 

18 

19from __future__ import annotations 

20 

21import asyncio 

22import hashlib 

23import logging 

24import time 

25from typing import Any, Callable, Optional 

26 

27from agentos.channels.base import BaseChannelAdapter 

28from agentos.channels.message import ChannelMessage, ChannelType, ConversationContext 

29from agentos.channels.router import ChannelRouter 

30 

31logger = logging.getLogger("agentos.channels.gateway") 

32 

33_router = ChannelRouter() 

34 

35# ── 用户自定义消息处理器 ── 

36 

37_on_message: Optional[Callable[[ChannelMessage, ConversationContext], Any]] = None 

38 

39 

40def on_message(handler: Callable[[ChannelMessage, ConversationContext], Any]): 

41 """装饰器：注册消息处理器。 

42 

43 当任意渠道收到用户消息时，自动调用此 handler。 

44 handler 签名: async def handler(msg: ChannelMessage, ctx: ConversationContext) -> str | dict 

45 返回字符串直接作为回复；返回 dict 可包含 {"reply": "...", "image": "..."} 等。 

46 """ 

47 global _on_message 

48 _on_message = handler 

49 return handler 

50 

51 

52def _check_signature(adapter: BaseChannelAdapter, raw_data: bytes, headers: dict) -> bool: 

53 """检查请求签名（安全校验）。""" 

54 return adapter.verify_signature(raw_data, headers) 

55 

56 

57def _process_message_async(adapter: BaseChannelAdapter, raw_data: bytes, 

58 headers: dict) -> Any: 

59 """异步处理消息：签名校验 → 解析 → 路由 → 调用 handler → 回复。 

60 

61 此函数作为同步入口，内部使用 asyncio.run 驱动异步流程。 

62 符合 FastAPI 的同步/异步兼容要求。 

63 """ 

64 loop = asyncio.get_event_loop() 

65 if loop.is_running(): 

66 # 已在事件循环中，创建 task 

67 return _process_message_coro(adapter, raw_data, headers) 

68 return asyncio.run(_process_message_coro(adapter, raw_data, headers)) 

69 

70 

71async def _process_message_coro(adapter: BaseChannelAdapter, raw_data: bytes, 

72 headers: dict) -> Any: 

73 """消息处理协程。""" 

74 # 1. 签名校验 

75 if not _check_signature(adapter, raw_data, headers): 

76 logger.warning(f"[{adapter.channel_type.value}] 签名校验失败") 

77 return adapter.build_reply( 

78 ChannelMessage(channel=adapter.channel_type, content="signature_error"), 

79 "签名校验失败", success=False, 

80 ) 

81 

82 try: 

83 # 2. 解析消息 

84 msg: ChannelMessage = adapter.parse_webhook(raw_data, headers) 

85 except Exception as e: 

86 logger.exception(f"[{adapter.channel_type.value}] 消息解析失败: {e}") 

87 return adapter.build_reply( 

88 ChannelMessage(channel=adapter.channel_type, content="parse_error"), 

89 "消息解析失败", success=False, 

90 ) 

91 

92 if not msg.content and not msg.media_url: 

93 # 空消息/心跳 — 直接返回 200 

94 return adapter.build_reply(msg, "", success=True) 

95 

96 # 3. 获取会话上下文 

97 ctx = _router.get_context(msg) 

98 

99 # 4. 调用用户注册的消息处理器 

100 reply_text = "" 

101 try: 

102 if _on_message: 

103 result = _on_message(msg, ctx) 

104 if asyncio.iscoroutine(result): 

105 result = await result 

106 if isinstance(result, str): 

107 reply_text = result 

108 elif isinstance(result, dict): 

109 reply_text = result.get("reply", "") 

110 else: 

111 reply_text = f"[AgentOS Gateway] 收到来自 {msg.channel.value} 的消息，但未注册消息处理器。" 

112 except Exception as e: 

113 logger.exception(f"[{adapter.channel_type.value}] handler 异常: {e}") 

114 reply_text = f"处理失败: {e}" 

115 

116 # 5. 记录到上下文 

117 if reply_text: 

118 _router.update_context(msg.session_id or msg.sender_id, reply_text) 

119 

120 # 6. 构建回复 

121 return adapter.build_reply(msg, reply_text, success=True) 

122 

123 

124# ── FASTAPI 应用工厂 ── 

125 

126def create_app(title: str = "AgentOS Channel Gateway", 

127 version: str = "1.0.0", 

128 adapter_webhook_paths: Optional[dict[ChannelType, str]] = None, 

129 ) -> Any: 

130 """创建 FastAPI 应用实例。 

131 

132 Args: 

133 title: 应用标题 

134 version: 版本号 

135 adapter_webhook_paths: 渠道 → webhook 路径映射，如 {ChannelType.WECHAT: "/webhook/wechat"} 

136 

137 Returns: 

138 FastAPI app 实例 

139 

140 用法: 

141 from agentos.channels.gateway import create_app, on_message 

142 from agentos.channels.adapters import WeChatAdapter, WeComAdapter, FeishuAdapter 

143 from agentos.channels.message import ChannelType 

144 

145 app = create_app(adapter_webhook_paths={ 

146 ChannelType.WECHAT: "/webhook/wechat", 

147 ChannelType.WECOM: "/webhook/wecom", 

148 }) 

149 

150 @on_message 

151 async def handle_message(msg, ctx): 

152 return f"Echo: {msg.content}" 

153 

154 if __name__ == "__main__": 

155 import uvicorn 

156 uvicorn.run(app, host="0.0.0.0", port=8000) 

157 """ 

158 try: 

159 from fastapi import FastAPI, Request, HTTPException 

160 from fastapi.responses import JSONResponse, PlainTextResponse, Response 

161 except ImportError: 

162 raise ImportError( 

163 "agentos.channels.gateway requires fastapi and uvicorn. " 

164 "Install with: pip install fastapi uvicorn" 

165 ) 

166 

167 app = FastAPI(title=title, version=version) 

168 

169 # 默认 webhook 路径映射 

170 default_paths: dict[ChannelType, str] = { 

171 ChannelType.WECHAT_MP: "/webhook/wechat", 

172 ChannelType.WECOM: "/webhook/wecom", 

173 ChannelType.FEISHU: "/webhook/feishu", 

174 ChannelType.DINGTALK: "/webhook/dingtalk", 

175 ChannelType.QQ: "/webhook/qq", 

176 } 

177 

178 if adapter_webhook_paths: 

179 default_paths.update(adapter_webhook_paths) 

180 

181 # ── 注册渠道 webhook 路由 ── 

182 

183 def _make_webhook_handler(adapter: BaseChannelAdapter): 

184 """为每个渠道适配器创建 webhook handler。""" 

185 async def handler(request: Request): 

186 # 读取原始 body 

187 try: 

188 raw_data = await request.body() 

189 except Exception: 

190 raw_data = b"" 

191 

192 # 处理 GET 请求（飞书 URL 验证、微信 Token 验证） 

193 if request.method == "GET": 

194 query = dict(request.query_params) 

195 if adapter.channel_type == ChannelType.FEISHU: 

196 return adapter.build_reply( 

197 ChannelMessage(channel=ChannelType.FEISHU, content=""), 

198 query.get("challenge", ""), success=True, 

199 ) 

200 if adapter.channel_type == ChannelType.WECHAT_MP: 

201 # 微信公众号 URL 验证 

202 token = adapter.config.token 

203 sig = query.get("signature", "") 

204 ts = query.get("timestamp", "") 

205 nonce = query.get("nonce", "") 

206 echostr = query.get("echostr", "") 

207 if token: 

208 params = sorted([token, ts, nonce]) 

209 expected = hashlib.sha1("".join(params).encode()).hexdigest() 

210 if sig == expected: 

211 return PlainTextResponse(echostr) 

212 else: 

213 raise HTTPException(status_code=403, detail="Signature verification failed") 

214 return PlainTextResponse(echostr) 

215 return JSONResponse({"status": "ok"}) 

216 

217 # POST 请求 — 消息处理 

218 headers = dict(request.headers) 

219 try: 

220 result = await _process_message_coro(adapter, raw_data, headers) 

221 if result is None: 

222 return JSONResponse({"status": "ok"}) 

223 if isinstance(result, (str, bytes)): 

224 return Response(content=result if isinstance(result, bytes) else result.encode(), 

225 media_type="application/xml" if adapter.channel_type in 

226 (ChannelType.WECHAT_MP, ChannelType.WECOM) else "application/json") 

227 if isinstance(result, dict): 

228 return JSONResponse(result) 

229 return result 

230 except Exception as e: 

231 logger.exception(f"Gateway handler error: {e}") 

232 raise HTTPException(status_code=500, detail=str(e)) 

233 

234 return handler 

235 

236 # 从已注册适配器动态创建路由 

237 # 注意：路由在 app startup 事件中注册，因为适配器可能在 create_app 之后才注册 

238 @app.on_event("startup") 

239 async def _register_routes(): 

240 """应用启动时注册所有渠道路由。""" 

241 channel_adapters: dict[ChannelType, BaseChannelAdapter] = {} 

242 for webhook_path, adapter in _router._adapters.items(): 

243 channel_adapters[adapter.channel_type] = adapter 

244 for channel, adapter in _router._by_channel.items(): 

245 if channel not in channel_adapters: 

246 channel_adapters[channel] = adapter 

247 if channel in default_paths: 

248 _router._adapters[default_paths[channel]] = adapter 

249 

250 for channel, adapter in channel_adapters.items(): 

251 webhook_path = adapter.config.webhook_path or default_paths.get(channel, "") 

252 if webhook_path: 

253 _router._adapters[webhook_path] = adapter 

254 app.add_api_route(webhook_path, _make_webhook_handler(adapter), 

255 methods=["GET", "POST"], name=f"webhook_{channel.value}") 

256 

257 # ── 管理接口 ── 

258 

259 @app.get("/channels") 

260 async def list_channels(): 

261 return { 

262 "channels": _router.active_channels, 

263 "active_count": _router.active_count, 

264 "timestamp": time.time(), 

265 } 

266 

267 @app.get("/health") 

268 async def health_check(): 

269 return {"status": "healthy", "channels": _router.active_count, "timestamp": time.time()} 

270 

271 return app 

272 

273 

274# ── 便利函数 ── 

275 

276def get_router() -> ChannelRouter: 

277 """获取全局路由器实例。""" 

278 return _router 

279 

280 

281def register_adapter(adapter: BaseChannelAdapter, webhook_path: str = "") -> ChannelRouter: 

282 """手动注册渠道适配器到全局路由器。""" 

283 _router.register(adapter, webhook_path) 

284 return _router