Clash API自动切换节点：进阶脚本与故障转移配置指南

external-controller API:自动切换的工作原理

Clash 的图形客户端负责展示节点、策略组和日志,真正执行测速、连接与切换的仍然是后台内核。只要内核开放了 external-controller,外部程序就可以通过 HTTP API 读取当前配置、查询代理状态、测试节点延迟,并向可切换的策略组发出选择指令。自动切换脚本的本质,就是把“读取状态—逐个测速—判断故障—更新策略组”这条人工流程固定下来。

API 默认只监听本机地址,例如 127.0.0.1:9090。这意味着脚本与 Clash 在同一台电脑或服务器上运行时,不需要把控制端口暴露到局域网。配置中的 secret 是 API 鉴权密钥,客户端通过 Authorization: Bearer 请求头提交。没有密钥的环境也能工作,但只适合隔离测试机;一旦控制端口被其他设备访问,攻击者可能读取订阅中的节点信息,甚至强制修改代理出口。

external-controller: 127.0.0.1:9090
secret: "change-this-to-a-long-random-value"

修改配置后需要在客户端重新加载配置或重启内核,然后再确认接口是否可用。可以先执行下面的读取请求,返回 JSON 就说明监听地址、端口和鉴权基本正确:

curl -H "Authorization: Bearer change-this-to-a-long-random-value" \
  http://127.0.0.1:9090/proxies

返回结果中,顶层 proxies 是所有代理和策略组的映射。普通节点通常带有 history、type 等字段,策略组则会带 all 节点列表和当前选中的 now 值。脚本不要根据节点在列表中的位置判断可用性,而应读取策略组的 all 字段,因为订阅更新后节点顺序可能改变。

策略组设计:先让内核具备故障转移能力

如果需求只是“节点坏了就换下一个”,优先考虑直接使用 mihomo 的 fallback 策略组。它会按照节点列表顺序测试可用性,当前节点失败后切换到下一个节点。url-test 更适合追求低延迟的场景,会根据测试结果选择较快节点;但低延迟并不等于稳定,在网络抖动明显的环境里,过短的测试间隔可能造成频繁跳转。

proxy-groups:
  - name: "自动故障转移"
    type: fallback
    proxies:
      - "节点-A"
      - "节点-B"
      - "节点-C"
    url: "https://connectivity-check.example.test/generate_204"
    interval: 300
    lazy: true

  - name: "低延迟优选"
    type: url-test
    proxies:
      - "节点-A"
      - "节点-B"
      - "节点-C"
    url: "https://connectivity-check.example.test/generate_204"
    interval: 600
    tolerance: 80

interval 是自动测试周期,单位为秒。生产环境不建议设置得过短,否则会增加节点请求、触发服务端限制,也会让策略组在短暂抖动时反复切换。lazy: true 可以让策略组只在实际使用时测速,适合节点数量较多但并非一直有流量的设备。tolerance 只对 url-test 的切换敏感度有影响,它不是“超时阈值”,不能代替脚本中的失败判定。

需要脚本介入时,通常把一个 select 策略组作为规则唯一出口,脚本只修改这个组的当前选择,规则本身不随节点变化。示例如下:

proxy-groups:
  - name: "脚本出口"
    type: select
    proxies:
      - "节点-A"
      - "节点-B"
      - "节点-C"
      - DIRECT

rules:
  - MATCH,脚本出口

这种结构便于回滚:脚本异常时,可以在客户端界面手动选择节点;如果策略组类型是 fallback 或 url-test,脚本还应避免与内核同时改写同一个组,否则日志里会出现来回切换,很难判断究竟是哪一方做出的决定。

Python 脚本:读取节点、判定延迟并执行切换

下面的示例只使用 Python 标准库,适合放在桌面电脑、小型服务器或定时任务中。脚本逻辑分为四层:调用 API 的通用函数、读取目标策略组、逐个节点测试、连续失败后的切换。它不会接触订阅原文,也不会把节点配置写回磁盘,因此风险范围比直接修改 YAML 更小。

#!/usr/bin/env python3
import json
import time
import urllib.error
import urllib.parse
import urllib.request

API = "http://127.0.0.1:9090"
SECRET = "change-this-to-a-long-random-value"
GROUP = "脚本出口"
TEST_URL = "https://connectivity-check.example.test/generate_204"
TIMEOUT_MS = 5000
FAIL_LIMIT = 2

def api(method, path, payload=None):
    headers = {
        "Authorization": "Bearer " + SECRET,
        "Content-Type": "application/json",
    }
    data = None
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        API + path, data=data, headers=headers, method=method
    )
    with urllib.request.urlopen(request, timeout=8) as response:
        raw = response.read().decode("utf-8")
        return json.loads(raw) if raw else {}

def delay(proxy_name):
    query = urllib.parse.urlencode({
        "url": TEST_URL,
        "timeout": str(TIMEOUT_MS),
    })
    path = "/proxies/" + urllib.parse.quote(proxy_name, safe="") + "/delay?" + query
    try:
        result = api("GET", path)
        value = int(result.get("delay", 0))
        return value if value > 0 else None
    except (urllib.error.URLError, ValueError, KeyError, TimeoutError):
        return None

def main():
    state = api("GET", "/proxies")
    group = state["proxies"][GROUP]
    candidates = [
        name for name in group.get("all", [])
        if name != "DIRECT" and name != "REJECT"
    ]
    current = group.get("now", "")
    healthy = []

    for name in candidates:
        ms = delay(name)
        print("{}: {} ms".format(name, ms if ms is not None else "failed"))
        if ms is not None:
            healthy.append((ms, name))

    if not healthy:
        print("no healthy node; keep current: {}".format(current))
        return

    healthy.sort()
    best_delay, best = healthy[0]
    if best != current:
        path = "/proxies/" + urllib.parse.quote(GROUP, safe="")
        api("PUT", path, {"name": best})
        print("switched: {} -> {} ({} ms)".format(current, best, best_delay))
    else:
        print("keep current: {} ({} ms)".format(current, best_delay))

if __name__ == "__main__":
    main()

调用单个节点延迟时,节点名称必须经过 URL 编码,因为名称可能包含空格、斜杠、括号或其他特殊字符。切换策略组使用 PUT /proxies/<group>,请求体是 JSON 格式的 {"name":"节点名"}。脚本中没有直接拼接未经编码的名称,这是实际部署时很容易遗漏的一点。

示例按最低延迟选择节点,但生产环境不建议只看一次结果就切换。可以把结果写入内存,要求当前节点连续两轮失败才切换;新节点则连续两轮成功后才接管。还可以加入迟滞条件,例如新节点至少比当前节点快 80 毫秒才切换,避免两者延迟接近时不断来回跳转。

脚本安全边界与异常处理

• API 超时与连接失败。控制 API 无响应时不要继续发送切换请求,保留当前节点并记录错误,避免脚本自身造成重复操作。
• 节点全部失败。不要自动切换到未知节点,也不要把 DIRECT 当成代理故障的默认答案。是否允许直连应由明确的安全策略决定。
• 组名与节点名变化。订阅更新后名称可能被服务商修改。启动时先检查策略组是否存在,候选列表为空时立即退出并报警。
• 重复切换。保存最近一次选择和连续失败计数,设置最短切换间隔,例如 10 分钟内最多切换一次。
• 日志脱敏。日志只记录策略组、节点显示名、延迟和错误类型,不要输出订阅 URL、secret 或完整配置内容。

日志分析与生产部署:从能运行到可维护

部署前先用手动命令验证三个环节:API 能否鉴权、策略组是否存在、每个候选节点是否能通过测试。只有这三项都正常,再加入定时运行。Clash 客户端的日志级别可以暂时调到 info 或 debug,重点观察配置加载、代理拨号、DNS 解析和策略组切换;长时间运行时应恢复到 info,避免日志增长过快。

Linux 可以使用 systemd timer 或 cron 定时执行,Windows 可以使用任务计划程序,macOS 则可以使用 launchd。无论使用哪种调度器,都应设置“网络服务启动后延迟执行”或在脚本开头增加 API 重试。Clash 刚启动时可能仍在加载配置,此时立即请求 /proxies 会得到连接失败,这不是节点故障。

# cron 示例:每 5 分钟执行一次,输出追加到独立日志
*/5 * * * * /usr/bin/python3 /opt/clash/node_switch.py >> /var/log/clash-node-switch.log 2>&1

生产环境还应明确切换策略:延迟异常不等于节点不可用,可以同时设置连接超时、连续失败次数和延迟上限。例如单次测试超过 5 秒记为失败,连续两次失败才切换;延迟高于 1500 毫秒但仍能连接时,先记录告警,连续三轮超过阈值再执行切换。这样既能应对真正断连,也不会因为一次瞬时拥塞导致业务中断。

最后保留一个人工接管入口。发生误切换时,在客户端策略组中手动选回稳定节点,暂停定时任务,查看脚本日志与 Clash 内核日志的时间线,确认是测试地址不可达、DNS 异常、节点服务端限流,还是 API 选择请求失败。自动化的目标不是完全取消人工判断,而是把重复测速和初步故障转移交给脚本,让人工只处理异常情况。

配置检查清单

完成部署后,可以按下面的清单做一次最终验收:

1. 确认 external-controller 仅监听必要地址,并已配置 secret。
2. 确认规则指向脚本控制的策略组,而不是某个固定节点名称。
3. 手动调用 /proxies 与单节点 /delay,确认名称编码和鉴权正常。
4. 模拟当前节点超时或断开,观察脚本是否等待达到失败阈值后再切换。
5. 检查切换后现有连接与新建连接的表现,部分长连接可能需要重新建立。
6. 确认日志不会泄露订阅链接、secret 或完整代理参数,并设置日志轮转。

如果客户端已经支持稳定的 fallback 或 url-test 策略组,优先使用内核自带能力;只有在需要自定义健康检查、冷却时间、告警通知或跨设备调度时,再引入外部脚本。这样可以减少维护组件,同时保留 API 自动化带来的可观测性与控制能力。