Python 示例

使用 OpenAI 官方 Python SDK，把 base_url 指向 kukuai.fyi 即可调用。提供同步、异步、流式、function calling 与重试范式。

直接使用 OpenAI 官方 SDK 即可，不需要任何 kukuai 私有依赖。建议 1.40 及以上版本，能完整支持流式 / tools / response_format。

请将下方脚本中的 sk-xxx 替换为你在 https://kukuai.fyi 控制台「API Keys」页面创建的 Key。

bash

pip install openai>=1.40.0

# 注入 API Key（推荐使用环境变量而不是硬编码）
export KUKUAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

基础调用

basic.pypython

from openai import OpenAI

# 唯一的差异：base_url 指向 kukuai.fyi
client = OpenAI(
    base_url="https://www.kukuai.fyi/api-proxy/china/v1",
    api_key="YOUR_KUKUAI_API_KEY",  # 或读 os.environ["KUKUAI_API_KEY"]
)

resp = client.chat.completions.create(
    model="claude-opus-4-7",
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "用 30 字介绍 kukuai.fyi"},
    ],
    temperature=0.7,
    max_tokens=1024,
)

print(resp.choices[0].message.content)
print("usage:", resp.usage.total_tokens, "tokens")

流式响应

stream.pypython

stream = client.chat.completions.create(
    model="claude-opus-4-7",
    stream=True,
    messages=[{"role": "user", "content": "讲一个关于 API 网关的冷笑话"}],
)

for chunk in stream:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)
print()  # 换行

function calling

tools.pypython

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询某个城市的当前天气",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

resp = client.chat.completions.create(
    model="claude-opus-4-7",
    messages=[{"role": "user", "content": "上海现在几度？"}],
    tools=tools,
    tool_choice="auto",
)

tc = resp.choices[0].message.tool_calls
if tc:
    print("model wants to call:", tc[0].function.name, tc[0].function.arguments)

当模型返回 tool_calls 时，由你的应用执行工具，再把结果以 role: "tool" 消息回传给模型，发起下一轮请求。

错误处理与重试

retry.pypython

import time
from openai import OpenAI, RateLimitError, APIStatusError

client = OpenAI(base_url="https://www.kukuai.fyi/api-proxy/china/v1")

def call_with_retry(messages, model="claude-opus-4-7", max_retries=3):
    """简易退避重试包装：仅对可重试错误退避。"""
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages,
            )
        except RateLimitError:
            # 429：尊重 Retry-After，简单退避即可
            time.sleep(2 ** attempt)
        except APIStatusError as e:
            # 5xx 才退避，4xx 直接抛出
            if 500 <= e.status_code < 600:
                time.sleep(2 ** attempt)
            else:
                raise
    raise RuntimeError("max retries exceeded")

完整错误码语义见错误码页。

异步用法

高并发或在 FastAPI / asyncio 应用中，建议直接使用 AsyncOpenAI：

async.pypython

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="https://www.kukuai.fyi/api-proxy/china/v1")

async def main():
    resp = await client.chat.completions.create(
        model="claude-opus-4-7",
        messages=[{"role": "user", "content": "hi"}],
    )
    print(resp.choices[0].message.content)

asyncio.run(main())

Python 示例 ​

基础调用 ​

流式响应 ​

function calling ​