opensre 网关聊天传输层gateway/transports架构指南从入站适配器到统一启停循环【免费下载链接】opensreBuild your own AI SRE agents. The open source toolkit for the AI era.项目地址: https://gitcode.com/GitHub_Trending/op/opensre本文围绕 opensre 项目中gateway/transports/目录的架构约定展开它如何把 Slack、Discord、Telegram、Buzz 四种聊天平台统一成一个平台一个包、彼此零依赖的入站适配器如何通过单一注册表与统一启停循环start_transports/stop_transports接入 AI SRE Agent 的 turn 运行器以及测试如何以关注点清单 known-gaps 台账的方式钉死每个传输的职责完整性。读完本文你将掌握该网关在聊天接入侧的分层边界、契约接口与新增传输的标准路径可直接对照 gateway/transports/AGENTS.md 与下方源码深入排查和二次开发。一、传输层是什么聊天平台侧的 ingress 适配器在 opensre 的网关gateway架构里gateway/transports/是所有入站聊天平台的归属地。用 API 框架的术语讲每个传输包就是一个ingress 适配器它负责完成平台侧消息的接收、鉴权、会话解析、turn 输出组装并把最终的处理交给 turn 运行器turn runner。根据 gateway/transports/AGENTS.md 的定位说明每个传输包需要实现四条主链路授权authorizes校验消息来源是否可信如 Slack 的签名校验、Telegram 的 token 校验解析会话resolves a session把平台消息映射到既有的 Agent 会话构建 turn 输出builds turn output把 Agent 的处理结果组装成该平台可发送的回复格式调用 turn runner通过统一的 turn 契约把消息交给 Agent 执行。而这几条链路的公共零件全部来自网关的共享层传输包本身不持有业务分发逻辑传输实现的 turn 契约是infrastructure.turn_host.turn_callback与infrastructure.turn_host.turn_output传输注册的注册表是gateway.transports.startup传输复用的每-turn 共享步骤是gateway.core.middleware启动所有传输的门面facade是gateway/startup.py。一句话理解分层传输包只负责平台差异所有Agent 语义都收敛到共享层这样新增一个聊天平台时不会触碰任何 Agent 逻辑。二、四个传输包与注册表映射当前网关内置四个聊天传输包每个包持有自己的 settings、入站 worker、安全逻辑、turn 输出与startup.py包启动入口运行形态slack/startup.start_slack_workerSocket Mode WebSocket 或 Events API HTTPdiscord/startup.start_discord_workerGateway WebSocket 长连接telegram/startup.start_telegram_worker长轮询long-pollbuzz/startup.start_buzz_worker提及轮询mention-poll这份映射并不是手写的文档而是注册表源码本身。gateway/transports/startup.py 中定义了唯一的TRANSPORTS元组TRANSPORTS: tuple[TransportRegistration, ...] ( TransportRegistration(TransportName.TELEGRAM, start_telegram_worker, polling for messages), TransportRegistration(TransportName.SLACK, start_slack_worker, inbound connected), TransportRegistration(TransportName.DISCORD, start_discord_worker, connected via gateway), TransportRegistration(TransportName.BUZZ, start_buzz_worker, polling for messages), )其中TransportRegistration是 gateway/transports/registration.py 中定义的冻结数据类一行即一个注册条目name传输名、start启动函数、running_status启动成功后上报的状态文案。值得注意的架构细节Web 不在这个注册表里。按 gateway/transports/names.py 的注释Web 是一个 channel 但不是 chat transport因此TransportName枚举只有四个成员class TransportName(StrEnum): TELEGRAM telegram SLACK slack DISCORD discord BUZZ buzzTransportName同时充当StartedGateway.transports字典与组件状态映射的键见 gateway/startup.py从而保证状态键与查询不会漂移。三、统一启停循环start_transports / stop_transports注册表与 worker 的启停循环都住在 gateway/transports/startup.py——它是这个包里唯一允许导入 peer 模块的文件且只允许导入各 peer 的startup子模块这样做保证了导入一个平台包绝不会连带加载另外三个平台的 SDK 栈。3.1 启动一次遍历、三类结局start_transportsgateway/transports/startup.py#L64-L97对注册表做单次遍历每个传输只会有三种结局正常启动把TransportHandle(name, worker, status)加入handles状态记为注册表中的running_status未配置抛出GatewayConfigurationError如缺少凭据记录为not configured (…)并跳过启动失败抛出GatewayTransportFailedError如 Discord readiness 超时记录为failed (…)并跳过。for registration in TRANSPORTS: try: worker, _settings registration.start(loggerlogger, handlerhandler) except GatewayConfigurationError as exc: logger.warning(%s chat disabled: %s, registration.name.capitalize(), exc) statuses[registration.name] fnot configured ({exc}) continue except GatewayTransportFailedError as exc: logger.warning(%s chat failed: %s, registration.name.capitalize(), exc) statuses[registration.name] ffailed ({exc}) continue ...两个关键语义所有传输在同一趟遍历中启动没有先后顺序依赖某个传输失败/未配置时网关继续服务其余成功启动的传输没有凭据 跳过而非报错这是刻意的设计让同一份网关代码既能在只配了 Telegram 的环境运行也能在四个平台全配齐的环境运行。ChatStartup返回值同时携带handles已启动的 worker 列表与statuses所有尝试过的传输的状态这样调用方无需深入状态映射就能上报哪些没配置。3.2 停止共享关闭预算stop_transportsgateway/transports/startup.py#L100-L117用ShutdownBudget对超时做统一管理逐个请求停止即使某个失败也继续尝试其余——一个卡住的传输不能拖住其他传输超时预算按顺序共享停止第一个 worker 花费的时间会从剩余预算中扣除budget.mark()/budget.consume(started)精确记账避免累计超时。budget ShutdownBudget(timeout) stopped True for handle in handles: started budget.mark() stopped handle.worker.stop(timeoutbudget.remaining) and stopped budget.consume(started) return stopped默认停止超时取自config.constants.gateway.DEFAULT_STOP_TIMEOUT_SECONDS。worker 的契约定义在 gateway/transports/worker.pyTransportWorker是一个仅含stop(*, timeout) - bool的 ProtocolTransportStarter则是入参不定、返回(worker, settings)二元组的 Callable。这个每个传输的 startup 返回 worker 加自有 settings 对象的约定让组合根能够持有解析后的配置而不必了解平台细节。四、每个传输的 startup平台差异收口处AGENTS.md 明确规定传输特定工作settings 加载、Discord readiness 等待留在各包的startup.py。四个包的实现正好展示了三种典型形态。4.1 Slack双入站模式 单副本去重护栏gateway/transports/slack/startup.py 支持两种入站传输由SlackInboundTransport枚举选择_TRANSPORT_STARTERS映射表加一行即加一种模式而不是加一个分支Socket Mode_start_socket_mode持有一条 WebSocket适合无法开放公网端口的部署Events API HTTP_start_events_api_http在自有端口上提供 HTTP 服务并接收带签名的 POST 请求。其中_submit_turn值得注意Slack 要求路由在 3 秒内应答因此 turn 不在请求线程内执行而是提交到共享执行器stack.executor.submit(stack.dispatcher.dispatch, message)请求快速返回Agent 在后台处理。Events API HTTP 模式还有一个强制的去重护栏_build_handled_event_repositorySlack 的投递是至少一次重试落在另一副本上会被重复受理因此该模式要求共享事件存储配置了DATABASE_URL→ 使用共享的HandledSlackEventRepository未配置 → 默认抛出GatewayConfigurationError提示设置DATABASE_URL或显式设置SLACK_GATEWAY_ALLOW_LOCAL_DEDUP1接受仅单副本安全的进程内去重即使选择了进程内回退也会输出 warning 日志保证这不是静默决定。4.2 Discordreadiness 等待后统一返回gateway/transports/discord/startup.py 的差异点是启动后要等待 Gateway 就绪worker.wait_until_ready(timeoutsettings.startup_timeout_seconds)超时则停止 worker 并抛GatewayTransportFailedError(startup timeout)。这样统一注册表可以把 Discord 与 Telegram/Slack 同等对待——start 返回时必然是活的 worker。4.3 Telegram 与 Buzz长轮询/提及轮询gateway/transports/telegram/startup.py 与 gateway/transports/buzz/startup.py 结构几乎一致加载平台 settings创建PollingBackgroundworker注入initialize_*_polling_runtime/shutdown_*_polling_runtime与 turn 回调。二者的共同点是把轮询运行时作为参数注入传输包本身不持有 Agent/分发逻辑。五、共享契约TurnCallback 与 turn 输出所有传输最终都汇入同一个回调契约。infrastructure/turn_host/turn_callback.py 定义了TurnCallback Callable[[str, SessionCore, TurnOutput, logging.Logger], None]签名固定为(text, session, output, logger)每条入站聊天消息最终都归结为对这一契约的一次调用。契约的反向依赖约束同样严格——TurnCallback所在模块不得 import 任何传输传输依赖契约、契约不依赖传输形成单向依赖环。turn 输出的合作式取消、单消息输出基类等约定如SingleMessageTurnOutput声明turn_cancel由infrastructure.turn_host.turn_output提供而会话解析、入站决策apply_inbound_decision、审批approvals、注意力attention、对话锁conversation locks等每-turn 共享步骤全部沉淀在 gateway/core/middleware含active_turns.py、approvals.py、attention.py、conversation_locks.py、identity_policy.py、inbound_decision.py、terminal_outcome.py。任何两个传输都需要的能力按规定必须提升到gateway.core而不是在某个传输里复制一份。六、组合根gateway/startup.py 与 StartedGatewayWeb 与聊天传输的组合发生在 gateway/startup.py它把start_transports的产物与 Web 服务并成一个StartedGatewaystart_gateway先启动 Web 服务器再启动所有聊天传输把web与各聊天状态合并进同一个statuses字典StartedGateway.transports以TransportName为键保存TransportHandlestop()时先停 Web独立超时WEB_STOP_TIMEOUT_SECONDS再用共享预算停所有聊天 workerGatewayController只持有这个不透明的StartedGateway不知道各平台的启动细节——传输特定工作因此不会泄漏到控制器。导入方向被严格钉死只有GatewayController导入gateway.startup也只有gateway.startup导入gateway.transports.startup传输 peer 之间不得 importgateway.startup或gateway.web。七、边界纪律与测试保障AGENTS.md 指出仅靠边界测试禁止做什么不足以保证每个传输必须包含什么为此仓库用两层测试把纪律固化下来。7.1 传输契约测试关注点清单 known-gaps 台账gateway/tests/test_transport_contract.py 用包源码 AST/文本检测而非 import验证每个传输是否实现了全部共享关注点避免把平台 SDK 拉进测试。必须实现的关注点清单_REQUIRED_CONCERNS包括关注点源码标记principal 作用域解析turn 数据归属任何存储访问前绑定def resolve_围绕 turn 的存储作用域绑定bound_storage_scope共享入站决策步骤会话生命周期走共享实现apply_inbound_decisionturn 超时设置挂起的 turn 不能永久挂起会话turn_timeout_seconds停止命令处理用户可终止运行中的 turnis_stop_command信用计量绑定turn 在容量判定前绑定计费bound_turn_meteringturn 输出声明turn_cancel配合宿主侧取消self.turn_cancel/SingleMessageTurnOutput台账_KNOWN_GAPS当前为空——即每个传输都实现了全部关注点。台账语义是账本而非白名单断言是精确相等补上一个缺口必须删除对应条目台账只能缩小新传输自动被发现、不得手写加入因此新传输要么完全合规要么测试全红。测试还禁止三类被提升后又在本地复制的实现_FORBIDDEN_LOCAL_COPIES本地重新实现 identity-policy 存储def _load_policy、容量判定前消费信用consume_credits、轮换会话的哨兵字面量__ROTATE_SESSION__。对信用计量测试还做了 AST 级检查每次bound_turn_metering(...)调用的idempotency_key必须是含插值的 f-stringast.JoinedStr且含ast.FormattedValue防止常量或空 key 让客户端省略幂等头、从而重新引入双扣费问题。7.2 包边界测试gateway/tests/test_package_borders.py钉死 peer 之间的 import DAG保证一个平台 ≠ 四套 SDK 栈的隔离gateway/tests/discord/test_transport_borders.pyDiscord ↔ Slack 的额外隔离验证。八、新增一个聊天传输的标准路径综合 AGENTS.md 与源码约定为网关新增一个聊天平台的完整清单如下新建gateway/transports/name/包包含settings.py平台配置加载、入站 worker、security.py等安全逻辑、turn 输出组装、以及startup.py在startup.py中实现start_name_worker(*, logger, handler) - tuple[worker, settings]加载 settings、启动 worker未配置时抛GatewayConfigurationError启动失败抛GatewayTransportFailedError在 gateway/transports/names.py 的TransportName中加入成员并在 gateway/transports/startup.py 的TRANSPORTS元组中注册实现全部七个共享关注点见 7.1 清单——契约测试会自动发现新包并强制全绿不得import peer 包、importgateway.startup/gateway.web、在本地复制已提升到gateway.core的实现。按此路径新增平台只需要交付平台差异Agent 分发、会话、审批、计量、停止命令等能力全部继承自共享层。参考路径速查架构约定 gateway/transports/AGENTS.md注册表与启停循环 gateway/transports/startup.py、gateway/transports/registration.py、gateway/transports/worker.py、gateway/transports/names.py各平台 startup gateway/transports/slack/startup.py、gateway/transports/discord/startup.py、gateway/transports/telegram/startup.py、gateway/transports/buzz/startup.py共享契约与组合根 infrastructure/turn_host/turn_callback.py、gateway/core/middleware、gateway/startup.py契约与边界测试 gateway/tests/test_transport_contract.py、gateway/tests/test_package_borders.py、gateway/tests/discord/test_transport_borders.py【免费下载链接】opensreBuild your own AI SRE agents. The open source toolkit for the AI era.项目地址: https://gitcode.com/GitHub_Trending/op/opensre创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考