Integrate OpenAI Python

Run non-streaming and streaming requests with the E2E-tested OpenAI Python client.

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

Terminalsh

python3 -m venv .venv
. .venv/bin/activate
pip install openai==2.46.0

2. Configure

Terminalsh

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

openai-python.pypython

"""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,
        }
    )
)

Terminalsh

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 和 Quota diagnosis。