Python 示例

示例使用 requests，API Key 从 TOPAPIS_API_KEY 环境变量读取。

安装依赖

python -m pip install requests

查询模型并对话

模型列表是可安全重试的只读请求。下面的函数对网络中断、429 和临时 5xx 进行最多 3 次尝试，并遵循整数形式的 Retry-After。

import os
import random
import time

import requests

BASE_URL = "https://api.topapis.cn/v1"
API_KEY = os.environ["TOPAPIS_API_KEY"]
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}
RETRYABLE_STATUS = {429, 500, 502, 503}


def get_json_with_retry(session: requests.Session, url: str) -> dict:
    for attempt in range(1, 4):
        try:
            response = session.get(url, headers=HEADERS, timeout=(10, 45))
        except (requests.ConnectionError, requests.Timeout):
            if attempt == 3:
                raise
            delay = 2 ** (attempt - 1) + random.uniform(0, 0.5)
            time.sleep(delay)
            continue
        else:
            if response.status_code not in RETRYABLE_STATUS:
                response.raise_for_status()
                return response.json()
            if attempt == 3:
                response.raise_for_status()

            retry_after = response.headers.get("Retry-After", "")
            if retry_after.isdigit():
                delay = int(retry_after)
            else:
                delay = 2 ** (attempt - 1) + random.uniform(0, 0.5)
            time.sleep(delay)

    raise RuntimeError("unreachable")


with requests.Session() as session:
    models = get_json_with_retry(session, f"{BASE_URL}/models")
    model_id = models["data"][0]["id"]

    response = session.post(
        f"{BASE_URL}/chat/completions",
        headers=HEADERS,
        json={
            "model": model_id,
            "messages": [
                {"role": "user", "content": "简要说明模型发现接口的作用。"}
            ],
        },
        timeout=(10, 120),
    )
    response.raise_for_status()
    print(response.json()["choices"][0]["message"]["content"])

对话和生成请求可能产生计费，上例不会自动重试 POST。生产应用只有在确认幂等性后才应重试写请求。

图片结果兼容处理

import base64
from pathlib import Path

import requests


def save_image_result(item: dict, destination: Path) -> None:
    if item.get("b64_json"):
        destination.write_bytes(base64.b64decode(item["b64_json"], validate=True))
        return

    image_url = item.get("url")
    if not image_url:
        raise ValueError("Image response contains neither b64_json nor url")

    download = requests.get(image_url, timeout=(10, 120))
    download.raise_for_status()
    destination.write_bytes(download.content)

下载后还应验证文件类型、文件大小和真实像素尺寸。临时 URL 不应长期保存为最终资源地址。

异步任务轮询

import time

import requests


def wait_for_video(task_id: str, headers: dict, timeout_seconds: int = 600) -> dict:
    deadline = time.monotonic() + timeout_seconds
    url = f"https://api.topapis.cn/v1/video/generations/{task_id}"

    while time.monotonic() < deadline:
        response = requests.get(url, headers=headers, timeout=(10, 30))
        response.raise_for_status()
        payload = response.json()
        task = payload.get("data", payload)

        if task.get("status") == "SUCCESS":
            return task
        if task.get("status") == "FAILURE":
            raise RuntimeError("Video generation failed")

        time.sleep(15)

    raise TimeoutError("Video task did not finish before the deadline")

轮询超时后保留 task_id，稍后继续查询，而不是立即创建新任务。