Python SDK Guide: DeepSeek, GLM and Kimi in Python

Calling DeepSeek, GLM, Kimi or Qwen from Python uses the official openai package — there is no separate SDK to learn. Install it, set base_url, and every method you already know works. This guide covers installation, sync and async clients, streaming, structured output, tool calling, retries, cost tracking and the testing patterns that keep an LLM-backed service maintainable.

Installation

pip install openai
# or, with uv
uv add openai

That is the only dependency required. Version 1.x of the SDK is what the examples below assume; if you are on 0.x, the client construction differs and upgrading is worth doing first.

Step 1: Construct the client

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url="https://aiwave.live/v1",
)

resp = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "Explain MoE routing in two sentences."}],
)
print(resp.choices[0].message.content)
print(resp.usage)          # exact token counts

Create a key in the console — email or GitHub signup, no Chinese phone number, billing in USD. The endpoint reference is in the API documentation and key handling in authentication.

The SDK also reads OPENAI_BASE_URL and OPENAI_API_KEY from the environment, so OpenAI() with no arguments works if you prefer configuration over code.

Step 2: Choose a model

Current rates, read from the live pricing endpoint:

Model ID	Input / 1M tokens	Output / 1M tokens
deepseek-v4-flash	$0.206	$0.412
deepseek-v4-pro	$1.09	$2.17
glm-4.7-flash	free	free
glm-5	$1.55	$4.96
kimi-k2.5	$0.66	$3.30
qwen3-coder-480b-a35b-instruct	$0.12	$0.36
ernie-4.0-turbo-8k	$0.0012	$0.0012

Rates read from the AIWave pricing endpoint on 2026-07-26. Check live pricing before budgeting — providers revise rates.

Do not hard-code a model string and forget about it. Model IDs are versioned; read the served list at startup so a retired ID fails loudly at boot rather than on a user request:

MODEL = "deepseek-v4-pro"
available = {m.id for m in client.models.list().data}
if MODEL not in available:
    raise SystemExit(f"{MODEL} is no longer served; available: {sorted(available)[:10]}")

Step 3: Stream responses

stream = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Write a haiku about cold starts."}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Streaming responses do not carry a usage object per chunk. If you need exact accounting, measure with a sampled non-streaming call rather than estimating. The wire format is documented in streaming responses.

Step 4: Use the async client for concurrency

Anything processing more than a handful of requests should be async. The API is identical apart from await:

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url="https://aiwave.live/v1",
)

sem = asyncio.Semaphore(8)          # bound concurrency, not just retries

async def ask(prompt: str) -> str:
    async with sem:
        r = await client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main(prompts):
    return await asyncio.gather(*(ask(p) for p in prompts))

print(asyncio.run(main(["Summarise X", "Summarise Y", "Summarise Z"])))

The semaphore matters more than most people expect. Firing a thousand coroutines at once produces 429 responses that retries then amplify. Bounding concurrency prevents the problem instead of recovering from it — see rate limits.

Step 5: Structured output

When the response feeds code rather than a human, constrain it structurally and validate the result. Pydantic makes the validation honest:

import json
from pydantic import BaseModel, ValidationError

class Ticket(BaseModel):
    severity: str
    component: str
    summary: str

resp = client.chat.completions.create(
    model="deepseek-v4-pro",
    temperature=0,
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content":
         'Return JSON only: {"severity": "low|medium|high|critical", '
         '"component": string, "summary": string}'},
        {"role": "user", "content": "Checkout 500s since the 14:20 deploy."},
    ],
)

try:
    ticket = Ticket.model_validate_json(resp.choices[0].message.content)
except ValidationError as e:
    ticket = None
    print("model returned unusable output:", e)

Three details make this reliable. Set temperature=0 — sampling variety has no upside in extraction. State the schema in the prompt as well as via response_format, because JSON mode is an upstream capability that varies by model. And always validate; a parser that assumes success will fail in production on the one response that came back wrapped in prose.

Step 6: Tool calling

import json

tools = [{
    "type": "function",
    "function": {
        "name": "get_order_status",
        "description": "Look up the status of an order by ID",
        "parameters": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
    },
}]

messages = [{"role": "user", "content": "Where is order A-4471?"}]
resp = client.chat.completions.create(
    model="deepseek-v4-pro", messages=messages, tools=tools)

msg = resp.choices[0].message
if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = lookup_order(args["order_id"])          # your code
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    final = client.chat.completions.create(
        model="deepseek-v4-pro", messages=messages, tools=tools)
    print(final.choices[0].message.content)

Tool support is a property of the upstream model rather than the gateway — verify on the model directory before depending on it. The protocol in more depth is in function calling with Chinese AI models.

Step 7: Retries, timeouts and fallback

The SDK retries some failures itself, but it does not know which model you would rather fall back to. Wrap it:

import time, random
from openai import OpenAI, APIStatusError, APITimeoutError

client = OpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url="https://aiwave.live/v1",
    timeout=60.0,        # never rely on an unbounded default
    max_retries=0,       # we control the policy below
)

RETRYABLE = {408, 429, 500, 502, 503}

def chat(messages, model="deepseek-v4-pro",
         fallback="deepseek-v4-flash", attempts=5, **kw):
    for i in range(attempts):
        try:
            return client.chat.completions.create(
                model=model, messages=messages, **kw)
        except APITimeoutError:
            pass
        except APIStatusError as e:
            if e.status_code not in RETRYABLE:
                raise                       # permanent: fix the request
            if i == attempts - 2 and fallback:
                model = fallback            # last attempt on another model
        time.sleep(min(32, 2 ** i) * (0.5 + random.random()))
    raise RuntimeError("all retries exhausted")

Retrying a 400 or 401 forever is a common and expensive bug. The decision table for every status code is in error codes.

Step 8: Track cost from day one

import logging

PRICES = {                                   # USD per 1M tokens
    "deepseek-v4-flash": (0.206, 0.412),     # fetch these from /pricing
    "deepseek-v4-pro":  (1.0875, 2.175),
}

def priced_chat(messages, model="deepseek-v4-flash", **kw):
    r = client.chat.completions.create(model=model, messages=messages, **kw)
    pin, pout = PRICES.get(model, (0, 0))
    cost = (r.usage.prompt_tokens / 1e6) * pin + (r.usage.completion_tokens / 1e6) * pout
    logging.info("model=%s in=%d out=%d cost=$%.6f",
                 model, r.usage.prompt_tokens, r.usage.completion_tokens, cost)
    return r

Keep the price table in configuration rather than code, and refresh it from the pricing page — providers revise rates. Logging per-request cost from the start turns a monthly invoice surprise into a metric you can alert on.

Testing code that calls a model

LLM calls make tests slow, expensive and flaky if you let them. The fix is to isolate the boundary.

from unittest.mock import Mock, patch

def test_summarise_extracts_text():
    fake = Mock()
    fake.choices = [Mock(message=Mock(content="a summary"))]
    fake.usage = Mock(prompt_tokens=10, completion_tokens=3)

    with patch.object(client.chat.completions, "create", return_value=fake):
        assert summarise("long text") == "a summary"

Mock for unit tests, which should cover your parsing, retry and error paths without a network call. Keep a small number of integration tests that do hit the API, run them on the cheapest model, and pin temperature=0 so output is as stable as the model allows. There is no reason for CI to spend premium rates verifying that your plumbing works.

Managing conversation history

The API is stateless: every request carries the full conversation. That is simple to reason about and easy to make expensive, because history is resent as input tokens on every turn.

class Conversation:
    def __init__(self, client, model, system=None, budget_chars=12_000):
        self.client, self.model = client, model
        self.budget = budget_chars
        self.messages = [{"role": "system", "content": system}] if system else []

    def _trim(self):
        system = [m for m in self.messages if m["role"] == "system"]
        rest = [m for m in self.messages if m["role"] != "system"]
        used, kept = 0, []
        for m in reversed(rest):                 # keep the most recent turns
            size = len(m.get("content") or "")
            if used + size > self.budget and kept:
                break
            used += size
            kept.insert(0, m)
        self.messages = system + kept

    def ask(self, text, **kw):
        self.messages.append({"role": "user", "content": text})
        self._trim()
        r = self.client.chat.completions.create(
            model=self.model, messages=self.messages, **kw)
        reply = r.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply, r.usage

Trim by size rather than message count — ten short exchanges may cost less than one pasted stack trace. Always preserve the system message; dropping it mid-conversation produces a model that suddenly forgets its instructions, which is a confusing bug to diagnose.

For long-running assistants, summarising older turns beats discarding them. Run the summary on a cheap model: condensing history is exactly the kind of undemanding work that does not need a premium model.

Caching: the optimisation nobody regrets

In most production workloads a meaningful share of requests are exact repeats. A cache in front of the API removes them entirely.

import hashlib, json, sqlite3

db = sqlite3.connect("llm_cache.db")
db.execute("CREATE TABLE IF NOT EXISTS cache (k TEXT PRIMARY KEY, v TEXT)")

def key_for(model, messages, **kw):
    blob = json.dumps({"model": model, "messages": messages, **kw}, sort_keys=True)
    return hashlib.sha256(blob.encode()).hexdigest()

def cached_chat(messages, model="deepseek-v4-flash", **kw):
    k = key_for(model, messages, **kw)
    row = db.execute("SELECT v FROM cache WHERE k=?", (k,)).fetchone()
    if row:
        return row[0]
    r = client.chat.completions.create(model=model, messages=messages, **kw)
    text = r.choices[0].message.content
    db.execute("INSERT OR REPLACE INTO cache VALUES (?,?)", (k, text))
    db.commit()
    return text

Two caveats worth stating. Only cache deterministic calls — include temperature in the key, and do not cache anything above temperature 0 unless variability is acceptable. And include the model in the key, otherwise switching models silently serves stale answers from the previous one.

Batching to cut round trips

When you have many small independent items, one request handling twenty of them costs far less than twenty requests: the system prompt is sent once and per-request overhead disappears.

def classify_batch(items, model="deepseek-v4-flash"):
    numbered = "\n".join(f"[{i}] {t}" for i, t in enumerate(items))
    r = client.chat.completions.create(
        model=model,
        temperature=0,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content":
             'Return {"results": [{"index": int, "label": string}]} with one '
             'entry per input. Labels: billing, technical, sales, spam.'},
            {"role": "user", "content": numbered},
        ],
    )
    data = json.loads(r.choices[0].message.content)
    out = {d["index"]: d["label"] for d in data["results"]}
    return [out.get(i, "unknown") for i in range(len(items))]   # never silently drop

Keep batches modest — ten to fifty items — and always reconcile the returned indices against what you sent. A batch that comes back short should fall back to per-item processing rather than losing records. The dedicated treatment is batch processing with Chinese AI models.

Multi-model routing in a few lines

Because the model is a parameter, routing by difficulty is trivial to implement and usually the single largest saving available.

CHEAP, STRONG = "deepseek-v4-flash", "deepseek-v4-pro"

def smart_chat(messages, hard=False, **kw):
    model = STRONG if hard else CHEAP
    r = client.chat.completions.create(model=model, messages=messages, **kw)
    return r

def with_escalation(messages, validate, **kw):
    """Try cheap first; escalate only if the result fails your own check."""
    r = smart_chat(messages, hard=False, **kw)
    text = r.choices[0].message.content
    if validate(text):
        return text
    r = smart_chat(messages, hard=True, **kw)
    return r.choices[0].message.content

The validate callback is the important part: schema validation, a required substring, a length check — whatever “good enough” means for that call site. Escalating on failure rather than in advance means you pay premium rates only for the requests that genuinely need them. The full pattern is in building a multi-model router.

Using it in a web service

A FastAPI endpoint that streams shows most of the pieces together:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(api_key=os.environ["AIWAVE_API_KEY"],
                     base_url="https://aiwave.live/v1")

@app.post("/chat")
async def chat(payload: dict):
    async def gen():
        stream = await client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=payload["messages"],
            stream=True,
            max_tokens=1024,               # a cost ceiling, always set one
        )
        async for chunk in stream:
            piece = chunk.choices[0].delta.content
            if piece:
                yield f"data: {piece}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(gen(), media_type="text/event-stream")

Three production notes. Set max_tokens — an unbounded generation is an unbounded bill. Rate-limit the endpoint, because it spends money on every call and an unprotected route is an open invitation. And if chunks arrive all at once in production but stream fine locally, a proxy in front of your app is buffering; disable buffering there rather than changing the code.

Project layout that stays maintainable

The pattern that ages well is keeping all model access behind one module, so switching models, adding caching or changing retry policy is one file rather than a search across the codebase.

myapp/
  llm/
    __init__.py      # exports: chat(), stream(), embed()
    client.py        # client construction, timeouts, base_url
    retry.py         # backoff policy and fallback model
    cache.py
    prompts/         # prompt templates, in version control
      classify.txt
      summarise.txt
  services/
    tickets.py       # imports from myapp.llm, never constructs a client

Keep prompts as files rather than string literals scattered through business logic. A quality regression then shows up in git log, which makes it bisectable — and prompt changes cause quality regressions far more often than code changes do.

Production checklist

• Explicit timeout on the client; never inherit an unbounded default.
• Own the retry policy so you control which statuses retry and what the fallback is.
• Bound concurrency with a semaphore rather than relying on retries to absorb bursts.
• Set max_tokens on every generation.
• Log usage per request and alert on daily spend.
• Validate the model ID at startup against models.list().
• Keep the key in the environment, never in code or committed config.
• Pin temperature=0 for anything parsed downstream.

Frequently asked questions

Is there a DeepSeek Python SDK?

You do not need one. The official openai package works against any OpenAI-compatible endpoint — set base_url and use the model ID.

Can I use async and sync clients together?

Yes. OpenAI and AsyncOpenAI are independent and can coexist in one application.

How do I count tokens before sending?

Tokenisers differ between model families, so a count from one family is approximate for another. The authoritative number is the usage field on the response; log it rather than estimating.

Do I need a Chinese phone number?

No. Email or GitHub signup, USD billing. See accessing Chinese AI models without a Chinese phone number.

Sources

Every claim above traces back to one of these. Pricing figures come from the AIWave pricing endpoint on 2026-07-26; capability claims come from the vendors’ own documentation.

• openai-python SDK on GitHub
• OpenAI Chat Completions API reference
• Pydantic documentation
• AIWave API documentation

Related Articles

• Migrate from OpenAI in One Line
• Streaming LLM Responses with Chinese Models
• Function Calling with Chinese AI Models
• LangChain with Chinese AI Models
• Access Chinese AI Models Without a Chinese Phone Number