Полный пример на Python: официальные SDK через наш шлюз

Официальные SDK anthropic и openai для Python не привязаны к серверам своих компаний намертво — оба принимают кастомный base_url в конструкторе клиента. Дальше весь остальной код, включая стриминг и обработку ошибок, работает без изменений.

Установка

pip install anthropic openai

Версии SDK значения не имеют, если они не старше пары лет — форматы /v1/messages и /v1/chat/completions, которые они дергают, стабильны.

Клиент Anthropic SDK

import anthropic

client = anthropic.Anthropic(
    base_url="https://api.zukko.pro",
    api_key="fetch_ваш_ключ",
)

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Объясни разницу между list и tuple в Python"}],
)
print(message.content[0].text)

Обратите внимание: base_url здесь без /v1 — Anthropic SDK сам добавляет путь /v1/messages при вызове .messages.create(). Если приписать /v1 вручную, каждый запрос уйдёт на /v1/v1/messages и вернёт 404.

Клиент OpenAI SDK

from openai import OpenAI

client = OpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key="fetch_ваш_ключ",
)

response = client.chat.completions.create(
    model="gpt-5.6-sol",
    messages=[{"role": "user", "content": "Объясни разницу между list и tuple в Python"}],
)
print(response.choices[0].message.content)

Здесь наоборот: base_url заканчивается на /v1, потому что OpenAI SDK путь сам не добавляет, а подставляет chat/completions к уже готовому базовому адресу.

Стриминг

Anthropic SDK:

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Напиши короткую историю про кота-программиста"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

OpenAI SDK:

stream = client.chat.completions.create(
    model="gpt-5.6-sol",
    messages=[{"role": "user", "content": "Напиши короткую историю про кота-программиста"}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Оба варианта передают чанки по SSE так же, как с официальными серверами — шлюз не меняет протокол, только маршрутизирует запрос к нужному провайдеру модели.

Vision: изображение в запросе

Изображения принимают модели Claude и GPT из каталога (у GLM vision нет). Через Anthropic SDK:

import base64

with open("screenshot.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

message = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image_data}},
            {"type": "text", "text": "Что не так на этом скриншоте?"},
        ],
    }],
)
print(message.content[0].text)

Подсчёт токенов без затрат

/v1/messages/count_tokens не выполняет сам запрос и ничего не списывает — удобно, чтобы прикинуть стоимость до вызова:

count = client.messages.count_tokens(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "Длинный текст для анализа..."}],
)
print(count.input_tokens)

Прямого метода в OpenAI SDK под /v1/responses для подсчёта нет — если нужно то же самое для GPT-моделей, дергайте эндпоинт напрямую через client._client.post() либо через requests.

Обработка ошибок

import anthropic

try:
    client.messages.create(
        model="claude-sonnet-5",
        max_tokens=100,
        messages=[{"role": "user", "content": "тест"}],
    )
except anthropic.AuthenticationError:
    print("Неверный ключ — проверьте Bearer fetch_...")
except anthropic.APIStatusError as e:
    if e.status_code == 402:
        print("Баланс пуст:", e.response.json())
    elif e.status_code == 429:
        print("Лимит ключа (key_daily/monthly_limit_reached) или 429 от провайдера")
    else:
        raise

Для openai SDK структура похожая: openai.AuthenticationError, openai.RateLimitError, а произвольный код (в частности 402, которого в оригинальном OpenAI API не существует) ловится через openai.APIStatusError и e.status_code.

Idempotency-Key на уровне SDK

Оба SDK поддерживают дополнительные заголовки через extra_headers:

client.messages.create(
    model="claude-sonnet-5",
    max_tokens=100,
    messages=[{"role": "user", "content": "тест"}],
    extra_headers={"Idempotency-Key": "job-2024-report-part1"},
)

Повтор с тем же значением ключа и тем же телом запроса за 24 часа возвращает сохранённый ответ без повторного списания. Ключ задаётся именно в конкретном вызове — уникальный на каждый логический запрос и одинаковый при его повторах. Не кладите его в default_headers клиента: тогда второй же запрос с другим телом получит 422 idempotency_conflict.

Асинхронный клиент

Обе библиотеки поставляют async-версии для приложений на asyncio — интерфейс идентичен синхронному, меняется только префикс класса и необходимость await:

import asyncio
from anthropic import AsyncAnthropic

async def main():
    client = AsyncAnthropic(
        base_url="https://api.zukko.pro",
        api_key="fetch_ваш_ключ",
    )
    message = await client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Кратко объясни GIL в Python"}],
    )
    print(message.content[0].text)

asyncio.run(main())

Для openai SDK то же самое — AsyncOpenAI вместо OpenAI. Асинхронный клиент особенно полезен при параллельной обработке батча запросов: asyncio.gather() с несколькими вызовами create() отправляет их одновременно, а не по очереди, что заметно ускоряет обработку, например, ста строк из CSV на классификацию.

Ретраи и таймауты

Оба SDK ретраят по умолчанию сетевые ошибки, 429 и 5xx-ответы (в том числе 503 provider_unavailable), но не ретраят 401 и 402 — их нужно обрабатывать в коде явно, потому что повторный запрос с тем же ключом и пустым балансом просто вернёт ту же ошибку. Отдельный случай — 503 model_disabled: модель выключена, и повторы тут не помогут, нужно сменить модель (актуальный список — GET https://api.zukko.pro/v1/models):

client = anthropic.Anthropic(
    base_url="https://api.zukko.pro",
    api_key="fetch_ваш_ключ",
    max_retries=2,
    timeout=60.0,
)

Увеличивать max_retries для 429 стоит с осторожностью — своего rate limit у шлюза нет, и 429 от Zukko означает дневной или месячный лимит конкретного ключа (key_daily_limit_reached — сброс в 00:00 UTC, key_monthly_limit_reached). Ретраи в течение часа тут ничего не изменят, и правильнее поймать код ошибки и остановиться; повторять имеет смысл только 429 от провайдера.

Заголовки стоимости в ответе

Чтобы прочитать X-Zukko-Cost-Micros и X-Fetch-Balance-Cents, нужен доступ к сырым HTTP-заголовкам, а не только к распарсенному объекту:

response = client.messages.with_raw_response.create(
    model="claude-sonnet-5",
    max_tokens=100,
    messages=[{"role": "user", "content": "тест"}],
)
print(response.headers.get("x-zukko-cost-micros"))
message = response.parse()

В openai SDK то же самое: client.chat.completions.with_raw_response.create(...). Учтите две вещи: эти заголовки есть только у обычных (нестриминговых) ответов — у stream=True их нет, и расход для таких запросов смотрят в кабинете, GET /v1/usage или GET /v1/balance; а X-Fetch-Balance-Cents, несмотря на название, — в микродолларах (1 000 000 = $1), точный баланс в центах — поле balance_cents из GET /v1/balance.