Ark Assets SDK

火山方舟（Ark）素材库控制面 Python SDK —— 单文件，零业务依赖，封装素材库全部管理操作。

底层基于 volcengine-python-sdk 的签名与通用调用通道（UniversalApi），上层提供简洁的 Pythonic API。

环境要求

项目	要求
Python	>= 3.9
依赖	`volcengine-python-sdk >= 5.0.40`
凭证	火山引擎 AK / SK（Access Key / Secret Key）
网络	能访问目标 API 端点（`art-api.yuyuflow.com`）

安装

方式一：uv（推荐）

bash

# 初始化项目（如果还没有）
uv init my-project && cd my-project

# 安装 SDK 依赖
uv add volcengine-python-sdk

# 将 ark_assets.py 复制到项目目录
cp ark_assets.py .

方式二：pip

bash

pip install volcengine-python-sdk

然后将 ark_assets.py 复制到你的项目目录中即可。

本 SDK 为单文件，无需额外安装，直接 import 即可。

配置

SDK 支持两种配置方式：代码传参 和 环境变量。

方式一：代码传参（推荐生产环境）

python

from ark_assets import ArkAssetsConfig, ArkAssetsClient

config = ArkAssetsConfig(
    ak="YOUR_AK",
    sk="YOUR_SK",
    # 以下为默认值，可省略
    region="cn-beijing",
    host="art-api.yuyuflow.com",
    scheme="https",
    verify_ssl=True,
    project_name="default",
)

client = ArkAssetsClient(config)

方式二：环境变量

bash

export VOLC_AK="YOUR_AK"
export VOLC_SK="YOUR_SK"
# 以下为默认值，可省略
export VOLC_HOST="art-api.yuyuflow.com"
export VOLC_SCHEME="https"
export VOLC_PROJECT="default"

python

from ark_assets import ArkAssetsConfig, ArkAssetsClient

# 自动从环境变量读取
client = ArkAssetsClient()  # 等价于 ArkAssetsClient(ArkAssetsConfig())

配置项说明

参数	环境变量	默认值	说明
`ak`	`VOLC_AK`	—	Access Key（必填）
`sk`	`VOLC_SK`	—	Secret Key（必填）
`region`	`VOLC_REGION`	`cn-beijing`	区域
`host`	`VOLC_HOST`	`art-api.yuyuflow.com`	API 端点域名
`scheme`	`VOLC_SCHEME`	`https`	`http` 或 `https`
`verify_ssl`	`VOLC_VERIFY_SSL`	`true`	是否校验 SSL 证书
`project_name`	`VOLC_PROJECT`	`default`	资源所属项目（大小写敏感）

快速开始

30 秒上手：创建 AIGC 素材组 → 上传素材 → 轮询就绪

python

from ark_assets import ArkAssetsConfig, ArkAssetsClient

client = ArkAssetsClient(ArkAssetsConfig(
    ak="YOUR_AK",
    sk="YOUR_SK",
))

# 1. 创建 AIGC 素材组
group = client.create_asset_group(name="my-portrait", group_type="AIGC")
group_id = group["Id"]
print(f"素材组 ID: {group_id}")

# 2. 上传素材并等待处理完成（内置轮询）
asset = client.upload_and_wait(
    group_id=group_id,
    url="https://example.com/photo.jpg",
    asset_type="Image",
    name="my-photo",
    interval=5,      # 轮询间隔（秒）
    timeout=300,     # 超时（秒）
)
print(f"素材 ID: {asset['Id']}, 状态: {asset['Status']}")

# 3. 使用完毕后清理
client.delete_asset(asset["Id"])
client.delete_asset_group(group_id)

API 参考

所有方法返回原始 dict（即服务端 Result 对象），不做额外封装，保留全部字段。

真人认证

真人认证素材库（LivenessFace）不是手动创建的，而是通过真人认证流程自动创建。

`create_liveness_session(callback_url) -> dict`

拉起 H5 真人认证会话。

参数	类型	必填	说明
`callback_url`	`str`	是	认证结束后跳转的 URL

返回值：

python

{
    "BytedToken": "20260710...",   # 认证凭证，有效期 30 分钟，仅限一次
    "H5Link": "https://...",        # 终端用户需打开此链接完成认证
    "CallbackURL": "https://..."    # 回显
}

python

session = client.create_liveness_session("https://example.com/callback")
print(session["BytedToken"])
print(session["H5Link"])

`get_liveness_result(byted_token) -> dict`

查询真人认证结果。认证成功后系统自动创建 LivenessFace 素材组。

参数	类型	必填	说明
`byted_token`	`str`	是	来自 `create_liveness_session`

返回值：

python

{"GroupId": "mag_xxxxx"}  # 认证成功时返回；未完成时服务端返回错误

python

result = client.get_liveness_result("20260710...")
print(result["GroupId"])

`run_liveness_flow(callback_url, poll_interval, timeout, on_session) -> dict`

高级方法：拉起认证 + 自动轮询直到拿到 GroupId。

参数	类型	默认值	说明
`callback_url`	`str`	—	认证跳转 URL
`poll_interval`	`float`	`15.0`	轮询间隔（秒）
`timeout`	`float`	`1800.0`	超时（秒）
`on_session`	`callable`	`None`	回调函数 `fn(byted_token, h5_link)`，拿到会话信息后调用

python

def on_session(byted_token, h5_link):
    print(f"请打开此链接完成认证: {h5_link}")
    print(f"凭证: {byted_token}")

result = client.run_liveness_flow(
    callback_url="https://example.com/callback",
    poll_interval=15,
    timeout=1800,
    on_session=on_session,
)
print(f"认证成功! GroupId: {result['GroupId']}")

素材组管理

`create_asset_group(name, group_type, description) -> dict`

参数	类型	必填	默认值	说明
`name`	`str`	是	—	组名称，上限 64 字符
`group_type`	`str`	否	`"AIGC"`	`AIGC`（虚拟人像）或 `LivenessFace`（通常由真人认证自动创建）
`description`	`str`	否	`""`	组描述，上限 300 字符

python

resp = client.create_asset_group(name="my-group", group_type="AIGC", description="test")
# {"Id": "mag_xxxxx"}

`list_asset_groups(page_number, page_size, group_type, group_ids, name, sort_by, sort_order) -> dict`

参数	类型	必填	默认值	说明
`page_number`	`int`	否	`1`	页码，从 1 开始
`page_size`	`int`	否	`20`	每页数量，上限 100
`group_type`	`str`	否	`None`	`AIGC` / `LivenessFace`
`group_ids`	`list[str]`	否	`None`	按 ID 列表过滤
`name`	`str`	否	`None`	名称模糊搜索
`sort_by`	`str`	否	`"CreateTime"`	`CreateTime` / `UpdateTime`
`sort_order`	`str`	否	`"Desc"`	`Desc` / `Asc`

python

resp = client.list_asset_groups(group_type="AIGC", page_size=10)
# {"TotalCount": 1, "Items": [{...}], "PageNumber": 1, "PageSize": 10}

`get_asset_group(group_id) -> dict`

python

resp = client.get_asset_group("mag_xxxxx")
# {"Id": "mag_xxx", "Name": "my-group", "GroupType": "AIGC", "Description": "...", ...}

`update_asset_group(group_id, name, description) -> dict`

参数	类型	必填	说明
`group_id`	`str`	是	组 ID
`name`	`str`	否	新名称（传 `None` 表示不修改）
`description`	`str`	否	新描述（传 `None` 表示不修改）

python

resp = client.update_asset_group("mag_xxxxx", name="new-name", description="updated")
# {"Id": "mag_xxxxx"}

`delete_asset_group(group_id) -> dict`

python

resp = client.delete_asset_group("mag_xxxxx")
# {"Id": "mag_xxxxx"}

素材管理

`create_asset(group_id, url, asset_type, name) -> dict`

参数	类型	必填	默认值	说明
`group_id`	`str`	是	—	所属素材组 ID
`url`	`str`	是	—	素材公网可访问 URL（不支持 Base64）
`asset_type`	`str`	否	`"Image"`	`Image` / `Video` / `Audio`
`name`	`str`	否	`""`	名称，仅用于 ListAssets 模糊搜索

素材格式要求：

类型	格式	大小限制	其他
Image	jpeg/png/webp/bmp/tiff/gif/heic/heif	< 30 MB	宽高比 (0.4, 2.5)，宽高 (300, 6000) px
Video	mp4/mov	< 50 MB	时长 2-15s，480p/720p/1080p
Audio	wav/mp3	< 15 MB	时长 2-15s

python

resp = client.create_asset(
    group_id="mag_xxxxx",
    url="https://example.com/photo.jpg",
    asset_type="Image",
    name="my-photo",
)
# {"Id": "mas_xxxxx", "Status": "Processing"}

`get_asset(asset_id) -> dict`

python

resp = client.get_asset("mas_xxxxx")
# {"Id": "mas_xxx", "Name": "my-photo", "Status": "Active", "URL": "...", ...}

Status 取值： Active（就绪） / Processing（处理中） / Failed（失败）

`list_assets(page_number, page_size, group_ids, group_type, statuses, name, sort_by, sort_order) -> dict`

参数	类型	必填	默认值	说明
`page_number`	`int`	否	`1`	页码
`page_size`	`int`	否	`20`	每页数量，上限 100
`group_ids`	`list[str]`	否	`None`	按组 ID 过滤
`group_type`	`str`	否	`None`	`AIGC` / `LivenessFace`
`statuses`	`list[str]`	否	`None`	`Active` / `Processing` / `Failed`
`name`	`str`	否	`None`	名称模糊搜索
`sort_by`	`str`	否	`"CreateTime"`	排序字段
`sort_order`	`str`	否	`"Desc"`	排序方向

python

resp = client.list_assets(group_ids=["mag_xxx"], statuses=["Active"], page_size=10)
# {"Items": [{...}], "TotalCount": 1, "PageNumber": 1, "PageSize": 10}

`update_asset(asset_id, name) -> dict`

python

resp = client.update_asset("mas_xxxxx", name="new-name")
# {"Id": "mas_xxxxx"}

`delete_asset(asset_id) -> dict`

python

resp = client.delete_asset("mas_xxxxx")
# {"Id": "mas_xxxxx"}

轮询与组合方法

`wait_asset_active(asset_id, interval, timeout) -> dict`

轮询 get_asset 直到 Status = Active。Failed 时抛出异常，超时抛 TimeoutError。

参数	类型	默认值	说明
`asset_id`	`str`	—	素材 ID
`interval`	`float`	`5.0`	轮询间隔（秒）
`timeout`	`float`	`300.0`	超时（秒）

python

asset = client.wait_asset_active("mas_xxxxx", interval=5, timeout=300)

`upload_and_wait(group_id, url, asset_type, name, interval, timeout) -> dict`

上传素材 + 轮询就绪，一步到位。

python

asset = client.upload_and_wait(
    group_id="mag_xxx",
    url="https://example.com/photo.jpg",
    asset_type="Image",
    name="my-photo",
    interval=5,
    timeout=300,
)

异步用法

SDK 提供 AsyncArkAssetsClient，通过 asyncio.to_thread 代理同步方法，支持 asyncio.gather 并发调用。

python

import asyncio
from ark_assets import ArkAssetsConfig, AsyncArkAssetsClient

async_client = AsyncArkAssetsClient(ArkAssetsConfig(
    ak="YOUR_AK",
    sk="YOUR_SK",
))

async def main():
    # 并发创建两个素材组
    g1, g2 = await asyncio.gather(
        async_client.create_asset_group(name="group-1", group_type="AIGC"),
        async_client.create_asset_group(name="group-2", group_type="AIGC"),
    )
    print(g1["Id"], g2["Id"])

    # 并发上传多个素材
    assets = await asyncio.gather(
        async_client.upload_and_wait(g1["Id"], "https://example.com/a.jpg", "Image", "a"),
        async_client.upload_and_wait(g2["Id"], "https://example.com/b.jpg", "Image", "b"),
    )
    for a in assets:
        print(a["Id"], a["Status"])

    # 清理
    await asyncio.gather(
        async_client.delete_asset(assets[0]["Id"]),
        async_client.delete_asset(assets[1]["Id"]),
        async_client.delete_asset_group(g1["Id"]),
        async_client.delete_asset_group(g2["Id"]),
    )

asyncio.run(main())

AsyncArkAssetsClient 的方法签名与同步版完全一致，仅多一个 async/await。

错误处理

所有 API 错误统一抛出 ArkAssetsError，包含 action、status、body 三个属性。

python

from ark_assets import ArkAssetsClient, ArkAssetsError

client = ArkAssetsClient(ArkAssetsConfig(ak="...", sk="..."))

try:
    resp = client.create_asset(group_id="mag_xxx", url="https://example.com/bad.jpg", asset_type="Image")
except ArkAssetsError as e:
    print(f"操作: {e.action}")
    print(f"HTTP状态: {e.status}")
    print(f"响应体: {e.body}")
    # 响应体中包含 Error.Code 和 Error.Message

常见错误码：

Error.Code	说明
`FaceMismatch`	真人脸一致性校验失败（LivenessFace 组上传非同一人的素材）
`DownloadFailed`	无法下载素材 URL
`ContentRestricted`	内容安全审核未通过
`InvalidParameter`	参数错误
`InternalError`	服务内部异常，建议重试

完整端到端示例

场景一：AIGC 虚拟人像完整流程

python

from ark_assets import ArkAssetsConfig, ArkAssetsClient

client = ArkAssetsClient(ArkAssetsConfig(
    ak="YOUR_AK",
    sk="YOUR_SK",
))

# 1. 建组
group = client.create_asset_group(name="my-aigc", group_type="AIGC", description="demo")
group_id = group["Id"]

# 2. 上传 + 等待就绪
asset = client.upload_and_wait(
    group_id=group_id,
    url="https://example.com/portrait.jpg",
    asset_type="Image",
    name="portrait-1",
)
asset_id = asset["Id"]

# 3. 改名
client.update_asset(asset_id, name="renamed")

# 4. 查列表
resp = client.list_assets(group_ids=[group_id], statuses=["Active"])
print(f"素材总数: {resp['TotalCount']}")

# 5. 查单个
detail = client.get_asset(asset_id)
print(f"素材 URL: {detail['URL']}")

# 6. 清理
client.delete_asset(asset_id)
client.delete_asset_group(group_id)

场景二：真人认证素材库

python

from ark_assets import ArkAssetsConfig, ArkAssetsClient

client = ArkAssetsClient(ArkAssetsConfig(
    ak="YOUR_AK",
    sk="YOUR_SK",
))

def on_session(byted_token, h5_link):
    print(f"请打开链接完成认证: {h5_link}")

# 拉起认证 + 轮询，认证完成后自动拿到 GroupId
result = client.run_liveness_flow(
    callback_url="https://example.com/callback",
    poll_interval=15,
    on_session=on_session,
)
group_id = result["GroupId"]

# 向真人素材组上传素材（系统会做人脸一致性比对）
asset = client.upload_and_wait(
    group_id=group_id,
    url="https://example.com/my-photo.jpg",
    asset_type="Image",
    name="my-face",
)

场景三：异步并发上传

python

import asyncio
from ark_assets import ArkAssetsConfig, AsyncArkAssetsClient

async def main():
    client = AsyncArkAssetsClient(ArkAssetsConfig(ak="...", sk="..."))

    # 并发上传 3 张图片到同一素材组
    group = await client.create_asset_group(name="batch", group_type="AIGC")
    group_id = group["Id"]

    urls = [
        "https://example.com/1.jpg",
        "https://example.com/2.jpg",
        "https://example.com/3.jpg",
    ]
    assets = await asyncio.gather(*[
        client.upload_and_wait(group_id, url, "Image", f"img-{i}")
        for i, url in enumerate(urls)
    ])
    for a in assets:
        print(a["Id"], a["Status"])

asyncio.run(main())

CLI Demo

SDK 自带 CLI 演示入口，可直接运行：

AIGC 完整流程演示

bash

export VOLC_AK="YOUR_AK"
export VOLC_SK="YOUR_SK"

python ark_assets.py --mode aigc --asset-url "https://example.com/photo.jpg"

输出：

text

============================================================
 AIGC 素材组完整流程
============================================================
[1] CreateAssetGroup  -> mag_xxxxx
[2] UploadAndWait     -> mas_xxxxx  Status=Active
[3] UpdateAsset       -> {'Id': 'mas_xxxxx'}
[4] ListAssets        -> TotalCount=1
[5] DeleteAsset       -> {'Id': 'mas_xxxxx'}
[6] DeleteAssetGroup  -> {'Id': 'mag_xxxxx'}

真人认证流程演示

bash

python ark_assets.py --mode liveness --callback-url "https://example.com/callback"

Async 并发演示

bash

python ark_assets.py --mode async

FAQ

Q: 素材上传后多久能用？

create_asset 是异步接口，返回 Status=Processing。需轮询 get_asset 或使用 wait_asset_active 直到 Status=Active。图片通常几秒，视频可能更久。

Q: LivenessFace 素材组可以手动创建吗？

不可以。LivenessFace 素材组只能通过真人认证（CreateVisualValidateSession → H5 认证 → GetVisualValidateResult）自动创建。create_asset_group 仅支持 AIGC 类型。

Q: 上传素材时提示 FaceMismatch 怎么办？

这表示上传的人脸与真人认证时采集的基准人脸不一致。LivenessFace 素材组要求所有素材为同一人。请确认素材图片正确。

Q: 素材 URL 有什么要求？

必须是公网可访问的 URL（不支持 Base64）
火山引擎 OBS 签名 URL 注意有效期，过期后上传会报 DownloadFailed

Q: 签名会因自定义 host 失败吗？

不会。SDK 的签名密钥派生只依赖 SK/date/region/service，与 host 无关。Host 仅作为被签名的 Header 之一，客户端发什么 host 就签什么 host。

限流说明

接口	账号维度限流
CreateVisualValidateSession	3 QPS
GetVisualValidateResult	3 QPS
CreateAsset	联系销售
GetAsset	100 QPS
ListAssets / ListAssetGroups	10 QPS
GetAssetGroup / UpdateAssetGroup / UpdateAsset	10 QPS
DeleteAsset	10 QPS
DeleteAssetGroup	5 QPS

Ark Assets SDK​

目录​

环境要求​

安装​

方式一：uv（推荐）​

方式二：pip​

配置​

方式一：代码传参（推荐生产环境）​

方式二：环境变量​

配置项说明​

快速开始​

30 秒上手：创建 AIGC 素材组 → 上传素材 → 轮询就绪​

API 参考​

真人认证​

create_liveness_session(callback_url) -> dict​

get_liveness_result(byted_token) -> dict​

run_liveness_flow(callback_url, poll_interval, timeout, on_session) -> dict​

素材组管理​

create_asset_group(name, group_type, description) -> dict​

list_asset_groups(page_number, page_size, group_type, group_ids, name, sort_by, sort_order) -> dict​

get_asset_group(group_id) -> dict​

update_asset_group(group_id, name, description) -> dict​

delete_asset_group(group_id) -> dict​

素材管理​

create_asset(group_id, url, asset_type, name) -> dict​

get_asset(asset_id) -> dict​

list_assets(page_number, page_size, group_ids, group_type, statuses, name, sort_by, sort_order) -> dict​

update_asset(asset_id, name) -> dict​

delete_asset(asset_id) -> dict​

轮询与组合方法​

wait_asset_active(asset_id, interval, timeout) -> dict​

upload_and_wait(group_id, url, asset_type, name, interval, timeout) -> dict​

异步用法​

错误处理​

完整端到端示例​

场景一：AIGC 虚拟人像完整流程​

场景二：真人认证素材库​

场景三：异步并发上传​

CLI Demo​

AIGC 完整流程演示​

真人认证流程演示​

Async 并发演示​

FAQ​

Q: 素材上传后多久能用？​

Q: LivenessFace 素材组可以手动创建吗？​

Q: 上传素材时提示 FaceMismatch 怎么办？​

Q: 素材 URL 有什么要求？​

Q: 签名会因自定义 host 失败吗？​

限流说明​

Ark Assets SDK

目录

环境要求

安装

方式一：uv（推荐）

方式二：pip

配置

方式一：代码传参（推荐生产环境）

方式二：环境变量

配置项说明

快速开始

30 秒上手：创建 AIGC 素材组 → 上传素材 → 轮询就绪

API 参考

真人认证

`create_liveness_session(callback_url) -> dict`

`get_liveness_result(byted_token) -> dict`

`run_liveness_flow(callback_url, poll_interval, timeout, on_session) -> dict`

素材组管理

`create_asset_group(name, group_type, description) -> dict`

`list_asset_groups(page_number, page_size, group_type, group_ids, name, sort_by, sort_order) -> dict`

`get_asset_group(group_id) -> dict`

`update_asset_group(group_id, name, description) -> dict`

`delete_asset_group(group_id) -> dict`

素材管理

`create_asset(group_id, url, asset_type, name) -> dict`

`get_asset(asset_id) -> dict`

`list_assets(page_number, page_size, group_ids, group_type, statuses, name, sort_by, sort_order) -> dict`

`update_asset(asset_id, name) -> dict`

`delete_asset(asset_id) -> dict`

轮询与组合方法

`wait_asset_active(asset_id, interval, timeout) -> dict`

`upload_and_wait(group_id, url, asset_type, name, interval, timeout) -> dict`

异步用法

错误处理

完整端到端示例

场景一：AIGC 虚拟人像完整流程

场景二：真人认证素材库

场景三：异步并发上传

CLI Demo

AIGC 完整流程演示

真人认证流程演示

Async 并发演示

FAQ

Q: 素材上传后多久能用？

Q: LivenessFace 素材组可以手动创建吗？

Q: 上传素材时提示 FaceMismatch 怎么办？

Q: 素材 URL 有什么要求？

Q: 签名会因自定义 host 失败吗？

限流说明