## 一. 开始使用

调用 BizyAir AI 应用 API 前，请先完成以下准备。

### 1. 注册并充值

登录或注册 BizyAir 账号，并确保余额充足。**AI 应用按运行耗时从余额扣费，优先使用赠送余额，不足时再使用充值余额。**

### 2. 获取您的 API Key

每位用户都拥有 BizyAir 提供的独特 **API Key**。

您可以将光标移动至右上角用户头像处，点击【API 密钥】获取您的专属密钥。

**请妥善保存您的 API Key**，它将参与接下来的每一步操作。

### 3. 获取 AI 应用 ID（web_app_id）

调用 AI 应用需要知道应用的 `web_app_id`。应用详情页 URL 中的数字即为 `web_app_id`。

---

## 二. 调用模式总览

BizyAir 提供 3 种调用模式，通过 HTTP Header 切换，**无需修改请求体**：

| 模式 | 启用方式 | 行为 | 适用场景 |
| :--- | :--- | :--- | :--- |
| 同步阻塞 | 不设置任何特殊 Header | 这次请求一直等到任务结束，响应里直接是结果 | 短任务、简单调试 |
| 异步查询 | `X-Bizyair-Task-Async: enable` | 立即返回 `request_id`，客户端轮询查结果 | 长任务、网络不稳定、无回调地址 |
| WebHook | `X-Bizyair-Task-WebHook-Url: <url>` | 立即返回 `request_id`，任务完成后主动 POST 回调 | 生产环境、高并发、长耗时任务 |

### 如何选择调用模式

| 你的场景 | 推荐模式 | 理由 |
| :--- | :--- | :--- |
| 本地脚本快速验证 | 同步阻塞 | 最简单，一次请求直接拿结果 |
| 短任务集成到后端 | 同步阻塞 | 一次请求等到结果即可 |
| 视频生成 / 3D 渲染（长任务） | WebHook | 提交后马上拿到 `request_id`，结果由回调送达 |
| 客户端网络不稳定 | 异步查询 | 先拿到 `request_id`，再按需轮询 |
| 批量并行上百个任务 | 异步查询 或 WebHook | 全部排队后慢慢回收 |
| 生产环境高可用 | WebHook | 主动推送 + 重试机制 |

---

## 三. 提交请求

### 1. 请求参数说明

| 参数名 | 类型 | 必填 | 说明 |
| :--- | :--- | :--- | :--- |
| `web_app_id` | integer | 是 | AI 应用 ID，从应用详情页 URL 获取 |
| `input_values` | object | 是 | 输入参数键值对，key 格式为 `{node_id}:{node_type}.{field_name}`，如 `"84:CLIPTextEncode.text"` |
| `input_values.19:ResolutionSelector.aspect_ratio` | string | 是 | ⟨bz_enum_json⟩["1:1 (Square)","2:3 (Portrait Photo)","3:2 (Photo)","3:4 (Portrait Standard)","4:3 (Standard)","9:16 (Portrait Widescreen)","16:9 (Widescreen)","21:9 (Ultrawide)"]⟨/bz_enum_json⟩<br/>Aspect ratio |
| `input_values.19:ResolutionSelector.megapixels` | number | 是 | 取值范围：0.1 ~ 16<br/>步进：0.1<br/>Megapixels<br/>小数位数：1 |
| `input_values.19:ResolutionSelector.multiple` | integer | 是 | 取值范围：8 ~ 128<br/>步进：4<br/>Multiple |
| `input_values.28:PrimitiveStringMultiline.value` | string | 是 | Prompt |
| `suppress_preview_output` | boolean | 否 | 是否关闭预览输出。为 true 时不返回 Preview 节点的预览图，默认 `false` |

### 2. 同步阻塞调用

默认模式。这次请求一直等到任务结束，响应里直接是结果，无需额外 Header。

**请求示例**

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://api.bizyair.ai/v1/webapp/task/openapi/create"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
payload = {
    "web_app_id": 58151,
    "suppress_preview_output": False,
    "input_values": {
        "13:ResolutionSelector.aspect_ratio": "1:1 (Square)",
          "471:TextEncodeQwenImage21.prompt": "A cute cat, simple style",
          "471:TextEncodeQwenImage21.negative_prompt": ""
    }
}

response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("status") != "Success":
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print(result)
```

> **注意**：同步模式下，本次请求将阻塞至任务结束，响应体即为任务结果。单次运行耗时可能较长，建议改用异步查询或 WebHook。二者在提交后立即返回 `request_id`，再通过轮询或回调获取结果。客户端提前断开不会取消服务端任务，任务仍可能执行完成并产生费用。

**响应示例**

```json
{
  "type": "API",
  "status": "Success",
  "created_at": "2026-09-23T07:44:34Z",
  "updated_at": "2026-09-23T07:45:19Z",
  "executed_at": "2026-09-23T07:44:34Z",
  "ended_at": "2026-09-23T07:45:19Z",
  "expired_at": "2026-10-08T00:00:00Z",
  "request_id": "2e752d06-3585-4a70-a55e-7c552d1399ef",
  "outputs": [
    {
      "object_url": "https://storage.bizyair.ai/outputs_examples/lXobDnouo90P3XoF1stRJhup4mgx4n26.png",
      "output_ext": ".png",
      "cost_time": 42789,
      "error_type": "NOT_ERROR"
    }
  ],
  "cost_times": {
    "inference_cost_time": 42930,
    "running_cost_time": 44873,
    "total_cost_time": 44880,
    "real_cpu_cost_time": 430,
    "real_total_cost_time": 430
  }
}
```

**响应字段说明**：

| 参数名 | 类型 | 说明 |
| :--- | :--- | :--- |
| type | string | 任务类型，如 `"API"`。 |
| status | string | 任务状态：`Queuing` 排队中 / `Preparing` 准备中 / `Running` 运行中 / `Success` 成功 / `Failed` 失败 / `Canceled` 已取消。 |
| request_id | string | 请求 ID，可用于后续查询。 |
| created_at | string | 创建时间（ISO 8601，UTC）。 |
| updated_at | string | 最后更新时间（ISO 8601，UTC）。 |
| executed_at | string | 开始执行时间（ISO 8601，UTC）。 |
| ended_at | string | 结束时间（ISO 8601，UTC）。 |
| expired_at | string | 结果文件过期时间（ISO 8601，UTC），请及时下载。 |
| outputs | array[object (Output)] | 结果列表。 |
| outputs[].object_url | string | 结果文件下载 URL。 |
| outputs[].output_ext | string | 文件扩展名（含 `.`）。 |
| outputs[].cost_time | integer | 从任务开始到产出此结果的耗时，单位毫秒。 |
| outputs[].error_type | string | 成功为 `"NOT_ERROR"`。 |
| cost_times | object | 耗时统计，单位毫秒。 |
| cost_times.inference_cost_time | integer | 推理耗时，单位毫秒。 |
| cost_times.running_cost_time | integer | 运行耗时，单位毫秒。 |
| cost_times.total_cost_time | integer | 总耗时，单位毫秒。 |
| cost_times.real_cpu_cost_time | integer | CPU 耗时，单位毫秒。 |
| cost_times.real_total_cost_time | integer | 与 `real_cpu_cost_time` 对应的总耗时，单位毫秒。 |

### 3. 异步查询模式

启用方式：在请求头加入 `X-Bizyair-Task-Async: enable`。

**步骤 1：发起异步任务，立即获取 `request_id`**

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://api.bizyair.ai/v1/webapp/task/openapi/create"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json",
    "X-Bizyair-Task-Async": "enable"
}
payload = {
    "web_app_id": 58151,
    "suppress_preview_output": False,
    "input_values": {
        "13:ResolutionSelector.aspect_ratio": "1:1 (Square)",
          "471:TextEncodeQwenImage21.prompt": "A cute cat, simple style",
          "471:TextEncodeQwenImage21.negative_prompt": ""
    }
}

response = requests.post(url, json=payload, headers=headers, timeout=30)
response.raise_for_status()
result = response.json()
request_id = result.get("request_id")
if not request_id:
    raise RuntimeError("Failed to get request_id")
print(f"任务已提交: {request_id}")
```

**响应示例**

```json
{
  "request_id": "5fc66bc7-2c07-446c-af6b-7bf1f8f8ddc3"
}
```

**完整 Python 示例（含轮询）**

```python
import os
import time
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
base_url = "https://api.bizyair.ai/v1/webapp/task/openapi"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

# 1. 提交异步任务
async_headers = {**headers, "X-Bizyair-Task-Async": "enable"}
payload = {
    "web_app_id": 58151,
    "suppress_preview_output": False,
    "input_values": {
        "13:ResolutionSelector.aspect_ratio": "1:1 (Square)",
          "471:TextEncodeQwenImage21.prompt": "A cute cat, simple style",
          "471:TextEncodeQwenImage21.negative_prompt": ""
    }
}
resp = requests.post(f"{base_url}/create", json=payload, headers=async_headers)
resp.raise_for_status()
result = resp.json()
request_id = result.get("request_id")
if not request_id:
    raise RuntimeError("Failed to get request_id")
print(f"任务已提交: {request_id}")

# 2. 轮询状态
while True:
    r = requests.get(f"{base_url}/{request_id}", headers=headers)
    r.raise_for_status()
    status_result = r.json()
    if status_result.get("code") != 20000 or status_result.get("status") is not True:
        raise RuntimeError(status_result.get("message") or "Query failed")
    status = status_result["data"]["status"]
    print(f"当前状态: {status}")
    if status == "Success":
        break
    elif status in ("Failed", "Canceled"):
        print(f"任务失败或取消: {status_result['data'].get('message')}")
        exit(1)
    time.sleep(2)

# 3. 获取结果
r = requests.get(f"{base_url}/{request_id}/outputs", headers=headers)
r.raise_for_status()
outputs_result = r.json()
if outputs_result.get("code") != 20000 or outputs_result.get("status") is not True:
    raise RuntimeError(outputs_result.get("message") or "Get outputs failed")
for out in outputs_result["data"]["outputs"]:
    print("结果 URL:", out["object_url"])
```

### 4. WebHook 回调模式

启用方式：在请求头加入 `X-Bizyair-Task-WebHook-Url`。

**步骤 1：发起带 WebHook 的任务**

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://api.bizyair.ai/v1/webapp/task/openapi/create"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json",
    "X-Bizyair-Task-WebHook-Url": "https://your-server.com/api/callback",
    "X-Bizyair-Task-Authorization": "Bearer YOUR_CALLBACK_TOKEN"
}
payload = {
    "web_app_id": 58151,
    "suppress_preview_output": False,
    "input_values": {
        "13:ResolutionSelector.aspect_ratio": "1:1 (Square)",
        "471:TextEncodeQwenImage21.prompt": "A cute cat, simple style",
        "471:TextEncodeQwenImage21.negative_prompt": ""
    }
}

response = requests.post(url, json=payload, headers=headers, timeout=30)
response.raise_for_status()
result = response.json()
request_id = result.get("request_id")
if not request_id:
    raise RuntimeError("Failed to get request_id")
print(f"任务已提交: {request_id}")
```

**步骤 2：在回调端接收结果**

任务完成后 BizyAir 会向你指定的 URL 发起 POST 请求：

- **Method**: `POST`
- **Headers**: `Content-Type: application/json`、`User-Agent: Go-http-client/1.1`、`Authorization`（由创建请求里的 `X-Bizyair-Task-Authorization` 改写而来）以及其他 `X-Bizyair-Task-*` Header
- **Body**:

```json
{
  "type": "API",
  "status": "Success",
  "created_at": "2026-09-28T07:49:59Z",
  "updated_at": "2026-09-28T07:50:10Z",
  "executed_at": "2026-09-28T07:49:59Z",
  "running_at": "2026-09-28T07:50:07Z",
  "ended_at": "2026-09-28T07:50:10Z",
  "expired_at": "2026-10-13T00:00:00Z",
  "request_id": "2e752d06-3585-4a70-a55e-7c552d1399ef",
  "outputs": [
    {
      "object_url": "https://storage.bizyair.ai/outputs_examples/lXobDnouo90P3XoF1stRJhup4mgx4n26.png",
      "output_ext": ".png",
      "cost_time": 4030,
      "error_type": "NOT_ERROR"
    }
  ],
  "cost_times": {
    "inference_cost_time": 4149,
    "running_cost_time": 11742,
    "total_cost_time": 11776,
    "real_cpu_cost_time": 430,
    "real_total_cost_time": 430
  }
}
```

**Body 字段说明**：

| 参数名 | 类型 | 说明 |
| :--- | :--- | :--- |
| type | string | 任务类型，如 `"API"`。 |
| status | string | 任务状态：`Queuing` 排队中 / `Preparing` 准备中 / `Running` 运行中 / `Success` 成功 / `Failed` 失败 / `Canceled` 已取消。 |
| request_id | string | 请求 ID。 |
| created_at | string | 创建时间（ISO 8601，UTC）。 |
| updated_at | string | 最后更新时间（ISO 8601，UTC）。 |
| executed_at | string | 开始执行时间（ISO 8601，UTC）。 |
| running_at | string | 进入运行的时间（ISO 8601，UTC）。 |
| ended_at | string | 结束时间（ISO 8601，UTC）。 |
| expired_at | string | 结果文件过期时间（ISO 8601，UTC）。 |
| outputs | array[object (Output)] | 结果列表。 |
| outputs[].object_url | string | 结果文件下载 URL。 |
| outputs[].output_ext | string | 文件扩展名（含 `.`）。 |
| outputs[].cost_time | integer | 从任务开始到产出此结果的耗时，单位毫秒。 |
| outputs[].error_type | string | 成功为 `"NOT_ERROR"`。 |
| cost_times | object | 耗时统计，单位毫秒。 |
| cost_times.inference_cost_time | integer | 推理耗时，单位毫秒。 |
| cost_times.running_cost_time | integer | 运行耗时，单位毫秒。 |
| cost_times.total_cost_time | integer | 总耗时，单位毫秒。 |
| cost_times.real_cpu_cost_time | integer | CPU 耗时，单位毫秒。 |
| cost_times.real_total_cost_time | integer | 与 `real_cpu_cost_time` 对应的总耗时，单位毫秒。 |

**WebHook 请求头说明**

| 参数名 | 类型 | 必填 | 说明 |
| :--- | :--- | :--- | :--- |
| `X-Bizyair-Task-WebHook-Url` | string | 选填 | 任务结束后的回调地址。不设置则不发送回调，请用查询接口主动获取结果。回调地址需使用 HTTP 或 HTTPS，且应接收 POST 请求。若回调地址在海外，BizyAir 不保证回调成功。 |
| `X-Bizyair-Task-Authorization` | string | 选填 | 如回调接口需要鉴权，请把凭据写在这里。它是唯一会被改写的请求头：回调时去掉 `X-Bizyair-Task-` 前缀，以 `Authorization: ${YOUR_WEBHOOK_AUTHORIZATION}` 的形式送达。 |
| `X-Bizyair-Task-${HEADER_NAME}` | string | 选填 | 除 `X-Bizyair-Task-Authorization` 外，其余以 `X-Bizyair-Task-` 开头的请求头都会连同前缀原样出现在回调请求头中。 |

**回调端 Python 示例**

```python
from flask import Flask, request, jsonify
import os

app = Flask(__name__)
EXPECTED_TOKEN = os.environ.get("CALLBACK_TOKEN")

@app.route("/api/callback", methods=["POST"])
def callback():
    auth = request.headers.get("Authorization", "")
    if EXPECTED_TOKEN:
        if not auth.startswith("Bearer ") or auth.split(" ")[1] != EXPECTED_TOKEN:
            return jsonify({"message": "invalid token"}), 401

    payload = request.get_json()
    request_id = payload.get("request_id")
    status = payload.get("status")
    outputs = payload.get("outputs")

    print(f"Received callback for request_id: {request_id}")
    print(f"Status: {status}")
    print(f"Outputs: {outputs}")

    return jsonify({"ok": True}), 200

if __name__ == "__main__":
    app.run(port=3000)
```

> 回调以 HTTP 200 视为接收成功。返回其他状态码时，BizyAir 会重试，一共 10 次，每次间隔 200 毫秒。

---

## 四. 查询结果

在这个阶段，您可以通过刚才产生的**请求 ID（request_id）**，去主动查询任务是否完成，以及获取最终结果。

### 1. 查询任务状态

获取任务的当前状态与详细信息（排队信息、运行时长等）。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
request_id = "your_request_id_here"
url = f"https://api.bizyair.ai/v1/webapp/task/openapi/{request_id}"
headers = {
    "Authorization": f"Bearer {api_key}"
}

response = requests.get(url, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print(result)
```

**响应示例**：

```json
{
  "code": 20000,
  "message": "Ok",
  "status": true,
  "data": {
    "type": "API",
    "status": "Success",
    "created_at": "2026-09-23T07:44:34Z",
    "updated_at": "2026-09-23T07:45:19Z",
    "executed_at": "2026-09-23T07:44:34Z",
    "ended_at": "2026-09-23T07:45:19Z",
    "expired_at": "2026-10-08T00:00:00Z",
    "cost_times": {
      "inference_cost_time": 42930,
      "running_cost_time": 44873,
      "total_cost_time": 44880,
      "real_cpu_cost_time": 430,
      "real_total_cost_time": 430
    }
  }
}
```

**响应字段说明**：

| 参数名 | 类型 | 说明 |
| :--- | :--- | :--- |
| code | integer | 响应状态码，20000 表示成功。 |
| message | string | 接口层面的提示或错误信息。 |
| status | boolean | 接口调用是否成功。与表示任务状态的字符串 `data.status` 不是同一个字段。 |
| data | object | 任务内容。 |
| data.type | string | 任务类型，如 `"API"`。 |
| data.status | string | 任务状态：`Queuing` 排队中 / `Preparing` 准备中 / `Running` 运行中 / `Success` 成功 / `Failed` 失败 / `Canceled` 已取消。 |
| data.created_at | string | 创建时间（ISO 8601，UTC）。 |
| data.updated_at | string | 最后更新时间（ISO 8601，UTC）。 |
| data.executed_at | string | 开始执行时间（ISO 8601，UTC）。尚未执行时可能不返回。 |
| data.ended_at | string 或 null | 结束时间（ISO 8601，UTC）。未结束时为 `null` 或不返回。 |
| data.expired_at | string | 结果文件过期时间（ISO 8601，UTC）。 |
| data.cost_times | object | 耗时统计，单位毫秒。未记录时为空对象 `{}`。 |
| data.cost_times.inference_cost_time | integer | 推理耗时，单位毫秒。 |
| data.cost_times.running_cost_time | integer | 运行耗时，单位毫秒。 |
| data.cost_times.total_cost_time | integer | 总耗时，单位毫秒。 |
| data.cost_times.real_cpu_cost_time | integer | CPU 耗时，单位毫秒。 |
| data.cost_times.real_total_cost_time | integer | 与 `real_cpu_cost_time` 对应的总耗时，单位毫秒。 |
| data.queue_info | object 或 null | 仅 `Queuing` 时出现。 |
| data.queue_info.queue_count | integer | 大于 0 才是前面的任务数，`-1` 不表示具体人数。 |

**状态枚举**：

| 状态值 | 含义 | 建议操作 |
| :--- | :--- | :--- |
| `Queuing` | 排队中，等待资源调度 | 继续轮询，间隔 1-3 秒 |
| `Preparing` | 准备中，正在加载模型 / 环境 | 继续轮询 |
| `Running` | 运行中 | 继续轮询 |
| `Success` | 成功 | **停止轮询**，获取结果 |
| `Failed` | 失败 | **停止轮询**。失败原因在「获取任务结果」的 `outputs[].error_msg` |
| `Canceled` | 已取消 | **停止轮询** |

### 2. 获取任务结果

获取**已完成任务**的结果（`object_url` 等）。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
request_id = "your_request_id_here"
url = f"https://api.bizyair.ai/v1/webapp/task/openapi/{request_id}/outputs"
headers = {
    "Authorization": f"Bearer {api_key}"
}

response = requests.get(url, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
for out in result["data"]["outputs"]:
    print("结果 URL:", out["object_url"])
```

**响应示例**：

```json
{
  "code": 20000,
  "message": "Ok",
  "status": true,
  "data": {
    "request_id": "29f53793-12d3-4dd3-b2a8-4d9848e0c7da",
    "status": "Success",
    "outputs": [
      {
        "object_url": "https://storage.bizyair.ai/outputs_examples/lXobDnouo90P3XoF1stRJhup4mgx4n26.png",
        "output_ext": ".png",
        "cost_time": 10657,
        "audit_status": 2,
        "error_type": "NOT_ERROR"
      }
    ]
  }
}
```

**响应字段说明**：

| 参数名 | 类型 | 说明 |
| :--- | :--- | :--- |
| code | integer | 响应状态码，20000 表示成功。 |
| message | string | 接口层面的提示或错误信息。 |
| status | boolean | 接口调用是否成功。与表示任务状态的字符串 `data.status` 不是同一个字段。 |
| data | object | 任务结果。 |
| data.request_id | string | 请求 ID。 |
| data.status | string | 任务状态：`Queuing` 排队中 / `Preparing` 准备中 / `Running` 运行中 / `Success` 成功 / `Failed` 失败 / `Canceled` 已取消。 |
| data.outputs | array[object (Output)] | 结果列表。 |
| data.outputs[].object_url | string | 结果文件下载 URL。 |
| data.outputs[].output_ext | string | 文件扩展名（含 `.`）。 |
| data.outputs[].cost_time | integer | 从任务开始到产出此结果的耗时，单位毫秒。 |
| data.outputs[].audit_status | integer | 审核状态：`0` 无需审查 / `1` 尚未审查 / `2` 通过 / `3` 未通过 / `4` 审查报错。 |
| data.outputs[].error_type | string | 失败时为具体错误码，成功为 `"NOT_ERROR"`。 |
| data.outputs[].error_msg | string | 失败时的详细原因。成功时可能不返回。 |

> **备注**：输出 URL 为 CDN 下载链接，不保证长期可用，请在任务完成后尽快转存到您自己的存储。

---

## 五. 文件上传

当工作流的 `input_values` 中需要传入图片、音频或视频时（例如 `LoadImage.image`、`ref_images` 等字段），不能直接传本地文件路径或 base64，必须先将文件上传到 BizyAir OSS，获得 URL 后再作为 `input_values` 的值传入。

**上传流程**：获取凭证 → 上传到 OSS → 提交资源。

### 1. 获取上传凭证与参数

调用获取上传凭证接口，服务端会返回本次上传所需的 OSS 信息与临时 STS 凭证。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://api.bizyair.ai/v1/upload/token"
params = {
    "file_name": "example.png",
    "file_type": "inputs_temp"
}
headers = {
    "Authorization": f"Bearer {api_key}"
}

response = requests.get(url, params=params, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print(result)
```

**获取凭证响应**：

```json
{
  "code": 20000,
  "message": "Ok",
  "status": true,
  "data": {
    "file": {
      "object_key": "inputs_temp/20260907/example.png",
      "access_key_id": "STS.xxxx",
      "access_key_secret": "xxxx",
      "expiration": "2026-09-07T08:00:00Z",
      "security_token": "xxxx"
    },
    "storage": {
      "endpoint": "oss-us-east-1.aliyuncs.com",
      "bucket": "example-bucket",
      "region": "oss-example-region"
    },
    "access_url": "https://storage.example/inputs_temp/20260907/example.png"
  }
}
```

### 2. 使用阿里云 OSS 简单上传

使用上一步返回的 `endpoint`、`bucket`、`region`、`object_key` 与 STS 凭证将本地文件上传到 OSS。

**Python 示例**：

```python
# pip install alibabacloud-oss-v2
import os
import alibabacloud_oss_v2 as oss

def upload_to_oss(region, endpoint, bucket, object_key, file_path, access_key_id, access_key_secret, security_token):
    os.environ["OSS_ACCESS_KEY_ID"] = access_key_id
    os.environ["OSS_ACCESS_KEY_SECRET"] = access_key_secret
    os.environ["OSS_SESSION_TOKEN"] = security_token

    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg.region = region[4:] if region.startswith("oss-") else region
    cfg.endpoint = endpoint

    client = oss.Client(cfg)
    return client.put_object_from_file(
        oss.PutObjectRequest(bucket=bucket, key=object_key),
        file_path,
    )
```

> **注意**：
> - 有些 SDK 需要去掉 `region` 的 `oss-` 前缀，如 `oss-us-east-1` → `us-east-1`。
> - 建议同时设置 `region` 与 `endpoint`，以返回的 `endpoint` 为准。
> - OSS 简单上传需要 STS 签名，curl 不提供可执行示例。请切换到 Python 或 JavaScript。

### 3. 提交输入资源

当 OSS 上传成功后，提交本次输入资源，便于后续任务直接引用。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://meta.bizyair.ai/v1/input_resource/commit"
payload = {
    "name": "example.png",
    "object_key": "inputs_temp/20260907/example.png"
}
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print(result)
```

**提交时必须复用获取凭证响应中的 `data.file.object_key`。提交成功后，将响应中的 `data.url` 写入调用请求里对应的 `input_values`**：

```json
{
  "code": 20000,
  "message": "Ok",
  "status": true,
  "data": {
    "id": 1711,
    "name": "example.png",
    "ext": ".png",
    "url": "https://storage.example/inputs_temp/20260907/example.png"
  }
}
```

### 4. 查询 inputs 列表（可选）

这项操作可以查询您的 inputs 列表，用于校验上传的内容。提交前若还没有记录，响应的 `data` 可能只有 `current` 和 `page_size`，没有 `list`；有记录后才会出现 `list` 和 `total`。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
url = "https://meta.bizyair.ai/v1/input_resource"
params = {
    "current": 1,
    "page_size": 20
}
headers = {
    "Authorization": f"Bearer {api_key}"
}

response = requests.get(url, params=params, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print(result)
```

---

## 六. 任务管理

### 1. 取消任务

仅对仍处于**排队中**（`Queuing`）的任务生效，且需要在任务离开队列前立刻调用。成功时返回 HTTP 200、`code` 20000，随后状态为 `Canceled`，不计费。

若先查询再取消，任务可能已经离开队列，返回 HTTP 409、`code` 30035（`The task has been executed, please use interrupt.`）。此时请改用中断任务。

HTTP 202、`code` 20002 只表示取消请求已提交，不表示任务已经取消。请再次查询状态确认。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
request_id = "your_request_id_here"
url = f"https://api.bizyair.ai/v1/webapp/task/openapi/{request_id}/cancel"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

response = requests.put(url, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print("任务已取消")
```

> **注意**：对已取消的任务再次调用取消，返回 HTTP 409、`code` 30036（`Task not in queue status.`）。任务状态保持 `Canceled`，不会因此扣费。

### 2. 中断任务

仅对**运行中**（`Running`）的任务生效，强制停止。成功时返回 HTTP 200、`code` 20000。任务不在 `Running` 时返回 HTTP 409、`code` 30038。中断成功后状态为 `Failed`，结果中 `error_type` 为 `INTERRUPT`，`error_msg` 为 `Interrupted`。

```python
import os
import requests

api_key = os.environ["BIZYAIR_API_KEY"]
request_id = "your_request_id_here"
url = f"https://api.bizyair.ai/v1/webapp/task/openapi/{request_id}/interrupt"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

response = requests.put(url, headers=headers)
response.raise_for_status()
result = response.json()
if result.get("code") != 20000 or result.get("status") is not True:
    raise RuntimeError(result.get("message") or "BizyAir API request failed")
print("任务已中断")
```

> **警告**：中断运行中的任务，**已执行部分仍会产生费用**。仅当任务处于 `Queuing` 时取消才不计费。

---

## 附录：日志字段脱敏

为保护您的业务敏感信息（如 prompt 设计等），可以在请求头中指定调用记录里需要脱敏的字段。接口返回的请求体仍是原值，`mask_fields` 只记录这些字段路径。调用记录页面会把匹配到的字段显示为 `[调用方要求隐藏]`，不影响实际请求的执行和计费。

**使用方法**：在请求头中携带 `X-Bizyair-Log-Mask-Fields`。值是 `input_values` 里的 key。多个字段用英文逗号分隔。

```http
Content-Type: application/json
Authorization: Bearer ${BIZYAIR_API_KEY}
X-Bizyair-Log-Mask-Fields: 471:TextEncodeQwenImage21.prompt
```