coze-workload-identity

Coze Workload Identity SDK

Python SDK，用于 Coze 工作负载身份认证（OAuth2.0 Token Exchange 流程）。

功能特性

• OAuth2.0 双步 Token 交换流程
• 进程级 Token 缓存，自动处理过期（提前 1 分钟刷新）
• 泳道（Lane）支持：BOE、PPE、自定义环境
• 集成凭证（Integration Credential）获取
• 项目环境变量获取
• HTTPS 代理 + 自定义 CA 证书支持
• 线程安全

安装

pip install coze-workload-identity

快速开始

1. 配置环境变量

环境变量	说明	是否必须
COZE_WORKLOAD_IDENTITY_CLIENT_ID	客户端 ID	✅
COZE_WORKLOAD_IDENTITY_CLIENT_SECRET	客户端密钥	✅
COZE_WORKLOAD_IDENTITY_TOKEN_ENDPOINT	ID Token 获取端点	✅
COZE_WORKLOAD_ACCESS_TOKEN_ENDPOINT	Access Token 交换端点	✅
COZE_OUTBOUND_AUTH_ENDPOINT	集成凭证 / 环境变量端点	仅特定接口需要
COZE_SERVER_ENV	泳道环境（默认 NONE）	❌

2. 获取 Access Token

from coze_workload_identity import Client

with Client() as client:
    token = client.get_access_token()
    print(f"Access Token: {token}")

3. 使用环境变量常量（推荐）

import os
from coze_workload_identity.env_keys import (
    COZE_WORKLOAD_IDENTITY_CLIENT_ID,
    COZE_WORKLOAD_IDENTITY_CLIENT_SECRET,
    COZE_WORKLOAD_IDENTITY_TOKEN_ENDPOINT,
    COZE_WORKLOAD_ACCESS_TOKEN_ENDPOINT,
)

os.environ[COZE_WORKLOAD_IDENTITY_CLIENT_ID] = "your_client_id"
os.environ[COZE_WORKLOAD_IDENTITY_CLIENT_SECRET] = "your_client_secret"
os.environ[COZE_WORKLOAD_IDENTITY_TOKEN_ENDPOINT] = "https://auth.example.com/token"
os.environ[COZE_WORKLOAD_ACCESS_TOKEN_ENDPOINT] = "https://auth.example.com/access-token"

API 说明

Client(timeout=(5, 30))

主入口类，支持上下文管理器（with 语句）。

• timeout：HTTP 请求超时，格式为 (连接超时秒数, 读取超时秒数)，默认 (5, 30)。

get_access_token() -> str

获取 Access Token（优先从缓存返回）。

内部流程：

• 用 client_credentials grant 请求 ID Token
• 用 ID Token 通过 token-exchange grant 换取 Access Token
• 缓存 Access Token（到期前 1 分钟自动过期）

get_integration_credential(integration_name: str) -> str

获取指定集成的凭证字符串。需要配置 COZE_OUTBOUND_AUTH_ENDPOINT。

get_project_env_vars() -> ProjectEnvVars

获取项目环境变量列表。需要配置 COZE_OUTBOUND_AUTH_ENDPOINT。

env_vars = client.get_project_env_vars()

# 迭代
for var in env_vars:
    print(f"{var.key} = {var.value}")

# 按 key 读取
value = env_vars.get("MY_KEY", default="")
value = env_vars["MY_KEY"]          # key 不存在时抛出 KeyError
exists = "MY_KEY" in env_vars

泳道（Lane）配置

通过 COZE_SERVER_ENV 控制，影响请求头注入：

值	注入的请求头
NONE（默认）	无
boe_<name>	x-tt-env: boe_<name>
ppe_<name>	x-tt-env: ppe_<name> + x-use-ppe: 1
其他值	x-tt-env: <值>

代理配置

通过以下环境变量配置出口代理：

环境变量	说明
COZE_OUTBOUND_AUTH_PROXY	HTTPS 代理地址
identity_ticket	代理认证票据（Coze Space 沙箱使用，作为 Basic Auth 密码）
COZE_OUTBOUND_AUTH_PROXY_CA	CA 证书内容（优先级高于路径）
COZE_OUTBOUND_AUTH_PROXY_CA_PATH	CA 证书文件路径

使用代理请求时，用 coze_workload_identity.requests 替代标准 requests：

from coze_workload_identity import requests

response = requests.get("https://api.example.com/data")
response = requests.post("https://api.example.com/data", json={"key": "value"})

ConfiguredSession 会自动注入代理和 CA 配置，默认超时同为 (5, 30)。

异常处理

from coze_workload_identity import (
    Client,
    WorkloadIdentityError,
    ConfigurationError,
    TokenRetrievalError,
    TokenExchangeError,
)

try:
    with Client() as client:
        token = client.get_access_token()
except ConfigurationError as e:
    print(f"配置缺失: {e}")
except TokenRetrievalError as e:
    print(f"ID Token 获取失败: {e}")
except TokenExchangeError as e:
    print(f"Token 交换失败: {e}")
except WorkloadIdentityError as e:
    print(f"SDK 错误: {e}")

异常层级：

WorkloadIdentityError
├── ConfigurationError    # 必要环境变量缺失或无效
├── TokenRetrievalError   # ID Token 请求失败
└── TokenExchangeError    # Access Token 交换失败

多线程

SDK 的 Token 缓存是进程级单例，所有 Client 实例共享同一缓存，线程安全：

import threading
from coze_workload_identity import Client

def worker():
    with Client() as client:
        token = client.get_access_token()  # 自动命中缓存，无重复请求

threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

开发

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试（含覆盖率）
python run_tests.py

# 运行测试（不含覆盖率）
python run_tests.py --no-coverage

# 运行单个测试文件
python -m pytest tests/test_client.py -v

# 代码格式化
black coze_workload_identity/

# 类型检查
mypy coze_workload_identity/

# Lint
flake8 coze_workload_identity/