先划清自动切换的边界:探测失败不等于节点必然失效
节点连接中断后手动打开客户端、找到策略组并换一个节点,通常能暂时恢复访问,但在无人值守设备或长时间运行的服务上,这种处理不够及时。Clash 系列内核提供的 external-controller HTTP API,可以让本机脚本读取代理组当前选择、主动测试组内节点,并在符合条件时切换选择。客户端仍负责启动和管理内核;脚本只是通过控制接口调用内核能力,不会替代订阅更新、节点配置或操作系统网络管理。
先判断是否真的需要自定义脚本。mihomo 的 url-test、fallback 等策略组已能按测试结果选择或回退节点,适用于希望把探测与选择交给内核的场景。API 脚本更适合需要自定义切换条件、发送告警、记录切换历史,或希望仅在当前节点连续异常时才干预的场景。若只需常规故障回退,优先评估内核策略组,减少外部进程、接口凭据和维护负担。
API 探测的结果也有边界。延迟测试只能说明某个节点在指定 URL、指定时刻完成了一次请求;目标站点不可达、上游 DNS 异常、局域网断网或本机时间错误,都可能让健康节点看起来失败。反过来,测速成功也不能证明所有网站、UDP 流量或特定应用都正常。可靠方案应结合连续失败次数、合理间隔、多个候选节点和冷却时间,而不是一次请求失败就立即切换。
- 适合 API 脚本:需要将代理组状态接入监控、告警或自定义恢复流程。
- 先用内核策略组:只需要定期测试并自动选择可用节点,不需要额外业务逻辑。
- 不要混为一谈:切换代理组选择不会修复订阅失效、配置解析错误、DNS 设置错误或操作系统没有接管流量的问题。
启用控制接口并确认代理组结构
external-controller 指定控制 API 的监听地址。独立运行 mihomo 时,可在最终生效配置中设置控制端口与密钥;图形客户端可能通过配置覆写或启动参数提供这些值,因此要检查客户端实际启动的内核配置,而不是只编辑订阅原文。以下配置是需要合并的示例片段,不是完整配置文件。不要在同一层级重复创建第二个 external-controller 字段。
external-controller: 127.0.0.1:9090
secret: "替换为足够长的随机密钥"
将监听地址限制在 127.0.0.1,可让接口仅接受本机连接。不要为了让脚本方便调用就监听 0.0.0.0,也不要把控制端口直接映射到公网。控制 API 可以读取代理组、发起延迟测试并更改当前选择;拿到密钥的人可能改变设备的代理出口。确有跨设备管理需求时,应通过受控的 VPN 或管理网络访问,并同时配置防火墙、访问控制和强密钥;不要把密钥写入公开仓库、日志或截图。
运行内核后,先在本机确认接口可以访问。下例用环境变量传入密钥,避免将凭据直接写入命令历史;其中 代理组名称 应替换为配置里的实际名称。若使用 PowerShell、客户端自带终端或容器部署,环境变量设置方式会不同,但请求地址和 API 结构相同。
export CLASH_SECRET='替换为控制接口密钥'
curl -H "Authorization: Bearer $CLASH_SECRET" \
http://127.0.0.1:9090/proxies
curl -H "Authorization: Bearer $CLASH_SECRET" \
http://127.0.0.1:9090/proxies/代理组名称
GET /proxies 返回可用代理及代理组信息;读取具体策略组时,返回内容通常包含该组当前选择的 now 和可选项 all。切换 Selector 组时,向 PUT /proxies/{组名} 发送 JSON,例如 {"name":"节点名称"}。在构造 URL 时必须对组名和节点名做 URL 编码,名称包含空格、斜线或中文时尤其不能直接拼接原字符串。组内成员可能包含其他策略组、DIRECT 或拒绝策略;自动测速候选应筛选为真正的节点,不能把嵌套组名当成普通代理节点。
动手操作:连续失败后测试候选节点并切换
以下 Python 示例针对一个可由 API 手动选择的 Selector 策略组。它每隔一段时间只检查当前节点;当前节点连续达到失败阈值后,才逐个测试组内普通节点,并选择延迟最低且不超过上限的候选。延迟接口使用 GET /proxies/{节点名}/delay,通过 url 指定探测地址、通过 timeout 指定超时毫秒数;切换使用带密钥的 PUT /proxies/{组名}。请将组名和探测 URL 改为适合自己环境的值。
脚本使用 Python 标准库,无须安装第三方包。将密钥放入环境变量 CLASH_SECRET 后运行;不要把真实密钥替换进脚本并提交到代码仓库。探测 URL 应稳定、可从当前网络访问,并尽量与实际使用场景相关。公共检测地址也可能限流或发生故障,生产环境宜选用自己有权使用、稳定可用的健康检查目标。
import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen
API = "http://127.0.0.1:9090"
GROUP = "自动选择"
SECRET = os.environ.get("CLASH_SECRET", "")
TEST_URL = "https://www.gstatic.com/generate_204"
INTERVAL = 30
TIMEOUT_MS = 5000
MAX_DELAY_MS = 2500
FAIL_LIMIT = 3
COOLDOWN = 120
GROUP_TYPES = {"Selector", "URLTest", "Fallback", "LoadBalance", "Relay"}
failure_count = 0
last_switch = 0
def request_json(path, method="GET", payload=None):
if not SECRET:
raise RuntimeError("请先设置 CLASH_SECRET")
data = None if payload is None else json.dumps(payload).encode("utf-8")
req = Request(
API + path,
data=data,
method=method,
headers={
"Authorization": "Bearer " + SECRET,
"Content-Type": "application/json",
},
)
with urlopen(req, timeout=8) as response:
return json.loads(response.read().decode("utf-8"))
def delay_for(name):
path = "/proxies/" + quote(name, safe="") + "/delay?"
path += urlencode({"url": TEST_URL, "timeout": TIMEOUT_MS})
result = request_json(path)
delay = result.get("delay")
if not isinstance(delay, (int, float)):
raise RuntimeError("延迟接口未返回有效 delay")
return delay
def choose_backup():
proxy_data = request_json("/proxies").get("proxies", {})
group = proxy_data.get(GROUP)
if not group:
raise RuntimeError("找不到策略组:" + GROUP)
candidates = []
for name in group.get("all", []):
item = proxy_data.get(name, {})
if item.get("type") not in GROUP_TYPES and name not in {
"DIRECT", "REJECT", "REJECT-DROP"
}:
candidates.append(name)
results = []
for name in candidates:
try:
delay = delay_for(name)
print("候选节点:", name, delay, "ms")
if delay <= MAX_DELAY_MS:
results.append((delay, name))
except (HTTPError, URLError, TimeoutError, RuntimeError) as exc:
print("候选节点测试失败:", name, exc)
return min(results)[1] if results else None
while True:
try:
state = request_json("/proxies/" + quote(GROUP, safe=""))
current = state.get("now")
if not current:
raise RuntimeError("策略组响应中没有 now 字段")
try:
delay = delay_for(current)
print("当前节点:", current, delay, "ms")
failed = delay > MAX_DELAY_MS
except (HTTPError, URLError, TimeoutError, RuntimeError) as exc:
print("当前节点测试失败:", exc)
failed = True
failure_count = failure_count + 1 if failed else 0
if failure_count >= FAIL_LIMIT and time.time() - last_switch >= COOLDOWN:
backup = choose_backup()
if backup and backup != current:
request_json(
"/proxies/" + quote(GROUP, safe=""),
method="PUT",
payload={"name": backup},
)
print("已切换至:", backup)
last_switch = time.time()
failure_count = 0
else:
print("没有找到符合延迟阈值的备用节点")
except (HTTPError, URLError, TimeoutError, RuntimeError, ValueError) as exc:
print("本轮检查未完成:", exc)
time.sleep(INTERVAL)
示例中的 GROUP 必须对应一个手动选择型组;若使用 URLTest、Fallback 或负载均衡组,应先确认内核对该组的控制行为,不要让脚本与内核同时争夺选择权。候选筛选依据 API 返回的代理类型排除已知策略组类型与内置出站,但不同内核版本可能返回不同类型名称。首次运行应先打印并核实组的 all 列表,确认不会把其他策略组、拒绝策略或不希望使用的出口纳入候选。
FAIL_LIMIT = 3 表示连续三轮不合格才开始选备份;以 INTERVAL = 30 秒计算,正常情况下最早会在约 90 秒后触发。MAX_DELAY_MS 是本地的可接受延迟阈值,不是 API 的最大超时。候选节点测试失败会被略过;如果没有通过阈值的节点,脚本保持当前选择并继续检查,不会擅自改到 DIRECT。切换后设置冷却时间,降低节点间来回抖动的风险。
监控策略、权限保护与运行维护
定时任务应由一个明确的进程或服务负责,不要同时在客户端计划任务、系统计划任务和多个终端里启动同一脚本。多个实例会重复发起延迟请求,还可能在不同判断下反复切换。可通过 systemd、Windows 任务计划程序或容器编排工具管理进程;具体服务配置应匹配系统用户、Python 路径、工作目录和环境变量。运行账户只需读取脚本、访问本机控制端口和写入有限日志,不应默认获得不必要的管理员权限。
将切换动作纳入可观测记录,至少保存检查时间、组名、原节点、目标节点、延迟值、失败原因和 API 响应状态。避免记录完整密钥、订阅链接和节点认证参数。对于需要告警的环境,可在成功切换、连续无可用节点、API 多次不可达时分别触发通知;通知失败不应导致主循环退出。日志应按大小或时间轮转,避免长时间运行后占满磁盘。
- 探测频率:间隔过短会增加请求负担并放大瞬时抖动;可从 30~60 秒开始,再按业务容忍度调整。
- 失败判定:同时考虑连续失败次数和延迟上限;不要只因单次延迟偏高就切换。
- 候选排序:最低延迟不一定代表最佳业务出口,也可增加节点优先级、地域或维护名单。
- 回切规则:脚本示例只负责故障时选择备用节点,不自动切回原节点,避免原节点短暂恢复后造成来回抖动。
- 接口防护:本机监听、强密钥、限制进程权限,并在不需要时关闭控制接口;不要将密钥写进调试输出。
API 调用成功只代表控制接口接受了请求,不代表所有应用的网络连接立即迁移。已有连接可能继续使用原出站,部分客户端可能重建连接后才应用新选择;系统代理、TUN 和路由接管状态也独立于 API。切换后应使用实际业务请求验证新连接,同时查看客户端连接列表、内核日志和代理组当前选择,区分“选择已变更”与“业务已恢复”。
常见问题与排障顺序
API 返回 401 或 403 怎么办?
核对请求使用的密钥是否与运行配置中的 secret 一致,并确认请求头为 Authorization: Bearer 密钥。客户端可能在启动时覆盖了配置,导致你编辑的文件与实际运行值不同。若刚修改密钥,重启或按客户端要求重新载入内核后再测试。不要用移除认证的方式解决授权错误,也不要将含密钥的完整命令粘贴到公开日志。
延迟测试失败,但浏览器仍能访问怎么办?
先用同一探测 URL 手动请求当前节点的 /delay 接口,检查该 URL 是否可达、超时是否过短,以及内核是否支持该延迟测试接口。浏览器可能使用不同规则、DNS 路径或缓存连接;API 测试的目标与浏览器正在访问的服务也可能不同。更换为受控的健康检查地址并观察数轮,不要仅凭一次探测结果认定节点失效。
组里明明有节点,脚本为什么没有切换?
检查 API 返回的组名、now 和 all,再确认候选筛选没有将节点误认为嵌套组。也要核对候选延迟是否超过阈值、连续失败次数是否达到上限、冷却时间是否尚未结束,以及 PUT 的组名和 JSON 中的节点名称是否精确匹配。若运行的是自动测速组,内核可能自行决定成员,宜避免外部脚本同时改写其状态。
部署前还要做哪些检查?
检查 API 仅绑定预期接口、密钥未泄露、组内备用节点确实可用,并验证切换失败时脚本不会错误地选择直连。先以较长间隔和较高失败阈值观察,再逐步调整参数。变更前保存配置,记录原始节点选择;若脚本出现异常切换,停止服务并通过客户端界面恢复人工选择,然后检查日志、组结构和内核版本。