---
title: "Integrate OpenAI Python"
description: "Run non-streaming and streaming requests with the E2E-tested OpenAI Python client."
---

> Documentation Index
> Fetch the complete documentation index at: https://nexus.microvoid.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Integrate OpenAI Python

## Goal

用同一个 OpenAI Python Client 完成普通请求和 Streaming 请求。

## Outcome

普通响应和流式文本都保留上游原始模型名；流结束时读取 Usage，Nexus 再进入幂等结算终态。

## Prerequisites

- Gateway Base URL、Inference Key 和目标 Provider 接受的模型名。
- Python 3 与虚拟环境支持。

## Lifecycle note

Streaming 会在 Provider 调用前 Reserve；正常完成、客户端取消、断开或超时后必须 Settle/Release/Expire。不要用 Key 生命周期推导 QuotaPeriod。

## 1. Install


```sh title="Terminal"
python3 -m venv .venv
. .venv/bin/activate
pip install openai==2.46.0
```


## 2. Configure


```sh title="Terminal"
export NEXUS_BASE_URL="https://gateway.nexus.microvoid.io/providers/00000000-0000-0000-0000-000000000141"
export NEXUS_API_KEY="<NEXUS_INFERENCE_KEY>"
export NEXUS_MODEL="provider/opaque-model-v9"
```


## 3. Run the complete client


```python title="openai-python.py"
"""Public OpenAI Python SDK quickstart executed by the Nexus compatibility E2E suite."""

import json
import os

import openai
from openai import OpenAI


def required_environment(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


client = OpenAI(
    base_url=required_environment("NEXUS_BASE_URL"),
    api_key=required_environment("NEXUS_API_KEY"),
    max_retries=0,
    timeout=10.0,
)
model = required_environment("NEXUS_MODEL")
messages = [{"role": "user", "content": "Keep this Python request unchanged."}]

completion = client.chat.completions.create(model=model, messages=messages)
stream = client.chat.completions.create(
    model=model,
    messages=messages,
    stream=True,
    stream_options={"include_usage": True},
)
stream_text = ""
stream_usage = None
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        stream_text += chunk.choices[0].delta.content
    if chunk.usage is not None:
        stream_usage = chunk.usage.total_tokens

print(
    json.dumps(
        {
            "sdk_version": openai.__version__,
            "model": completion.model,
            "content": completion.choices[0].message.content,
            "stream_text": stream_text,
            "stream_usage": stream_usage,
        }
    )
)
```


```sh title="Terminal"
python openai-python.py
```

## Check your work

JSON 输出应包含 SDK 版本、普通正文、拼接后的 `stream_text` 和末尾 `stream_usage`；对应 Reservation 最终不再占用 `reservedUnits`。

## Failure modes

- `metering_unsupported`：路径没有 Usage Inspector，Provider 未被调用。
- 流中断：用 `x-nexus-request-id` 查 Invocation 和最终 Reservation 状态，不要盲目重复业务请求。
- Provider Usage 缺失或不合法：按同一 Request ID 查看 Inspector/Provider Attempt 事实。

## Security notes

Key 只从 `NEXUS_API_KEY` 读取。不要记录 Authorization Header、流式正文中的敏感内容或 Provider Secret。

## Next steps

阅读 [Reserve, settle, and release](/docs/control-plans-and-quota/reserve-settle-release) 和 [Quota diagnosis](/cookbook/diagnose-quota-metering)。

Source: https://nexus.microvoid.io/cookbook/openai-python/index.mdx