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 collections.abc import Callable 

26from typing import Any 

27 

28from agentos.channels.base import BaseChannelAdapter 

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

30from agentos.channels.router import ChannelRouter 

31 

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

33 

34_router = ChannelRouter() 

35 

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

37 

38_on_message: Callable[[ChannelMessage, ConversationContext], Any] | None = None 

39 

40 

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

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

43 

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

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

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

47 """ 

48 global _on_message 

49 _on_message = handler 

50 return handler 

51 

52 

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

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

55 return adapter.verify_signature(raw_data, headers) 

56 

57 

58def _process_message_async(adapter: BaseChannelAdapter, raw_data: bytes, 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, headers: dict) -> Any: 

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

73 # 1. 签名校验 

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

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

76 return adapter.build_reply( 

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

78 "签名校验失败", 

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 "消息解析失败", 

90 success=False, 

91 ) 

92 

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

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

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

96 

97 # 3. 获取会话上下文 

98 ctx = _router.get_context(msg) 

99 

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

101 reply_text = "" 

102 try: 

103 if _on_message: 

104 result = _on_message(msg, ctx) 

105 if asyncio.iscoroutine(result): 

106 result = await result 

107 if isinstance(result, str): 

108 reply_text = result 

109 elif isinstance(result, dict): 

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

111 else: 

112 reply_text = ( 

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

114 ) 

115 except Exception as e: 

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

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

118 

119 # 5. 记录到上下文 

120 if reply_text: 

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

122 

123 # 6. 构建回复 

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

125 

126 

127# ── FASTAPI 应用工厂 ── 

128 

129 

130def create_app( 

131 title: str = "AgentOS Channel Gateway", 

132 version: str = "1.0.0", 

133 adapter_webhook_paths: dict[ChannelType, str] | None = None, 

134) -> Any: 

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

136 

137 Args: 

138 title: 应用标题 

139 version: 版本号 

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

141 

142 Returns: 

143 FastAPI app 实例 

144 

145 用法: 

146 from agentos.channels.gateway import create_app, on_message 

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

148 from agentos.channels.message import ChannelType 

149 

150 app = create_app(adapter_webhook_paths={ 

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

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

153 }) 

154 

155 @on_message 

156 async def handle_message(msg, ctx): 

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

158 

159 if __name__ == "__main__": 

160 import uvicorn 

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

162 """ 

163 try: 

164 from fastapi import FastAPI, HTTPException, Request 

165 from fastapi.responses import JSONResponse, PlainTextResponse, Response 

166 except ImportError: 

167 raise ImportError( 

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

169 "Install with: pip install fastapi uvicorn" 

170 ) 

171 

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

173 

174 # 默认 webhook 路径映射 

175 default_paths: dict[ChannelType, str] = { 

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

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

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

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

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

181 } 

182 

183 if adapter_webhook_paths: 

184 default_paths.update(adapter_webhook_paths) 

185 

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

187 

188 def _make_webhook_handler(adapter: BaseChannelAdapter): 

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

190 

191 async def handler(request: Request): 

192 # 读取原始 body 

193 try: 

194 raw_data = await request.body() 

195 except Exception: 

196 raw_data = b"" 

197 

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

199 if request.method == "GET": 

200 query = dict(request.query_params) 

201 if adapter.channel_type == ChannelType.FEISHU: 

202 return adapter.build_reply( 

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

204 query.get("challenge", ""), 

205 success=True, 

206 ) 

207 if adapter.channel_type == ChannelType.WECHAT_MP: 

208 # 微信公众号 URL 验证 

209 token = adapter.config.token 

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

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

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

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

214 if token: 

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

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

217 if sig == expected: 

218 return PlainTextResponse(echostr) 

219 else: 

220 raise HTTPException( 

221 status_code=403, detail="Signature verification failed" 

222 ) 

223 return PlainTextResponse(echostr) 

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

225 

226 # POST 请求 — 消息处理 

227 headers = dict(request.headers) 

228 try: 

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

230 if result is None: 

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

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

233 return Response( 

234 content=result if isinstance(result, bytes) else result.encode(), 

235 media_type=( 

236 "application/xml" 

237 if adapter.channel_type in (ChannelType.WECHAT_MP, ChannelType.WECOM) 

238 else "application/json" 

239 ), 

240 ) 

241 if isinstance(result, dict): 

242 return JSONResponse(result) 

243 return result 

244 except Exception as e: 

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

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

247 

248 return handler 

249 

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

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

252 @app.on_event("startup") 

253 async def _register_routes(): 

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

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

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

257 channel_adapters[adapter.channel_type] = adapter 

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

259 if channel not in channel_adapters: 

260 channel_adapters[channel] = adapter 

261 if channel in default_paths: 

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

263 

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

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

266 if webhook_path: 

267 _router._adapters[webhook_path] = adapter 

268 app.add_api_route( 

269 webhook_path, 

270 _make_webhook_handler(adapter), 

271 methods=["GET", "POST"], 

272 name=f"webhook_{channel.value}", 

273 ) 

274 

275 # ── 管理接口 ── 

276 

277 @app.get("/channels") 

278 async def list_channels(): 

279 return { 

280 "channels": _router.active_channels, 

281 "active_count": _router.active_count, 

282 "timestamp": time.time(), 

283 } 

284 

285 @app.get("/health") 

286 async def health_check(): 

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

288 

289 return app 

290 

291 

292# ── 便利函数 ── 

293 

294 

295def get_router() -> ChannelRouter: 

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

297 return _router 

298 

299 

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

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

302 _router.register(adapter, webhook_path) 

303 return _router