AgileConfig Python Client

这是一个参考 .NET AgileConfig.Client 协议实现的 Python 客户端。它直接连接 AgileConfig 节点，不需要额外的代理服务。

主要能力：

• 使用 HTTP Basic Auth 拉取已发布配置；
• 使用 WebSocket 接收 reload、ping、offline 实时指令；
• 多节点随机起点、顺序故障转移和自动重连；
• 与 .NET 客户端一致的 group:key 键格式和大小写不敏感读取；
• 本地配置缓存，可选兼容 .NET 客户端的 AES-ECB 加密；
• 服务注册、客户端心跳、注销和服务发现；
• 线程安全读取、配置快照和重载回调。

安装

pip install .

开发环境：

pip install -e ".[test]"
pytest

需要 Python 3.9 或更高版本。

快速开始

from agileconfig import AgileConfigClient, ClientOptions

options = ClientOptions(
    app_id="orders-api",
    secret="your-secret",
    nodes="http://localhost:5000,http://localhost:5001",
    env="DEV",
    name="orders-worker-1",
    tag="blue",
)

client = AgileConfigClient(options)
client.start()  # 首次拉取后在后台保持 WebSocket 连接
try:
    print(client.get("database:connection"))
    print(client["feature:enabled"])
finally:
    client.stop()

也可以用上下文管理器：

with AgileConfigClient(options) as client:
    timeout = client.get("http:timeout", "30")

只进行一次 HTTP 拉取，不启动 WebSocket：

client = AgileConfigClient(options)
loaded_from_server = client.load()

load() 在所有节点均失败时读取本地缓存，并返回 False。可通过 client.is_loaded_from_local 判断当前数据是否来自缓存。

appsettings.json

客户端能直接读取与 .NET 客户端相同结构的配置：

{
  "AgileConfig": {
    "appId": "orders-api",
    "secret": "your-secret",
    "nodes": "http://localhost:5000,http://localhost:5001",
    "name": "orders-worker-1",
    "tag": "blue",
    "env": "DEV",
    "httpTimeout": 30,
    "reconnectInterval": 5,
    "cache": {
      "enabled": true,
      "directory": "./agile-config-cache",
      "config_encrypt": false
    }
  }
}

client = AgileConfigClient.from_json("appsettings.json")
client.start()

配置更新回调

def on_reload(event):
    print("before:", event.old_configs)
    print("after:", event.new_configs)

remove_listener = client.add_reload_listener(on_reload)
client.start()

# 不再监听时调用
remove_listener()

回调在 WebSocket 工作线程中执行，耗时任务应交给应用自己的任务队列。

服务注册

from agileconfig import ServiceRegisterOptions

options.service_register = ServiceRegisterOptions(
    service_id="orders-api-01",       # 留空时自动生成 UUID
    service_name="orders-api",
    ip="127.0.0.1",
    port=8000,
    metadata=("v1", "zone-a"),
    heartbeat_mode="client",
    heartbeat_interval=30,
)

client = AgileConfigClient(options)
client.start()  # 自动注册并发送心跳；stop() 时注销

服务端主动健康检查时，将 heartbeat_mode 设为 server，并设置 check_url。

服务发现

from agileconfig import DiscoveryService

with DiscoveryService(client) as discovery:
    instance = discovery.random_one("orders-api", healthy_only=True)
    if instance:
        print(instance.as_host("http"))

服务发现会监听注册中心的 WebSocket 更新消息；网络不可用时使用独立的本地服务缓存。

与 .NET 客户端的对应关系

.NET	Python
ConfigClientOptions	ClientOptions
ConfigClient / IConfigClient	AgileConfigClient / ConfigClient
ConnectAsync()	start() / connect()
DisconnectAsync()	stop() / disconnect()
Load()	load()
Get(key)	get(key)
GetGroup(group)	get_group(group)
ReLoaded	add_reload_listener(callback)
IDiscoveryService	DiscoveryService

Python 的 client[key] 遵循标准映射语义：键不存在时抛出 KeyError；需要默认值时使用 client.get(key, default)。

日志

客户端使用标准库 logging，不会自行配置全局日志输出：

import logging

logging.basicConfig(level=logging.INFO)