@coze/workload-identity

TypeScript SDK for Coze Workload Identity — 基于 OAuth2.0 Token Exchange 的工作负载身份认证。

功能与 coze-workload-identity (Python SDK) 对齐,便于 Node.js 项目以 npm 依赖接入。

功能特性

• OAuth2.0 双步 Token 交换(client_credentials → token-exchange)
• 进程级 Access Token 缓存,自动处理过期(提前 1 分钟刷新)
• Single-flight 并发优化:多个 Promise 并发首次请求时共享同一次上游调用
• 泳道(Lane)支持:NONE / boe_* / ppe_* / 自定义值
• 集成凭证(Integration Credential)获取
• 项目环境变量(Project Env Vars)获取
• HTTPS 代理 + 自定义 CA 证书支持
• 全 TypeScript,自带 .d.ts 类型

环境要求

• Node.js >= 18

安装

npm install @coze/workload-identity
# or
pnpm add @coze/workload-identity
# or
yarn add @coze/workload-identity

快速开始

1. 配置环境变量

| 环境变量 | 必需 | 说明 | |---|---|---| | COZE_WORKLOAD_IDENTITY_CLIENT_ID | ✅ | Workload Identity Client ID | | COZE_WORKLOAD_IDENTITY_CLIENT_SECRET | ✅ | Workload Identity Client Secret | | COZE_WORKLOAD_IDENTITY_TOKEN_ENDPOINT | ✅ | ID Token 端点 | | COZE_WORKLOAD_ACCESS_TOKEN_ENDPOINT | ✅ | Access Token 交换端点 | | COZE_OUTBOUND_AUTH_ENDPOINT | ⚠️ 仅 integration / env API | 出站鉴权服务地址 | | COZE_SERVER_ENV | ❌ | 泳道,默认 NONE | | COZE_OUTBOUND_AUTH_PROXY | ❌ | HTTPS 代理 URL(仅代理工具需要) | | identity_ticket | ❌ | 代理 Basic Auth 密码(仅代理工具需要) | | COZE_OUTBOUND_AUTH_PROXY_CA | ❌ | CA 证书内容(仅代理工具需要) | | COZE_OUTBOUND_AUTH_PROXY_CA_PATH | ❌ | CA 证书文件路径(仅代理工具需要) |

2. 获取 Access Token

import { Client } from '@coze/workload-identity';

const client = new Client();
try {
  const token = await client.getAccessToken();
  console.log(token);
} finally {
  await client.close();
}

3. 使用环境变量常量(避免硬编码)

import { envKeys } from '@coze/workload-identity';

const lane = process.env[envKeys.COZE_SERVER_ENV] ?? envKeys.DEFAULT_COZE_SERVER_ENV;

API

new Client(options?)

interface ClientOptions {
  /** 单次请求超时时间(毫秒),默认 30000。 */
  timeoutMs?: number;
}

构造时会一次性读取必需环境变量;缺失则抛 ConfigurationError,错误消息会列出所有缺失项。

client.getAccessToken(): Promise<string>

执行两步 OAuth2.0 流程:

1. POST COZE_WORKLOAD_IDENTITY_TOKEN_ENDPOINT,grant_type=client_credentials 换取 ID Token;
2. POST COZE_WORKLOAD_ACCESS_TOKEN_ENDPOINT,使用步骤 1 的 Token 作为 subject_token,grant_type=urn:ietf:params:oauth:grant-type:token-exchange 换取 Access Token;
3. 缓存 Access Token(expires_in - 60 秒),返回。

同一进程内的所有 Client 实例共享缓存。

client.getIntegrationCredential(name): Promise<string>

通过 Bearer 认证 POST {COZE_OUTBOUND_AUTH_ENDPOINT}/integration,请求体 {"integration_name": name}。成功时返回 data.credential。若 COZE_OUTBOUND_AUTH_ENDPOINT 未配置,抛 ConfigurationError。

client.getProjectEnvVars(): Promise<ProjectEnvVars>

通过 Bearer 认证 GET {COZE_OUTBOUND_AUTH_ENDPOINT}/env。返回值是一个 ProjectEnvVars 集合,可迭代、可按 key 查询。

const envVars = await client.getProjectEnvVars();

// 迭代
for (const v of envVars) {
  console.log(v.key, v.value);
}

// 按 key 读取(不存在返回 undefined)
const apiKey = envVars.get('API_KEY');

// 按 key 读取并在缺失时抛错 —— 对应 Python 的 env_vars[key]
const required = envVars.getOrThrow('DB_URL');

// 判断是否存在
if (envVars.has('API_KEY')) { /* ... */ }

// 转为普通对象
const record = envVars.toRecord();

Python ↔ TypeScript 对照

| Python | TypeScript | |---|---| | env_vars[key] | envVars.getOrThrow(key) | | env_vars.get(key, default) | envVars.get(key, default) | | key in env_vars | envVars.has(key) | | len(env_vars) | envVars.size | | for v in env_vars: | for (const v of envVars) { ... } |

client.close(): Promise<void>

API 兼容保留,无实际资源释放(与 Python SDK 行为一致)。

泳道(Lane)配置

COZE_SERVER_ENV 的值决定是否注入路由头:

| 值 | x-tt-env | x-use-ppe | |---|---|---| | 未设置 / NONE | — | — | | boe_<name> | boe_<name> | — | | ppe_<name> | ppe_<name> | 1 | | 其他任意值 | 原值 | — |

代理配置

出站 HTTPS 代理是可选的,只有当你需要通过代理访问外部服务时使用。

import { createHttpClient, http, httpsProxy, caBundlePath } from '@coze/workload-identity';

// 方式一:直接用模块级便捷函数 —— 走懒加载默认客户端
const res1 = await http.get('https://api.example.com/health');

// 方式二:显式构建客户端,自行管理生命周期
const client = createHttpClient({ timeoutMs: 10_000 });
try {
  const res2 = await client.post('https://api.example.com/echo', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ hello: 'world' }),
  });
} finally {
  await client.close();
}

// 方式三:底层 helpers(只在你自己封装 HTTP 栈时需要)
const proxyUrl = httpsProxy();      // 带 Basic Auth 的完整代理 URL
const caPath = caBundlePath();      // CA 证书文件路径或 null

CA 证书优先级:

1. COZE_OUTBOUND_AUTH_PROXY_CA:证书内容,SDK 会在系统临时目录写入一个受限权限的文件,进程退出时自动清理;
2. COZE_OUTBOUND_AUTH_PROXY_CA_PATH:证书文件路径,使用前会校验文件存在;
3. 均未设置:caBundlePath() 返回 null,TLS 走系统默认 CA。

异常

Error
  └── WorkloadIdentityError
        ├── ConfigurationError    // 必需环境变量缺失
        ├── TokenRetrievalError   // ID Token 获取失败
        └── TokenExchangeError    // Access Token 交换失败

捕获示例:

import {
  Client,
  ConfigurationError,
  TokenRetrievalError,
  TokenExchangeError,
  WorkloadIdentityError,
} from '@coze/workload-identity';

try {
  const client = new Client();
  await client.getAccessToken();
} catch (e) {
  if (e instanceof ConfigurationError)    { /* 配置错误 */ }
  else if (e instanceof TokenRetrievalError) { /* ID Token 获取失败 */ }
  else if (e instanceof TokenExchangeError)  { /* Access Token 交换失败 */ }
  else if (e instanceof WorkloadIdentityError) { /* 其他 SDK 错误 */ }
  else throw e;
}

并发(Single-flight)

当多个 Promise 并发首次调用 getAccessToken() 时,只会发起一次上游 Token 请求,其余 caller 共享同一 Promise。这是 TS SDK 针对 Node 事件循环模型的增强(Python 的 RLock 只保证串行化,不去重)。

const results = await Promise.all([
  client.getAccessToken(),
  client.getAccessToken(),
  client.getAccessToken(),
]);
// 实际只对外发了一次两步 OAuth 请求

与 Python SDK 的关键差异

| 项 | Python | TypeScript | |---|---|---| | Token 请求 | requests.Session | 原生 fetch / undici.fetch | | 超时 | (connect, read) 二元组 | 单一 timeoutMs(fetch 不支持分离) | | 下标访问 env vars | env_vars[key] | envVars.getOrThrow(key) | | Token 缓存并发 | 锁串行化 | Promise single-flight 去重 | | close() | 关闭 Session | no-op(保留 API) |

开发

cd ts-sdk
npm install
npm run typecheck
npm test
npm run test:coverage
npm run build   # 产出 dist/index.{cjs,js,d.ts}

License

MIT