AIGC Auth Python SDK

Python 后端服务接入华策 AIGC 鉴权中心的 SDK 工具包，提供 10 大接入能力。

Node.js 服务端请使用 @huace/aigc-auth-client-node。 Node SDK 与本包共享 JWT、user/app/global Access Key 的鉴权语义，并提供固定版本的 npm 包清单；浏览器登录 SDK 仍位于 frontend/sdk，两者用途不同。

安装

pip install huace-aigc-auth-client

环境变量（必填）

AIGC_AUTH_APP_ID=your_app_id
AIGC_AUTH_APP_SECRET=your_app_secret
AIGC_AUTH_BASE_URL=https://your-auth-api-url

在鉴权中心创建应用后获取 APP_ID 和 APP_SECRET。

Nginx 代理（可选）

location /aigc-auth/ {
    proxy_pass https://aigc-auth.huacemedia.com/aigc-auth/;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}

SKILL 1: 鉴权中间件

为你的 FastAPI / Flask 应用加上统一鉴权拦截，未登录请求自动 401。

FastAPI — 中间件模式（推荐）

from fastapi import FastAPI, Request
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = FastAPI()
client = AigcAuthClient()
auth_mw = AuthMiddleware(client, exclude_paths=["/health", "/docs", "/openapi.json"])

@app.middleware("http")
async def auth(request: Request, call_next):
    return await auth_mw.fastapi_middleware(request, call_next)

@app.get("/me")
async def me(request: Request):
    user = request.state.user_info
    return {"username": user.username, "roles": user.roles}

排除路径配置：

FastAPI — 依赖注入

from fastapi import FastAPI, Depends
from huace_aigc_auth_client import AigcAuthClient, create_fastapi_auth_dependency, UserInfo

app = FastAPI()
client = AigcAuthClient()
get_user = create_fastapi_auth_dependency(client)

@app.get("/me")
async def me(user: UserInfo = Depends(get_user)):
    return {"username": user.username}

# 带权限校验
def require_perm(perm: str):
    async def check(user: UserInfo = Depends(get_user)):
        if not user.has_permission(perm):
            from fastapi import HTTPException
            raise HTTPException(403, "权限不足")
        return user
    return check

@app.get("/admin")
async def admin(user: UserInfo = Depends(require_perm("admin:access"))):
    return {"admin": True}

FastAPI — 装饰器

from huace_aigc_auth_client import AigcAuthClient, require_auth, UserInfo

client = AigcAuthClient()

@app.get("/protected")
@require_auth(client)
async def route(request: Request, user_info: UserInfo):
    return {"user": user_info.username}

@app.get("/editor")
@require_auth(client, permissions=["article:write", "article:edit"], any_permission=True)
async def editor(request: Request, user_info: UserInfo):
    return {"editor": True}

Flask

from flask import Flask, g, jsonify
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = Flask(__name__)
client = AigcAuthClient()
auth_mw = AuthMiddleware(client, exclude_paths=["/health", "/login"])

@app.before_request
def before():
    return auth_mw.flask_before_request()

@app.route("/me")
def me():
    user = g.user_info
    return jsonify({"username": user.username, "roles": user.roles})

SKILL 2: Token 验证与用户信息

主动验证 Token 或 Access Key，获取用户详情和权限。

验证 Token

from huace_aigc_auth_client import AigcAuthClient

client = AigcAuthClient()
result = client.verify_token("user-token")

if result.valid:
    print(result.user_id, result.username, result.expires_at)

获取用户信息（含角色、权限、用户组）

user = client.get_user_info(token)

print(user.username, user.nickname, user.email)
print(user.roles)         # ["admin", "editor"]
print(user.permissions)   # ["user:read", "user:write"]
print(user.groups)        # [{"id": 1, "name": "技术部", "code": "tech"}]

if user.is_admin or user.has_role("admin"):
    print("管理员")

if user.has_permission("user:write"):
    print("有写权限")

if user.has_any_permission(["user:read", "user:write"]):
    print("至少一个权限")

比较跨应用用户身份

Auth SDK 优先比较两个 Token 对应的统一 Auth 用户 ID；历史重复用户再回退比较 同一飞书应用范围内的 feishu_open_id：

same_user = client.compare_users(
    source_app_id=12,
    target_app_id=27,
    source_token=source_token,
    target_token=target_token,
)

目标应用还没有该用户时，可直接使用来源应用的 app_id 和用户 Token 完成幂等授权， 不需要配置来源应用 Secret：

user = client.authorize_user_by_token(source_app_id=12, source_token=source_token)

外部 SSO 飞书用户同步与临时 Token

当外部平台使用其他 SSO，但与鉴权中心使用同一企业飞书时，可以先幂等同步飞书用户， 再按本地用户 ID、飞书用户 ID、Open ID 或用户名签发短期 Token：

synced = client.sync_feishu_user(
    username="zhangsan",
    nickname="张三",
    department="研发中心",
    feishu_user_id="fs-user-id",
    feishu_open_id="ou-xxxxx",
    feishu_union_id="on-xxxxx",
    feishu_tenant_key="tenant-key",
)

temporary = client.issue_feishu_user_token(
    feishu_user_id="fs-user-id",
    expires_in=86400,  # 默认 1 天，最长 7 天
)
print(temporary["accessToken"], temporary["expiresAt"])

同步接口不会覆盖已存在用户资料，只会补齐跨应用访问授权。跨应用复用身份时应传入 feishu_tenant_key，避免不同租户中相同的飞书用户 ID 被错误合并。签发接口不会隐式创建 用户；用户不存在时会抛出错误码 40401，此时应先调用 sync_feishu_user()。

issue_feishu_user_token() 的 expires_in 单位为秒，默认 86400，最大 604800。 临时 Token 使用现有 access JWT 校验链路，可直接作为 Authorization: Bearer <token> 调用已接入 AIGC Auth 的服务。

验证 Access Key

result = client.verify_access_key("ak-xxxxx")
if result.valid:
    print(result.key_type, result.username, result.is_admin)
    # key_type: "app" / "global" / "user"

UserInfo 字段

SKILL 3: 用户查询与用户组

批量查询用户、分页搜索、获取用户组清单。

用户列表（分页 + 搜索）

# 第一页
result = client.list_users()
print(result["total"], result["items"])

# 搜索
result = client.list_users(keyword="张", page=1, page_size=10)

批量查询

# 按用户名
users = client.batch_get_users(["zhangsan", "lisi"])

# 按用户 ID
users = client.batch_get_users_by_ids([1, 2, 999])

获取用户组清单

groups = client.list_groups()
# [{"id": 1, "name": "技术部", "code": "tech", "description": "..."}, ...]

SKILL 4: 服务注册与配置拉取

服务启动时注册 IP + 端口，定时心跳上报，从鉴权中心拉取全局配置。

方式一：环境变量自动注册（推荐）

# .env — 设置 AIGC_SERVICE_NAME 即启用自动注册
AIGC_SERVICE_NAME=my-service
AIGC_SERVICE_TYPE=website        # website / plugin / mcp / other
AIGC_SERVICE_ENV=pro             # test / pre / pro
AIGC_SERVICE_PORT=8000
AIGC_SERVICE_TAGS=core,api
AIGC_SERVICE_DESCRIPTION=用户中心主服务

# 一行搞定：注册 + 心跳 + 配置 Pull
client = AigcAuthClient()

方式二：显式调用

client = AigcAuthClient(auto_register_service=False)

result = client.start_service(
    name="my-service",
    service_type="website",
    environment="pro",
    port=8000,
    tags=["core", "api"],
    description="用户中心主服务",
    heartbeat_interval=60,
    on_config_change=lambda cfg, ver: print(f"配置更新 v{ver}:", cfg),
)
print(f"service_id={result.service_id}, instance_id={result.instance_id}")

# 退出时注销
client.stop_service("my-service", "pro")

读取配置

# 从心跳缓存读取，无需额外请求
# 不传参时自动从环境变量 AIGC_SERVICE_NAME / AIGC_SERVICE_ENV 匹配
config = client.get_service_config()
db_url = config.get("DATABASE_URL")

# 多服务场景显式指定
config = client.get_service_config("my-service", "pro")

服务注册环境变量

Docker / 容器注意

容器内自动检测的 IP 可能是内网地址。两种应对：

1. 环境变量注入宿主机 IP：AIGC_SERVICE_IPS=203.0.113.10
2. 显式传入：start_service(ips=["203.0.113.10"], ...)

容器/k8s 建议设置 AIGC_SERVICE_INSTANCE_ID（如 Pod 名 + 端口），保证跨重启稳定。

服务唯一性模型

注册是幂等的：服务已存在时不覆盖平台已编辑的字段，仅更新本实例记录。

SKILL 5: Webhook 接收

接收鉴权中心推送的用户变更事件（创建、更新、删除）。

FastAPI

from fastapi import APIRouter
from huace_aigc_auth_client import register_webhook_router

router = APIRouter()

async def handle_webhook(data: dict) -> dict:
    event = data.get("event")
    print(f"收到事件: {event}")
    return {"status": "ok"}

register_webhook_router(
    router,
    handler=handle_webhook,
    prefix="/webhook",
    secret_env_key="AIGC_AUTH_WEBHOOK_SECRET",
)
# 端点: POST /webhook/auth

Flask

from huace_aigc_auth_client import register_flask_webhook_routes

def handle_webhook(data: dict) -> dict:
    return {"status": "ok"}

register_flask_webhook_routes(app, handler=handle_webhook)
# 端点: POST /api/webhook/auth

事件类型

签名验证

SDK 自动处理：从请求头 X-Webhook-Signature 读取签名，用环境变量 AIGC_AUTH_WEBHOOK_SECRET 的密钥做 HMAC-SHA256 验证。签名不匹配返回 401。

也可手动验证：

from huace_aigc_auth_client import verify_webhook_signature

is_valid = verify_webhook_signature(payload_bytes, signature_header, secret)

SKILL 6: API 统计与计费

自动收集 API 调用数据并上报到鉴权中心，支持动态计费规则。

自动采集

使用鉴权中间件（SKILL 1）时，默认启用 API 统计收集，无需额外配置。

手动初始化

from huace_aigc_auth_client import init_api_stats_collector

init_api_stats_collector(
    api_url="https://auth.example.com/sdk",
    app_id="your-app-id",
    app_secret="your-app-secret",
    batch_size=10,       # 批量提交大小
    flush_interval=5.0,  # 刷新间隔（秒）
    enabled=True,
)

手动上报

from huace_aigc_auth_client import collect_api_stat

collect_api_stat(
    api_path="/api/chat/completions",
    api_method="POST",
    status_code=200,
    response_time=1.2,
    token="user-token",
    error_message=None,
    request_params={"headers": {"X-Match-Username": "zhangsan"}},
    response_headers={
        "X-Match-Dynamic": "1",
        "X-Match-Cost": "0.0035",
    },
)

动态计费响应头

上游 API 通过响应头告诉鉴权中心如何计费：

规则：X-Match-Dynamic=1 → 创建动态计费规则；不设或非 1 → 静态规则（费用为 0）。

X-Match-Username

异步任务场景下，通过 request_params.headers 关联用户：

collect_api_stat(
    api_path="/api/task/run",
    api_method="POST",
    status_code=200,
    response_time=1.5,
    token=token,
    request_params={"headers": {"X-Match-Username": "zhangsan"}},
)

使用 auth_request（SKILL 7）会自动从上下文注入 X-Match-Username。

SKILL 7: 认证请求封装

发起 HTTP 请求时自动注入认证信息（App ID/Secret、Token、IP、Trace ID）并上报统计。

同步请求（requests）

from huace_aigc_auth_client import auth_request

response = auth_request("GET", "https://api.example.com/data")
response = auth_request("POST", "https://api.example.com/users",
                        json={"name": "John"})

快捷方法：

from huace_aigc_auth_client import auth_request_get, auth_request_post

r = auth_request_get("https://api.example.com/users", params={"page": 1})
r = auth_request_post("https://api.example.com/users", json={"name": "John"})

会话模式（连接池复用）：

from huace_aigc_auth_client import AuthSession

session = AuthSession()
response = session.get("https://api.example.com/data")

异步请求（httpx）

需安装 httpx：pip install httpx

from huace_aigc_auth_client import AsyncAuthClient

async def example():
    async with AsyncAuthClient(timeout=10.0) as c:
        r = await c.post("https://api.example.com/users", json={"name": "John"})
        r = await c.get("https://api.example.com/users", params={"page": 1})

快捷函数：

from huace_aigc_auth_client import async_auth_httpx_request

r = await async_auth_httpx_request("GET", "https://api.example.com/data")

自动注入的请求头

SKILL 8: 用户上下文

在任意代码位置（Service、Utility）获取当前请求的用户信息，无需层层传参。

前提：必须先注册 AuthMiddleware（SKILL 1）。

获取当前用户

from huace_aigc_auth_client import get_current_user, get_current_user_id, get_current_username, is_current_user_admin

def some_business_logic():
    user = get_current_user()
    if user:
        print(f"用户: {get_current_username()} (ID: {get_current_user_id()})")

    if is_current_user_admin():
        print("管理员操作")

请求上下文

from huace_aigc_auth_client import (
    set_request_context, get_client_ip, get_trace_id, get_request_token,
)

set_request_context(
    ip_address="192.168.1.1",
    user_agent="Mozilla/5.0...",
    trace_id="trace-123",
    token="user-token",
)

ip = get_client_ip()
trace = get_trace_id()
token = get_request_token()

在装饰器中使用

from functools import wraps
from huace_aigc_auth_client import is_current_user_admin

def require_admin(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        if not is_current_user_admin():
            raise HTTPException(403, "需要管理员权限")
        return await func(*args, **kwargs)
    return wrapper

可用函数

线程安全 + 异步安全（threading.local + contextvars.ContextVar）。

SKILL 9: 旧系统用户同步

已有用户表的旧系统低成本接入，无需修改历史代码和表结构。

原理

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   aigc-auth     │────>│   SDK 同步层     │────>│   旧系统         │
│   (鉴权中心)     │     │   (字段映射)     │     │   (用户表)       │
└─────────────────┘     └─────────────────┘     └─────────────────┘
        |                       |                       |
        |   1. 初始化同步        |                       |
        |<──────────────────────────────────────────────|
        |   (批量同步旧用户到 auth)                       |
        |                       |                       |
        |   2. 增量同步          |                       |
        |──────────────────────>|──────────────────────>|
        |   (auth 新用户自动同步到旧系统)                  |
        |                       |                       |
        |   3. Webhook 推送      |                       |
        |──────────────────────────────────────────────>|
        |   (用户变更主动通知)                            |

字段映射

from huace_aigc_auth_client import FieldMapping, create_sync_config, PasswordMode

field_mappings = [
    FieldMapping(auth_field="username", legacy_field="username", required=True),
    FieldMapping(auth_field="email", legacy_field="email"),
    FieldMapping(auth_field="nickname", legacy_field="nickname"),
    # 状态转换
    FieldMapping(
        auth_field="status", legacy_field="is_active",
        transform_to_legacy=lambda s: s == "active",
        transform_to_auth=lambda a: "active" if a else "disabled",
    ),
    # 角色转换
    FieldMapping(
        auth_field="roles", legacy_field="role",
        transform_to_legacy=lambda r: r[0] if r else "viewer",
        transform_to_auth=lambda r: [r] if r else ["viewer"],
    ),
]

sync_config = create_sync_config(
    field_mappings=field_mappings,
    password_mode=PasswordMode.UNIFIED,
    unified_password="Abc@123456",
    webhook_url="https://your-domain.com/api/v1/webhook/auth",
)

实现 LegacySystemAdapter

from huace_aigc_auth_client import LegacySystemAdapter, LegacyUserData
from typing import Optional, List, Any, Dict

class MyLegacyAdapter(LegacySystemAdapter):

    def __init__(self, db, sync_config, auth_client=None):
        super().__init__(sync_config, auth_client)
        self.db = db

    async def get_user_by_username_async(self, username: str) -> Optional[LegacyUserData]:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if not user:
            return None
        return LegacyUserData({"id": user.id, "username": user.username, "email": user.email})

    async def _create_user_async(self, user_data: Dict[str, Any]) -> Optional[Any]:
        user = User(username=user_data["username"], email=user_data.get("email"))
        self.db.add(user)
        await self.db.commit()
        return user.id

    async def _update_user_async(self, username: str, user_data: Dict[str, Any]) -> bool:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if user:
            for k, v in user_data.items():
                setattr(user, k, v)
            await self.db.commit()
            return True
        return False

    async def _delete_user_async(self, username: str) -> bool:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if user:
            await self.db.delete(user)
            await self.db.commit()
            return True
        return False

    async def get_all_users_async(self) -> List[LegacyUserData]:
        result = await self.db.execute(select(User))
        return [LegacyUserData({"username": u.username}) for u in result.scalars().all()]

集成到登录流程

from huace_aigc_auth_client import UserSyncService

adapter = MyLegacyAdapter(db, sync_config, client)
sync_service = UserSyncService(client, adapter)

# 在中间件中同步
@app.middleware("http")
async def auth_and_sync(request, call_next):
    response = await auth_mw.fastapi_middleware(request, call_next)
    if hasattr(request.state, "user_info"):
        await sync_service.sync_on_login_async(request.state.user_info)
    return response

Webhook 接收

使用 SKILL 5 注册 webhook 路由，handler 中调用 adapter：

async def handle_webhook(data: dict) -> dict:
    return await adapter.handle_webhook(data.get("event"), data.get("data", {}))

register_webhook_router(router, handler=handle_webhook)

初始化同步（一次性脚本）

async def init_sync():
    adapter = MyLegacyAdapter(db, sync_config, client)
    users = await adapter.get_all_users_async()
    for user in users:
        auth_data = adapter.transform_legacy_to_auth(user)
        password, is_hashed = adapter.get_password_for_sync(user)
        if is_hashed:
            auth_data["password_hashed"] = password
        else:
            auth_data["password"] = password
        client.sync_user_to_auth(auth_data)

asyncio.run(init_sync())

密码处理策略

# 推荐方式：直接同步 bcrypt 哈希
sync_config = create_sync_config(
    password_mode=PasswordMode.CUSTOM_MAPPING,
    password_mapper=lambda u: u.get("hashed_password"),
    password_is_hashed=True,
)

SKILL 10: 飞书机器人点对点通知

使用鉴权中心全局维护的飞书机器人，向已同步飞书用户标识的账号发送点对点消息。

前置条件

• 接入应用通过 AigcAuthClient 的 app_id、app_secret 完成 SDK 鉴权。
• 超级管理员已在鉴权中心“飞书机器人”页面维护并启用机器人。
• 接收用户已通过飞书登录同步至少一种飞书用户标识。
• 飞书应用已启用机器人能力并发布版本；目标用户已加入应用的可用范围。
• 机器人应用已开通 im:message:send_as_bot 权限（或飞书文档允许的等效消息权限）；使用 feishu_user_id 时还需 contact:user.employee_id:readonly。

发送文本消息

from huace_aigc_auth_client import AigcAuthClient

client = AigcAuthClient(
    app_id="your-auth-app-id",
    app_secret="your-auth-app-secret",
    base_url="https://auth.example.com/api/v1/sdk",
)

# recipient_type 包含："auth_user_id","username","feishu_user_id","feishu_open_id","feishu_union_id"
result = client.send_feishu_message(
    bot_app_id="cli_xxxxxxxxxxxx",
    recipient_type="auth_user_id",
    recipient=123,
    message_type="text",
    content={"text": "制作任务已完成"},
)

print(result["message_id"], result["status"])

机器人 app_secret 由鉴权中心保存，调用方只传机器人 bot_app_id，不应在业务服务中保存机器人密钥。

收件人类型

使用 auth_user_id 或 username 时，鉴权中心会校验用户是否属于当前接入应用、是否处于启用状态，并按 feishu_user_id -> feishu_open_id -> feishu_union_id 的顺序解析。

消息类型

content 必须是非空字典。成功响应包含 message_id、status="sent"，以及飞书返回的可选 root_id、parent_id、create_time。

常见发送失败

鉴权中心会在错误消息中保留飞书上游错误码，便于区分权限问题和应用范围问题：

异常处理

from huace_aigc_auth_client import AigcAuthError

try:
    user = client.get_user_info(token)
except AigcAuthError as e:
    print(f"错误码: {e.code}, 信息: {e.message}")
    print(f"诊断数据: {e.data}")

环境变量完整参考

鉴权（必填）

服务注册（设置 NAME 即启用）

旧系统同步

导出清单

from huace_aigc_auth_client import (
    # 核心类
    AigcAuthClient,
    AuthMiddleware,
    UserInfo,
    TokenVerifyResult,
    AccessKeyVerifyResult,
    AigcAuthError,
    ServiceRegistrationResult,

    # FastAPI 鉴权
    require_auth,
    create_fastapi_auth_dependency,

    # Webhook 接收
    register_webhook_router,
    verify_webhook_signature,

    # API 统计
    init_api_stats_collector,
    get_api_stats_collector,
    stop_api_stats_collector,
    collect_api_stat,

    # 认证请求封装
    auth_request,
    auth_request_get,
    auth_request_post,
    AuthSession,
    async_auth_httpx_request,
    async_auth_httpx_request_get,
    async_auth_httpx_request_post,
    AsyncAuthClient,

    # 用户上下文
    set_current_user,
    get_current_user,
    get_current_user_id,
    get_current_username,
    get_current_app_id,
    get_current_app_code,
    is_current_user_admin,
    clear_current_user,
    set_request_context,
    add_to_request_context,
    get_request_context,
    get_client_ip,
    get_user_agent,
    get_request_id,
    get_trace_id,
    get_request_token,
    get_request_app_id,
    get_request_app_secret,
    clear_request_context,

    # 旧系统接入
    LegacySystemAdapter,
    LegacyUserData,
    SyncConfig,
    SyncDirection,
    PasswordMode,
    FieldMapping,
    UserSyncService,
    WebhookSender,
    SyncResult,
    create_sync_config,
    create_default_field_mappings,

    # Flask Webhook
    create_flask_webhook_blueprint,
    register_flask_webhook_routes,

    # 工具
    setLogger,
)

API 变更日志

未发布 (2026-08-11)

• 新增 send_feishu_message()，支持使用全局机器人按鉴权用户 ID、用户名或飞书用户标识发送 text、post、interactive 点对点消息（详见 SKILL 10）
• UserInfo 和用户查询结果新增 feishu_user_id、feishu_open_id、feishu_union_id、feishu_tenant_key
• 机器人密钥由鉴权中心管理，SDK 调用方只需提供接入应用凭据和机器人 bot_app_id

v1.1.54 (2026-06-03)

• SDK Client 拆分为 client/ 包目录（Mixin 模式），API 不变
• 新增 list_groups() 获取用户组清单
• UserInfo 新增 groups 字段
• 用户信息接口返回 groups（含 id/name/code）

v1.1.52 (2026-06-02)

• 新增服务注册中心（详见 SKILL 4）
• 环境变量自动注册 + 心跳 + 配置 Pull
• get_service_config() 从心跳缓存读取配置

v1.1.50 (2026-06-01)

• 新增 Webhook 接收路由注册（详见 SKILL 5）
• register_webhook_router() / verify_webhook_signature()

v1.1.49 (2026-05-25)

• 新增 list_users() 分页搜索（详见 SKILL 3）

v1.1.47 (2026-05-18)

• collect_api_stat() 新增 response_headers 参数（详见 SKILL 6）
• 动态计费规则 X-Match-* 响应头
• X-Match-Username 用户标识

v1.1.42 (2026-04-28)

• 新增 batch_get_users() / batch_get_users_by_ids()（详见 SKILL 3）

v1.1.0 (2026-01-13)

• 支持直接同步已加密密码（详见 SKILL 9 密码策略）
• _request 支持自定义 headers

许可证

MIT License