Full Python example: official SDKs through our gateway

The official anthropic and openai Python SDKs aren't hard-wired to their own companies' servers — both accept a custom base_url in the client constructor. Everything else, including streaming and error handling, keeps working unchanged.

Install

pip install anthropic openai

SDK version doesn't matter much as long as it isn't a couple of years old — the /v1/messages and /v1/chat/completions shapes they call are stable.

Anthropic SDK client

import anthropic

client = anthropic.Anthropic(
    base_url="https://api.zukko.pro",
    api_key="fetch_your_key",
)

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain the difference between list and tuple in Python"}],
)
print(message.content[0].text)

Note that base_url here has no /v1 — the Anthropic SDK appends /v1/messages itself when you call .messages.create(). Add /v1 manually and every request goes to /v1/v1/messages, returning 404.

OpenAI SDK client

from openai import OpenAI

client = OpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key="fetch_your_key",
)

response = client.chat.completions.create(
    model="gpt-5.6-sol",
    messages=[{"role": "user", "content": "Explain the difference between list and tuple in Python"}],
)
print(response.choices[0].message.content)

Here it's the opposite: base_url ends in /v1, because the OpenAI SDK doesn't add the path itself — it appends chat/completions to whatever base address you gave it.

Streaming

Anthropic SDK:

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a short story about a cat who codes"}],
) 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": "Write a short story about a cat who codes"}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Both stream chunks over SSE exactly as they would against the official servers — the gateway doesn't change the protocol, it just routes the request to whichever provider serves that model.

Vision: sending an image

The Claude and GPT models in the catalog accept images (GLM has no vision). Through the 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": "What's wrong in this screenshot?"},
        ],
    }],
)
print(message.content[0].text)

Counting tokens for free

/v1/messages/count_tokens doesn't run the request and doesn't charge anything — handy for estimating cost before a call:

count = client.messages.count_tokens(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "Long text to analyze..."}],
)
print(count.input_tokens)

The OpenAI SDK has no direct method for this under /v1/responses — if you need the same thing for GPT models, hit the endpoint directly via client._client.post() or plain requests.

Error handling

import anthropic

try:
    client.messages.create(
        model="claude-sonnet-5",
        max_tokens=100,
        messages=[{"role": "user", "content": "test"}],
    )
except anthropic.AuthenticationError:
    print("Invalid key — check Bearer fetch_...")
except anthropic.APIStatusError as e:
    if e.status_code == 402:
        print("Balance is empty:", e.response.json())
    elif e.status_code == 429:
        print("Key limit (key_daily/monthly_limit_reached) or a 429 from the provider")
    else:
        raise

The openai SDK follows the same pattern: openai.AuthenticationError, openai.RateLimitError, and any nonstandard code (402 in particular, which doesn't exist in the original OpenAI API) is caught via openai.APIStatusError and e.status_code.

Idempotency-Key at the SDK level

Both SDKs support extra headers through extra_headers:

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

A repeat with the same key value and the same request body within 24 hours returns the saved response without charging again. Set the key on the specific call — unique per logical request and identical across its retries. Don't put it in the client's default_headers: the very next request with a different body would get 422 idempotency_conflict.

Async client

Both libraries ship async versions for asyncio apps — the interface is identical to the sync one, just a different class prefix and an await:

import asyncio
from anthropic import AsyncAnthropic

async def main():
    client = AsyncAnthropic(
        base_url="https://api.zukko.pro",
        api_key="fetch_your_key",
    )
    message = await client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Briefly explain the GIL in Python"}],
    )
    print(message.content[0].text)

asyncio.run(main())

Same pattern for the openai SDK — AsyncOpenAI instead of OpenAI. The async client pays off especially when processing a batch in parallel: asyncio.gather() with several create() calls fires them concurrently instead of one at a time, which noticeably speeds up something like classifying a hundred rows from a CSV.

Retries and timeouts

Both SDKs retry network errors, 429s and 5xx responses by default (including 503 provider_unavailable), but they don't retry 401 or 402 — those need explicit handling in your code, since retrying with the same key and an empty balance just returns the same error again. 503 model_disabled is a special case: the model is switched off, retries won't help, and you need to change the model (the current list is at GET https://api.zukko.pro/v1/models):

client = anthropic.Anthropic(
    base_url="https://api.zukko.pro",
    api_key="fetch_your_key",
    max_retries=2,
    timeout=60.0,
)

Be careful bumping max_retries for 429s — the gateway has no rate limit of its own, and a 429 from Zukko means a key's daily or monthly cap (key_daily_limit_reached, reset at 00:00 UTC, or key_monthly_limit_reached). Retrying for the next hour changes nothing, and it's better to catch the error code and stop; only a 429 from the provider is worth retrying.

Reading cost headers on the response

To read X-Zukko-Cost-Micros and X-Fetch-Balance-Cents you need access to the raw HTTP headers, not just the parsed object:

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

Same pattern in the openai SDK: client.chat.completions.with_raw_response.create(...). Two caveats: these headers are only present on regular (non-streaming) responses — stream=True responses don't have them, and spend for those is checked in the dashboard, GET /v1/usage or GET /v1/balance; and X-Fetch-Balance-Cents, despite the name, is in micro-dollars (1,000,000 = $1) — the exact balance in cents is the balance_cents field from GET /v1/balance.