添加平台适配器
本指南介绍如何向 Hermes 网关添加新的消息平台。平台适配器将 Hermes 与外部消息服务（Telegram、Discord、WeCom 等）连接，使用户能够通过该服务与 Agent 交互。
架构概览
用户 ↔ 消息平台 ↔ 平台适配器 ↔ 网关运行器 ↔ AIAgent
每个适配器都继承 gateway/platforms/base.py 中的 BasePlatformAdapter，并实现：
connect()— 建立连接（WebSocket、长轮询、HTTP 服务器等）（抽象）disconnect()— 清洁关闭 （抽象）send()— 向聊天发送文本消息 （抽象）send_typing()— 显示正在输入指示器（可选覆盖）get_chat_info()— 返回聊天元数据（可选覆盖）
入站消息由适配器接收，通过 self.handle_message(event) 转发，基类将其路由到网关运行器。
插件路径（推荐）
插件系统允许您在不修改任何 Hermes 核心代码的情况下添加平台适配器。您的插件是一个包含两个文件的目录：
~/.hermes/plugins/my-platform/
PLUGIN.yaml # 插件元数据
adapter.py # 适配器类 + register() 入口点
PLUGIN.yaml
插件元数据。requires_env 和 optional_env 块会自动填充 hermes config UI 条目（参见下方在 hermes config 中暴露环境变量）。
name: my-platform
label: My Platform
kind: platform
version: 1.0.0
description: My custom messaging platform adapter
author: Your Name
requires_env:
- MY_PLATFORM_TOKEN # bare string works
- name: MY_PLATFORM_CHANNEL # or rich dict for better UX
description: "Channel to join"
prompt: "Channel"
password: false
optional_env:
- name: MY_PLATFORM_HOME_CHANNEL
description: "Default channel for cron delivery"
password: false
adapter.py
import os
from gateway.platforms.base import (
BasePlatformAdapter, SendResult, MessageEvent, MessageType,
)
from gateway.config import Platform, PlatformConfig
class MyPlatformAdapter(BasePlatformAdapter):
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform("my_platform"))
extra = config.extra or {}
self.token = os.getenv("MY_PLATFORM_TOKEN") or extra.get("token", "")
async def connect(self) -> bool:
# Connect to the platform API, start listeners
self._mark_connected()
return True
async def disconnect(self) -> None:
self._mark_disconnected()
async def send(self, chat_id, content, reply_to=None, metadata=None):
# Send message via platform API
return SendResult(success=True, message_id="...")
async def get_chat_info(self, chat_id):
return {"name": chat_id, "type": "dm"}
def check_requirements() -> bool:
return bool(os.getenv("MY_PLATFORM_TOKEN"))
def validate_config(config) -> bool:
extra = getattr(config, "extra", {}) or {}
return bool(os.getenv("MY_PLATFORM_TOKEN") or extra.get("token"))
def _env_enablement() -> dict | None:
token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
if not (token and channel):
return None
seed = {"token": token, "channel": channel}
home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
if home:
seed["home_channel"] = {"chat_id": home, "name": "Home"}
return seed
def register(ctx):
"""Plugin entry point — called by the Hermes plugin system."""
ctx.register_platform(
name="my_platform",
label="My Platform",
adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
check_fn=check_requirements,
validate_config=validate_config,
required_env=["MY_PLATFORM_TOKEN"],
install_hint="pip install my-platform-sdk",
# Env-driven auto-configuration — seeds PlatformConfig.extra from
# env vars before adapter construction. See "Env-Driven Auto-
# Configuration" section below.
env_enablement_fn=_env_enablement,
# Cron home-channel delivery support. Lets deliver=my_platform cron
# jobs route without editing cron/scheduler.py. See "Cron Delivery"
# section below.
cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
# Per-platform user authorization env vars
allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
# Message length limit for smart chunking (0 = no limit)
max_message_length=4000,
# LLM guidance injected into system prompt
platform_hint=(
"You are chatting via My Platform. "
"It supports markdown formatting."
),
# Display
emoji="💬",
)
# Optional: register platform-specific tools
ctx.register_tool(
name="my_platform_search",
toolset="my_platform",
schema={...},
handler=my_search_handler,
)
配置
用户在 config.yaml 中配置平台：
gateway:
platforms:
my_platform:
enabled: true
extra:
token: "..."
channel: "#general"
或通过环境变量（适配器在 __init__ 中读取）。
插件系统自动处理的内容
调用 ctx.register_platform() 时，以下集成点会自动处理 — 无需修改核心代码：
| 集成点 | 工作方式 |
|---|---|
| 网关适配器创建 | 在内置 if/elif 链之前检查注册表 |
| 配置解析 | Platform._missing_() 接受任何平台名称 |
| 已连接平台验证 | 调用注册表 validate_config() |
| 用户授权 | 检查 allowed_users_env / allow_all_env |
| 环境变量自动启用 | env_enablement_fn 为 PlatformConfig.extra + home_channel 提供种子数据 |
| YAML 配置桥接 | apply_yaml_config_fn 将 config.yaml 的键翻译为环境变量 / 额外配置 |
| 定时任务投递 | cron_deliver_env_var 使 deliver=<name> 生效 |
hermes config UI 条目 | plugin.yaml 中的 requires_env / optional_env 自动填充 |
| send_message 工具 | 通过活跃网关适配器路由 |
| Webhook 跨平台投递 | 检查注册表中的已知平台 |
/update 命令访问 | allow_update_command 标志 |
| 频道目录 | 枚举时包含插件平台 |
| 系统提示词提示 | platform_hint 注入 LLM 上下文 |
| 消息分块 | max_message_length 用于智能分割 |
| PII 脱敏 | pii_safe 标志 |
hermes status | 显示插件平台并带 (plugin) 标签 |
hermes gateway setup | 插件平台出现在设置菜单中 |
hermes tools / hermes skills | 插件平台在每平台配置中 |
| 令牌锁（多配置文件） | 在 connect() 中使用 acquire_scoped_lock() |
| 孤立配置警告 | 插件缺失时输出描述性日志 |
环境变量驱动的自动配置
大多数用户通过将环境变量放入 ~/.hermes/.env 来设置平台，而不是编辑 config.yaml。env_enablement_fn 钩子允许你的插件在适配器构造之前读取这些环境变量，使得 hermes gateway status、get_connected_platforms() 和定时任务投递在不实例化平台 SDK 的情况下看到正确的状态。
def _env_enablement() -> dict | None:
"""Seed PlatformConfig.extra from env vars.
Called by the platform registry during load_gateway_config().
Return None when the platform isn't minimally configured — the
caller then skips auto-enabling. Return a dict to seed extras.
The special 'home_channel' key is extracted and becomes a proper
HomeChannel dataclass on the PlatformConfig; every other key is
merged into PlatformConfig.extra.
"""
token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
if not (token and channel):
return None
seed = {"token": token, "channel": channel}
home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
if home:
seed["home_channel"] = {
"chat_id": home,
"name": os.getenv("MY_PLATFORM_HOME_CHANNEL_NAME", "Home"),
}
return seed
def register(ctx):
ctx.register_platform(
name="my_platform",
label="My Platform",
adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
check_fn=check_requirements,
validate_config=validate_config,
env_enablement_fn=_env_enablement,
# ... other fields
)
YAML→env 配置桥接
某些用户更喜欢在 config.yaml 中设置键（my_platform.require_mention，my_platform.allowed_channels 等），而不是使用环境变量。apply_yaml_config_fn 钩子让你的插件拥有这种翻译能力，而不是强迫核心 gateway/config.py 了解你平台的 YAML 架构。
import os
def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
"""将 config.yaml 中的 `my_platform:` 键翻译为环境变量 / 额外配置。
yaml_cfg — 完整的顶层已解析 config.yaml 字典
platform_cfg — 平台自己的子字典 (yaml_cfg.get("my_platform", {}))
可以直接修改 os.environ（使用 `not os.getenv(...)` 保护来
保留 环境变量 > YAML 的优先级）和/或返回一个字典合并到
PlatformConfig.extra。如果没有额外配置则返回 None 或 {}。
"""
if "require_mention" in platform_cfg and not os.getenv("MY_PLATFORM_REQUIRE_MENTION"):
os.environ["MY_PLATFORM_REQUIRE_MENTION"] = str(platform_cfg["require_mention"]).lower()
allowed = platform_cfg.get("allowed_channels")
if allowed is not None and not os.getenv("MY_PLATFORM_ALLOWED_CHANNELS"):
if isinstance(allowed, list):
allowed = ",".join(str(v) for v in allowed)
os.environ["MY_PLATFORM_ALLOWED_CHANNELS"] = str(allowed)
return None # 没有需要合并到 PlatformConfig.extra 的内容
def register(ctx):
ctx.register_platform(
name="my_platform",
...,
apply_yaml_config_fn=_apply_yaml_config,
)
这个钩子在 load_gateway_config() 期间被调用，在通用共享键循环（处理常见键，如 unauthorized_dm_behavior、notice_delivery、reply_prefix、require_mention 等）之后，在 _apply_env_overrides() 之前，所以你的插件只需要桥接平台特定的键。
钩子抛出的异常会被吞掉并记录在调试级别——行为不当的插件永远不会中止网关配置加载。
定时任务投递
要让 deliver=my_platform 定时任务路由到已配置的主频道，将 cron_deliver_env_var 设置为保存默认 chat/room/channel ID 的环境变量名：
ctx.register_platform(
name="my_platform",
...
cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
)
调度器在解析 deliver=my_platform 任务的主目标时读取此环境变量，并将该平台视为有效的定时任务目标。如果你的 env_enablement_fn 提供了 home_channel 字典（见上方），则它优先——cron_deliver_env_var 是在环境种子化之前运行的定时任务的后备。
进程外定时任务投递
cron_deliver_env_var 使你的平台成为一个被识别的 deliver= 目标。为了使定时任务在与网关分离的进程中运行（即 hermes cron run 与 hermes gateway 分离）时实际发送成功，需要注册一个 standalone_sender_fn：
async def _standalone_send(
pconfig,
chat_id,
message,
*,
thread_id=None,
media_files=None,
force_document=False,
):
"""打开一个临时连接 / 获取新 token，发送，然后关闭。"""
# ... 打开连接，发送消息，返回结果 ...
return {"success": True, "message_id": "..."}
# 或 {"error": "..."}
ctx.register_platform(
name="my_platform",
...
cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
standalone_sender_fn=_standalone_send,
)
为什么需要这个钩子：内置平台（Telegram、Discord、Slack 等）在 tools/send_message_tool.py 中提供了直接的 REST 辅助函数，因此 cron 可以在不持有同一进程中网关的情况下投递。插件平台在历史上依赖于 _gateway_runner_ref()，该函数在网关进程外返回 None，因此如果没有 standalone_sender_fn，cron 侧的发送会失败并显示 No live adapter for platform '<name>'。
该函数接收与实时适配器相同的 pconfig 和 chat_id，加上可选的 thread_id、media_files 和 force_document 关键字参数。返回 {"success": True, "message_id": ...} 被视为成功投递；返回 {"error": "..."} 会在 cron 的 delivery_errors 中显示该消息。函数内部引发的异常会被调度器捕获并报告为 Plugin standalone send failed: <reason>。参考实现位于 plugins/platforms/{irc,teams,google_chat}/adapter.py。
平台特定的慢速 LLM UX
某些平台有约束条件，会改变慢速 LLM 响应应该如何呈现：
- LINE 发出一个单次使用的 回复令牌，该令牌在入站事件后大约 60 秒过期。使用该令牌回复是免费的；回退到计费的 Push API 则不是。如果 LLM 在截止时间前没有完成，选择是"消耗付费 Push 配额"或"在令牌过期前用更聪明的方式处理它"。
- WhatsApp 在 24 小时后将会话标记为不活跃，此后仅接受模板消息。
- SMS 没有正在输入指示器或渐进式更新的——长响应只是看起来像机器人离线了。
这些是基础的 BasePlatformAdapter 无法预期的真实约束。插件接口有意留下了空间，让适配器可以在基础打字循环之上分层平台特定的 UX，而无需扩展 kwarg 列表。
模式：子类化 _keep_typing 以分层飞行中 UX
BasePlatformAdapter._keep_typing 是打字指示器心跳——它在 LLM 生成时作为后台任务运行，并在响应被投递时取消。要在阈值时分层平台特定行为（例如，在 45 秒时发送一个"仍在思考"气泡），请在适配器中覆盖 _keep_typing，在 super()._keep_typing() 旁边调度你自己的任务，并在 finally 中拆除它：
class LineAdapter(BasePlatformAdapter):
async def _keep_typing(self, chat_id: str, *args, **kwargs) -> None:
if self.slow_response_threshold <= 0:
await super()._keep_typing(chat_id, *args, **kwargs)
return
async def _fire_at_threshold() -> None:
try:
await asyncio.sleep(self.slow_response_threshold)
except asyncio.CancelledError:
raise
# 这里的平台特定工作 — 对于 LINE，使用缓存的回复令牌发送一个模板
# 按钮"获取答案"气泡，以便用户可以稍后通过
# 回发回调中的一个（免费）新回复令牌获取缓存的响应。
await self._send_slow_response_button(chat_id)
side_task = asyncio.create_task(_fire_at_threshold())
try:
await super()._keep_typing(chat_id, *args, **kwargs)
finally:
if not side_task.done():
side_task.cancel()
try:
await side_task
except (asyncio.CancelledError, Exception):
pass
关键点：
- 始终
await super()._keep_typing(...)。 打字指示器心跳是独立有用的——不要替换它，要在它之上分层。 - 在
finally中拆除副作用任务。 当 LLM 完成（或/stop取消运行）时，网关会取消打字任务。你的副作用任务也必须遵守该取消，否则它会滞留并可能在响应已投递后触发。 - 配对
interrupt_session_activity以在用户发出/stop时解析任何孤立的 UX 状态。对于 LINE，这意味着将回发缓存条目从PENDING转换为ERROR，以便持久的"获取答案"按钮在下次点击时投递"运行在完成前被中断"消息，而不是循环。
模式：子类化 send 以通过缓存路由，而非立即发送
如果你的慢速响应 UX 缓存响应以供稍后检索（LINE 的回发流程），你的 send 覆盖需要识别三种模式：
hermes_cli/config.py 在导入时扫描 plugins/platforms/*/plugin.yaml，并从 requires_env 和（可选的）optional_env 块自动填充 OPTIONAL_ENV_VARS。使用富字典格式来贡献正确的描述、提示、密码标志和 URL——CLI 设置 UI 免费获得这些。
# plugins/platforms/my_platform/plugin.yaml
name: my_platform-platform
label: My Platform
kind: platform
version: 1.0.0
description: >
My Platform gateway adapter for Hermes Agent.
author: Your Name
requires_env:
- name: MY_PLATFORM_TOKEN
description: "Bot API token from the My Platform console"
prompt: "My Platform bot token"
url: "https://my-platform.example.com/bots"
password: true
- name: MY_PLATFORM_CHANNEL
description: "Channel to join (e.g. #hermes)"
prompt: "Channel"
password: false
optional_env:
- name: MY_PLATFORM_HOME_CHANNEL
description: "Default channel for cron delivery (defaults to MY_PLATFORM_CHANNEL)"
prompt: "Home channel (or empty)"
password: false
- name: MY_PLATFORM_ALLOWED_USERS
description: "Comma-separated user IDs allowed to talk to the bot"
prompt: "Allowed users (comma-separated)"
password: false
支持的字典键： name（必填）、description、prompt、url、password（布尔值；省略时从 *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON 后缀自动检测）、category（默认为 "messaging"）。
裸字符串条目（- MY_PLATFORM_TOKEN）仍然有效——它们会从插件的 label 自动派生通用描述。如果 OPTIONAL_ENV_VARS 中已存在同名变量的硬编码条目，则硬编码胜出（向后兼容）；plugin.yaml 格式作为后备。
参考实现
参见 plugins/platforms/line/adapter.py 查看完整的 LINE 回发实现 —— 一个 RequestCache 状态机（PENDING → READY → DELIVERED，加上 /stop 的 ERROR），一个在阈值时触发模板按钮气泡的 _keep_typing 覆盖，一个通过缓存路由的 send 覆盖，以及一个解析孤立 PENDING 条目的 interrupt_session_activity 覆盖。
参考实现（插件路径）
参见仓库中的 plugins/platforms/irc/，这是一个完整的工作示例 —— 一个无外部依赖的完整异步 IRC 适配器。plugins/platforms/teams/ 涵盖 Bot Framework / Adaptive Cards，plugins/platforms/google_chat/ 涵盖基于 OAuth 的 REST API，而 plugins/platforms/line/ 涵盖带有平台特定慢速 LLM UX 的 webhook 驱动消息 API。
分步清单（内置路径）
此清单适用于直接向 Hermes 核心代码库添加平台 — 通常由核心贡献者为官方支持的平台完成。社区/第三方平台应使用上方的插件路径。
1. Platform 枚举
在 gateway/config.py 的 Platform 枚举中添加您的平台：
class Platform(str, Enum):
# ... existing platforms ...
NEWPLAT = "newplat"
2. 适配器文件
创建 gateway/platforms/newplat.py：
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
BasePlatformAdapter, MessageEvent, MessageType, SendResult,
)
def check_newplat_requirements() -> bool:
"""Return True if dependencies are available."""
return SOME_SDK_AVAILABLE
class NewPlatAdapter(BasePlatformAdapter):
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.NEWPLAT)
# Read config from config.extra dict
extra = config.extra or {}
self._api_key = extra.get("api_key") or os.getenv("NEWPLAT_API_KEY", "")
async def connect(self) -> bool:
# Set up connection, start polling/webhook
self._mark_connected()
return True
async def disconnect(self) -> None:
self._running = False
self._mark_disconnected()
async def send(self, chat_id, content, reply_to=None, metadata=None):
# Send message via platform API
return SendResult(success=True, message_id="...")
async def get_chat_info(self, chat_id):
return {"name": chat_id, "type": "dm"}
对于入站消息，构建 MessageEvent 并调用 self.handle_message(event)：
source = self.build_source(
chat_id=chat_id,
chat_name=name,
chat_type="dm", # or "group"
user_id=user_id,
user_name=user_name,
)
event = MessageEvent(
text=content,
message_type=MessageType.TEXT,
source=source,
message_id=msg_id,
)
await self.handle_message(event)
3. 网关配置（gateway/config.py）
三个接触点：
get_connected_platforms()— 添加对您平台所需凭证的检查load_gateway_config()— 添加令牌环境变量映射条目：Platform.NEWPLAT: "NEWPLAT_TOKEN"_apply_env_overrides()— 将所有NEWPLAT_*环境变量映射到配置
4. 网关运行器（gateway/run.py）
五个接触点：
_create_adapter()— 添加elif platform == Platform.NEWPLAT:分支_is_user_authorized()allowed_users 映射 —Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"_is_user_authorized()allow_all 映射 —Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"- 早期环境检查
_any_allowlist元组 — 添加"NEWPLAT_ALLOWED_USERS" - 早期环境检查
_allow_all元组 — 添加"NEWPLAT_ALLOW_ALL_USERS" _UPDATE_ALLOWED_PLATFORMSfrozenset — 添加Platform.NEWPLAT
5. 跨平台投递
gateway/platforms/webhook.py— 在投递类型元组中添加"newplat"cron/scheduler.py— 添加到_KNOWN_DELIVERY_PLATFORMSfrozenset 和_deliver_result()平台映射
6. CLI 集成
hermes_cli/config.py— 将所有NEWPLAT_*变量添加到_EXTRA_ENV_KEYShermes_cli/gateway.py— 在_PLATFORMS列表中添加条目，包含 key、label、emoji、token_var、setup_instructions 和 varshermes_cli/platforms.py— 添加带 label 和 default_toolset 的PlatformInfo条目（供skills_config和tools_configTUI 使用）hermes_cli/setup.py— 添加_setup_newplat()函数（可委托给gateway.py），并将元组添加到消息平台列表hermes_cli/status.py— 添加平台检测条目："NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")hermes_cli/dump.py— 在平台检测字典中添加"newplat": "NEWPLAT_TOKEN"
7. 工具
tools/send_message_tool.py— 在平台映射中添加"newplat": Platform.NEWPLATtools/cronjob_tools.py— 在投递目标描述字符串中添加newplat
8. 工具集
toolsets.py— 添加带_HERMES_CORE_TOOLS的"hermes-newplat"工具集定义toolsets.py— 将"hermes-newplat"添加到"hermes-gateway"包含列表
9. 可选：平台提示词
agent/prompt_builder.py — 如果您的平台有特定渲染限制（无 Markdown、消息长度限制等），在 _PLATFORM_HINTS 字典中添加条目。这会将平台特定的指导注入系统提示词：
_PLATFORM_HINTS = {
# ...
"newplat": (
"You are chatting via NewPlat. It supports markdown formatting "
"but has a 4000-character message limit."
),
}
并非所有平台都需要提示词 — 仅在 Agent 行为应有所不同时才添加。
10. 测试
创建 tests/gateway/test_newplat.py，覆盖：
- 从配置构建适配器
- 消息事件构建
- Send 方法（模拟外部 API）
- 平台特定功能（加密、路由等）
11. 文档
| 文件 | 需添加内容 |
|---|---|
website/docs/user-guide/messaging/newplat.md | 完整的平台设置页面 |
website/docs/user-guide/messaging/index.md | 平台对比表、架构图、工具集表、安全章节、下一步链接 |
website/docs/reference/environment-variables.md | 所有 NEWPLAT_* 环境变量 |
website/docs/reference/toolsets-reference.md | hermes-newplat 工具集 |
website/docs/integrations/index.md | 平台链接 |
website/sidebars.ts | 文档页面的侧边栏条目 |
website/docs/developer-guide/architecture.md | 适配器计数 + 列表 |
website/docs/developer-guide/gateway-internals.md | 适配器文件列表 |
对等审查
在将新平台 PR 标记为完成之前，与已有平台进行对等审查：
# 查找所有提及参考平台的 .py 文件
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"
# 查找所有提及新平台的 .py 文件
search_files "newplat" output_mode="files_only" file_glob="*.py"
# 在第一组中但不在第二组中的文件都是潜在缺口
对 .md 和 .ts 文件重复上述操作。逐一检查每个缺口 — 是平台枚举（需要更新）还是平台特定引用（跳过）？
常见模式
长轮询适配器
如果您的适配器使用长轮询（如 Telegram 或 Weixin），使用轮询循环任务：
async def connect(self):
self._poll_task = asyncio.create_task(self._poll_loop())
self._mark_connected()
async def _poll_loop(self):
while self._running:
messages = await self._fetch_updates()
for msg in messages:
await self.handle_message(self._build_event(msg))
回调/Webhook 适配器
如果平台将消息推送到您的端点（如 WeCom 回调），运行 HTTP 服务器：
async def connect(self):
self._app = web.Application()
self._app.router.add_post("/callback", self._handle_callback)
# ... start aiohttp server
self._mark_connected()
async def _handle_callback(self, request):
event = self._build_event(await request.text())
await self._message_queue.put(event)
return web.Response(text="success") # Acknowledge immediately
对于响应截止时间严格的平台（例如 WeCom 的 5 秒限制），始终立即确认，之后通过 API 主动投递 Agent 的回复。Agent 会话运行 3–30 分钟——在回调响应窗口内内联回复是不可行的。
令牌锁
如果适配器持有使用唯一凭证的持久连接，添加范围锁以防止两个配置文件使用相同凭证：
from gateway.status import acquire_scoped_lock, release_scoped_lock
async def connect(self):
if not acquire_scoped_lock("newplat", self._token):
logger.error("Token already in use by another profile")
return False
# ... connect
async def disconnect(self):
release_scoped_lock("newplat", self._token)
参考实现
| 适配器 | 模式 | 复杂度 | 适合参考的场景 |
|---|---|---|---|
bluebubbles.py | REST + webhook | 中等 | 简单 REST API 集成 |
weixin.py | 长轮询 + CDN | 高 | 媒体处理、加密 |
wecom_callback.py | 回调/webhook | 中等 | HTTP 服务器、AES 加密、多应用 |
telegram.py | 长轮询 + Bot API | 高 | 支持群组、线程的完整功能适配器 |