API 最佳实践

Token 管好、错误重试、限流尊重之后，集成质量取决于三个高频场景：Webhook 触发、批量操作与 SSE 流式消费。本页收敛这三处的 Do / Don't，具体端点细节链到各自参考页。

Token 与认证

JWT 没有刷新端点：临近 session_timeout_days（默认 30 天）过期前，用 POST /api/v1/login/access-token 重新登录换取。服务端之间用 API Key（最小权限、定期轮换、不同服务不同 Key），泄露立即撤销。

应该做：Token 存环境变量或安全存储，不同环境用不同 Token，监控 Token 使用。不要做：硬编码进源码、提交到版本库、生产 Token 用于开发、用户之间共享 Token。

import os

class TokenManager:
    """临近过期重新登录换取 Token（无刷新端点）。"""

    def get_token(self):
        if not self.token or self.is_expiring_soon():
            data = requests.post(
                f"{API_BASE_URL}/api/v1/login/access-token",
                data={"username": os.getenv("CLOUISLE_USERNAME"), "password": os.getenv("CLOUISLE_PASSWORD")},
            ).json()["data"]
            self.token = data["access_token"]
        return self.token

POST /api/v1/workflows/webhook/{webhook_token} 是工作流唯一允许 API Key 触发的入口：Token 写在 URL 路径中，同时 Authorization 头必须携带 clou_ 前缀的 API Key。目标工作流必须已发布且触发类型为 webhook，Token 与流 URL 不是公开免认证访问，工作流访问规则仍然生效。需要更换 Token 时调用 POST /api/v1/workflows/{workflow_id}/regenerate-webhook-token。

应该做：把 webhook_token 当密钥保管（泄露即重新生成），触发失败先检查发布状态与触发类型，用幂等或去重逻辑处理外部系统的重试投递。不要做：把 Token 拼进可公开访问的前端代码、假设 Webhook 免认证可匿名调用、触发草稿或 manual 触发类型的工作流。

curl -X POST "https://your-domain.com/api/v1/workflows/webhook/wh_abc123" \
  -H "Authorization: Bearer clou_xxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{"customer_email": "customer@example.com", "inquiry_text": "I need help"}'

详见工作流执行。

批量操作

只用真实存在的批量端点（见约定），不要臆测通用 /batch 接口：批量删除会话用 DELETE /api/v1/conversations?ids=...（需对全部会话都有删除权限，否则返回 3000 或 4000）；团队模型授权/撤销用 POST / DELETE /api/v1/teams/{team_id}/models/batch（仅超级管理员）；检索配置对比用 POST /api/v1/knowledge-bases/{kb_id}/search/batch（1 个查询 × 最多 10 个配置，单个配置失败不影响其他）；文件解析用 POST /api/v1/upload/parse/batch（最多 5 个文件，超出返回 1001；混合成功时 data 只含成功项）。

应该做：批量授权代替 N 次单模型授权，对比检索配置时用 search/batch 复用同一次嵌入，解析多文件时用 parse/batch 代替循环。不要做：给任意资源拼 /batch 路径、一次发送超过 5 个解析文件或超过 10 个检索配置、假设 deleted_count 一定等于请求数（无权限的 ID 会被忽略或报错）。

SSE 流式消费

Agent 聊天流 POST /api/v1/agents/{agent_id}/chat/stream 与工作流运行流 GET /api/v1/workflows/runs/{run_id}/stream 都返回 text/event-stream：请求带 Accept: text/event-stream，用 curl -N 或流式 reader 逐行解析 event: / data:。工作流运行是异步的：run 端点立即返回 run_id 与 stream_url，再连流 URL 消费 workflow_start、node_complete、workflow_complete、workflow_error 等事件。

应该做：用 Webhook / SSE 接收更新而不是轮询状态接口，断线后用 from_sequence=0 重连工作流事件流，给流消费设置超时与重连上限。不要做：高频轮询 GET /runs/{run_id} 等状态、对 429 立即重试（用指数退避）、把原始错误直接展示给终端用户。

curl -N "https://your-domain.com/api/v1/workflows/runs/$RUN_ID/stream?from_sequence=0" \
  -H "Authorization: Bearer $CLOUISLE_TOKEN"

详见流式事件与工作流执行。

错误与限流

先判断响应 code 再使用 data；校验失败为 HTTP 422，data.errors 是字段到消息列表的字典。重试只给可重试错误用指数退避（最多 3 次）：4xx 客户端错误与 2xxx 认证错误不重试。Clouisle 不返回 X-RateLimit-* 响应头，限流只作用于特定安全敏感流程（登录锁定、TOTP 锁定、邮件配额、模型配额，如 6103 模型配额超限、5400 邮件/ provider 限流），遇到 HTTP 429 先退避再重试。

import time

def retry_with_backoff(func, max_retries=3, delay=1.0):
    for attempt in range(max_retries):
        try:
            return func()
        except ApiError as e:
            if 2000 <= e.code < 3000 or 4000 <= e.code < 5000:
                raise
            if attempt == max_retries - 1:
                raise
            time.sleep(delay)
            delay *= 2

缓存与分页

大数据集用分页逐页取（page / page_size，信封见约定），不要一次拉全量；稳定的读结果做客户端缓存；UI 搜索输入做防抖，不要每个键都发请求。

上线前检查

全部请求走 HTTPS，Token 与 Key 存安全存储
JWT 临近过期重新登录（无刷新端点），API Key 最小权限并定期轮换
Webhook Token 当密钥保管，触发前确认工作流已发布且为 webhook 类型
只用文档列出的批量端点，遵守 5 文件 / 10 配置上限
流式消费用 SSE + 重连，而不是轮询状态接口
422 校验、6103 / 5400 配额、429 退避都有处理路径
在测试环境验证错误与边界场景，不用生产凭据测试