> ## Documentation Index
> Fetch the complete documentation index at: https://docs.tapapi.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python

> Python 服务端、脚本和批量任务接入 Tapapi

Python 适合后端服务、批量跑图、数据处理和自动化脚本。新项目建议优先使用 OpenAI SDK 的兼容模式；不方便安装 SDK 时，也可以使用 `requests` 调用 HTTP 接口。

<Warning>不要把 Tapapi API Key 写进浏览器前端、公开 Notebook、公开仓库或客户端配置。Python 示例默认运行在服务端、脚本或队列 worker 中。</Warning>

## 安装

```bash theme={null}
pip install openai requests
```

## 环境变量

```bash theme={null}
export TAPAPI_API_KEY="sk-xxx"
export TAPAPI_BASE_URL="https://tapapi.ai/v1"
```

## 初始化客户端

```python theme={null}
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["TAPAPI_API_KEY"],
    base_url=os.getenv("TAPAPI_BASE_URL", "https://tapapi.ai/v1"),
    timeout=120.0,
)
```

## 文本请求

```python theme={null}
resp = client.chat.completions.create(
    model="gpt-5.4",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "用一句话解释 Tapapi 是什么"},
    ],
)

print(resp.choices[0].message.content)
```

## 流式输出

```python theme={null}
stream = client.chat.completions.create(
    model="gpt-5.4",
    stream=True,
    stream_options={"include_usage": True},
    messages=[{"role": "user", "content": "写一段 100 字的产品介绍"}],
)

for chunk in stream:
    if not chunk.choices:
        continue

    delta = chunk.choices[0].delta
    text = getattr(delta, "content", None)
    if text:
        print(text, end="", flush=True)
```

流式响应通常读取 `choices[0].delta.content`。生产代码要跳过空内容、保活事件和没有普通文本 delta 的 chunk。最终 usage 可能受上游或网络影响缺失，计费以控制台账单记录为准。

## Responses 请求

`/v1/chat/completions` 是 Python 接入的默认主线。需要 Responses API 时，先确认你的 `openai` SDK 版本、模型详情和返回字段：

```python theme={null}
resp = client.responses.create(
    model="gpt-5.4",
    input="用一句话解释 Tapapi 是什么",
)

print(resp.output_text)
```

Responses 的返回结构和 Chat Completions 不完全一样，不要共用同一套解析代码。

## 图片生成

```python theme={null}
resp = client.images.generate(
    model="nano-banana-pro",
    prompt="a clean product photo of a white ceramic mug",
    n=1,
    size="1024x1024",
    response_format="url",
)

if not resp.data:
    raise RuntimeError("Tapapi returned empty image data")

item = resp.data[0]
image = item.url or item.b64_json
if not image:
    raise RuntimeError("Tapapi returned image data without url or b64_json")

print(image)
```

图片返回可能是 `url` 或 `b64_json`。生产环境建议把图片转存到自己的对象存储，不要长期依赖临时 URL。批量生成时还要记录请求 `n`、实际返回数量和成功转存数量。

## 保存图片结果

```python theme={null}
import base64
import pathlib
import requests

def save_image_item(item, path: str) -> None:
    if item.url:
        response = requests.get(item.url, timeout=60)
        response.raise_for_status()
        pathlib.Path(path).write_bytes(response.content)
        return

    if item.b64_json:
        pathlib.Path(path).write_bytes(base64.b64decode(item.b64_json))
        return

    raise ValueError("image item has no url or b64_json")

save_image_item(item, "output.png")
```

如果 URL 下载失败，优先重试下载和转存，不要立刻重新生成图片。

## 原生 HTTP

```python theme={null}
import os
import requests

def parse_tapapi_error(response: requests.Response) -> dict:
    request_id = response.headers.get("X-Oneapi-Request-Id")

    try:
        payload = response.json()
    except ValueError:
        payload = {}

    error = payload.get("error") or {}
    return {
        "status_code": response.status_code,
        "request_id": request_id,
        "type": error.get("type"),
        "code": error.get("code") or payload.get("code"),
        "message": error.get("message") or payload.get("message") or response.text,
    }

response = requests.post(
    "https://tapapi.ai/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {os.environ['TAPAPI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json={
        "model": "gpt-5.4",
        "messages": [{"role": "user", "content": "Hello"}],
    },
    timeout=120,
)

if not response.ok:
    raise RuntimeError(parse_tapapi_error(response))

data = response.json()
print(data["choices"][0]["message"]["content"])
```

## 错误处理

```python theme={null}
from openai import APIError, APIStatusError

try:
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": "Hello"}],
    )
except APIStatusError as exc:
    request_id = exc.response.headers.get("X-Oneapi-Request-Id")
    try:
        body = exc.response.json()
    except ValueError:
        body = {}

    error = body.get("error") or {}
    print({
        "status_code": exc.status_code,
        "request_id": request_id,
        "type": error.get("type"),
        "code": error.get("code"),
        "message": error.get("message") or exc.response.text,
    })
    raise
except APIError as exc:
    print(exc)
    raise
```

常见处理原则：

| 状态                       | 建议                         |
| ------------------------ | -------------------------- |
| 400                      | 不重试，修请求字段                  |
| 401 / 403                | 不重试，检查 API Key、账户状态、余额和权限  |
| 429                      | 指数退避，限制最大重试次数              |
| 临时 5xx / network timeout | 可以重试 1-2 次，并记录业务 `task_id` |
| 504 / 524                | 先查业务状态和账单，再决定是否补跑          |

更多规则见 [错误码与重试](/errors)。

## 生产建议

| 场景     | 建议                                                      |
| ------ | ------------------------------------------------------- |
| 批量任务   | 用队列和 worker 控制并发，不要一次性开大量请求                             |
| 重试     | 只对 429、临时 5xx、网络中断做有限重试，并加随机抖动                          |
| 幂等     | 每个业务任务保存自己的 `task_id`，避免重复点击或 worker 重启重复请求             |
| 日志     | 保存接口、模型、HTTP 状态码、`request_id`、错误字段和最终用量                 |
| 图片保存   | 同时兼容 `url` 和 `b64_json`，成功后尽快转存                         |
| 视频 API | 受控开放，先看 [视频 API](/video-generation/overview)，不要默认写进批量任务 |

## 下一步

| 场景                   | 文档                                                    |
| -------------------- | ----------------------------------------------------- |
| OpenAI SDK 迁移总览      | [OpenAI SDK 迁移](/integrations/openai-sdk)             |
| 不使用 SDK              | [HTTP REST](/integrations/http-rest)                  |
| Node.js / TypeScript | [Node.js / TypeScript](/integrations/node-typescript) |
| PHP / Laravel        | [PHP / Laravel](/integrations/php-laravel)            |
| 生产上线检查               | [上线 Checklist](/production/launch-checklist)          |