Python SDK

The Pendra Python SDK is a drop-in OpenAI-compatible client for sovereign UK inference. Sync and async, with streaming. Python 3.10+.

Installation

$ pip install pendra

Requires Python 3.10 or later. View on PyPI.

Quick start

from pendra import Pendra

client = Pendra(
    api_key="pdr_sk_...",   # or set PENDRA_API_KEY env var
)

response = client.chat.completions.create(
    model="qwen3.6:27b",
    messages=[{
        "role": "user",
        "content": "What is the capital of the UK?"
    }]
)

print(response.choices[0].message.content)

Streaming

Stream responses token by token using a context manager.

with client.chat.completions.create(
    model="qwen3.6:27b",
    messages=[{"role": "user", "content": "Write a poem about London."}],
    stream=True,
) as stream:
    for chunk in stream:
        if not chunk.choices:  # usage / timing chunk — no text
            continue
        print(chunk.choices[0].delta.content or "", end="", flush=True)

Async client

Use AsyncPendra for asyncio applications. Supports both streaming and non-streaming.

import asyncio
from pendra import AsyncPendra

async def main():
    async with AsyncPendra(api_key="pdr_sk_...") as client:
        response = await client.chat.completions.create(
            model="qwen3.6:27b",
            messages=[{"role": "user", "content": "Hello!"}]
        )
        print(response.choices[0].message.content)

asyncio.run(main())

Image generation

Generate images from a text prompt. Returns base64-encoded PNGs by default — decode and save to disk, or set response_format="url" when supported by the model.

import base64

response = client.images.generations.create(
    model="flux.1-schnell",
    prompt="A red London double-decker bus at sunset",
)

with open("bus.png", "wb") as f:
    f.write(base64.b64decode(response.data[0].b64_json))

Use AsyncPendra for async applications:

async with AsyncPendra(api_key="pdr_sk_...") as client:
    response = await client.images.generations.create(
        model="sdxl-turbo",
        prompt="A red London double-decker bus at sunset",
    )

Image generation is non-streaming — the endpoint returns a single JSON response once the worker finishes.

Embeddings

Generate vector embeddings for retrieval, search, and RAG pipelines. OpenAI-compatible — pass a string or a list of strings and get back a CreateEmbeddingResponse with one embedding per input.

response = client.embeddings.create(
    model="nomic-embed-text",
    input=["The quick brown fox", "jumps over the lazy dog"],
)

for item in response.data:
    print(item.index, len(item.embedding), "dims")

print(response.usage.prompt_tokens)

Any embedding model in the Pendra catalogue works — nomic-embed-text, qwen3-embedding, bge-m3, embeddinggemma. Also available on AsyncPendra via await client.embeddings.create(...).

Audio transcription

Transcribe audio to text using Whisper-class models. Multipart upload — pass an open binary file, a path, or a (filename, bytes) tuple. Files capped at 25 MB.

with open("meeting.mp3", "rb") as f:
    result = client.audio.transcriptions.create(
        file=f,
        model="whisper-large-v3-turbo",
        language="en",
    )

print(result.text)

Use AsyncPendra for async applications:

async with AsyncPendra(api_key="pdr_sk_...") as client:
    with open("meeting.mp3", "rb") as f:
        result = await client.audio.transcriptions.create(
            file=f,
            model="whisper-large-v3-turbo",
        )

print(result.text)

Need subtitles? Pass response_format="srt" (or "vtt") and the call returns a subtitle string directly — write it to disk and you have a ready-to-use caption track:

# Get WebVTT/SRT subtitles instead of plain JSON
with open("meeting.mp3", "rb") as f:
    srt = client.audio.transcriptions.create(
        file=f,
        model="whisper-large-v3-turbo",
        response_format="srt",
    )

with open("meeting.srt", "w") as f:
    f.write(srt.text)  # the SDK wraps text/srt/vtt bodies as { text: ... }

Transcription is non-streaming — the endpoint returns a single JSON response once the worker finishes. result.duration and result.language are populated when the model supplies them; result.segments appears when response_format="verbose_json".

Text to speech

Turn text into speech with a speech model. You get a WAV file back (24 kHz, 16-bit mono); leave out voice to use the model's default.

speech = client.audio.speech.create(
    model="your-speech-model",  # an id from /models?type=speech
    input="Good morning, and welcome to today's briefing.",
)
speech.write_to_file("speech.wav")   # or speech.content for the raw bytes

Speech isn't available with private inference yet, so the call raises an error instead of sending your text unencrypted when private inference is on.

Content filter verdicts

When your API key has content filtering turned on, a reply the filter flagged, blocked or couldn't check says so on response.pendra.moderation (chat replies also have the response.moderation shortcut): the stage that decided it ("request" or "response"), the action, the categories and the moderation model. A stream carries it on the chunk that ends the answer, so read stream.moderation once you've iterated to the end. Both are None when there's nothing to report.

response = client.chat.completions.create(
    model="qwen3.6:27b",
    messages=[{"role": "user", "content": "Hello!"}],
)
if response.moderation and response.moderation.action == "flagged":
    print("flagged:", response.moderation.categories)

Which worker served a request

Every reply names the worker that ran it on response.pendra.worker: its id (the same one worker_id= pins a request to), the name you gave it in the console, and the worker version that served it (either can be None). response.pendra.request_id is the request's ID, worth logging alongside it and quoting to support.

response = client.chat.completions.create(
    model="qwen3.6:27b",
    messages=[{"role": "user", "content": "Hello!"}],
)
if response.pendra and response.pendra.worker:
    worker = response.pendra.worker
    print(worker.id, worker.name, worker.version)
    print(response.pendra.request_id)

A stream names the worker on its last chunk, so read stream.worker and stream.request_id once you've iterated to the end. Both are None until then, and the async client's streams have the same two properties.

with client.chat.completions.create(
    model="qwen3.6:27b",
    messages=[{"role": "user", "content": "Write a poem about London."}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices:
            print(chunk.choices[0].delta.content or "", end="", flush=True)

print(stream.worker.id if stream.worker else None, stream.request_id)

Embeddings, reranking, image generation, transcription and speech results carry the same pendra field, and it works the same with private inference turned on. Needs pendra 0.17 or later.

List models

Query available models from your Pendra instance.

models = client.models.list()

for model in models:
    print(model.id)

Migrating from OpenAI

The Pendra SDK mirrors the OpenAI interface. Two lines to switch — your existing code just works.

# Before
from openai import OpenAI
client = OpenAI(api_key="sk-...")

# After
from pendra import Pendra
client = Pendra(api_key="pdr_sk_...")

API reference

client.chat.completions.create()

Create a chat completion. Returns ChatCompletion or Stream.

Any other keyword argument is sent to the API as-is, so a Chat Completions field newer than your SDK version still reaches the model. To add an HTTP header to one request, pass extra_headers=, which every create() call accepts. On the way back, a reasoning model's working arrives on message.reasoning_content (mirrored on message.reasoning, and streamed as delta.reasoning_content), separate from the answer in message.content.

Replies also carry a Pendra-specific pendra field alongside choices, with response.notice as a shortcut to the part you most often want. Check it when an answer looks wrong for no obvious reason — a truncated_during_reasoning code is the model telling you it spent the whole max_tokens budget thinking and never reached an answer, which otherwise just looks like an empty message.content:

response = client.chat.completions.create(
    model="ornith-1.5:9b",
    messages=[{"role": "user", "content": "Which is heavier, 1kg of steel or 1kg of feathers?"}],
    max_tokens=200,
)

if response.notice and response.notice.code == "truncated_during_reasoning":
    # Retry with a bigger budget, or with thinking off.
    print(response.notice.message)
    print(response.usage.reasoning_tokens)  # where it went

The other codes include truncated_during_structured_output, reasoning_budget_reached, truncated_at_reasoning_budget, truncated_at_default_max_tokens, structured_output_not_json and strict_schema_not_enforced — see Notices for what each one means. The same field carries response.pendra.web_tool_steps when the serving worker has web tools enabled.

When streaming, the last few chunks report on the request rather than continuing the answer: one carries chunk.usage, one carries chunk.timing (ttft_ms, tokens_per_second, queue_wait_ms, worker_queue_wait_ms), and a notice arrives on its own chunk.notice too — so check it on every chunk, not just the last. Those chunks have no text, which is why the streaming example skips a chunk with no choices rather than indexing into it.

client.images.generations.create()

Generate images from a text prompt. Returns ImageResponse. Also available as await client.images.generations.create(...) on AsyncPendra.

client.embeddings.create()

Create embeddings. Returns CreateEmbeddingResponse. Also await client.embeddings.create(...) on AsyncPendra.

client.models.list()

Returns a list of Model objects available on the instance. Each model has id, object, created, and owned_by fields.

client.audio.transcriptions.create()

Transcribe an audio file. Returns TranscriptionResponse. Also available as await client.audio.transcriptions.create(...) on AsyncPendra.

client.audio.speech.create()

Returns a SpeechResponse: content (the WAV bytes), duration_ms, pendra, and the read() / write_to_file(path) helpers.

Environment variables

Error handling

All exceptions inherit from pendra.APIError.

err.message is always a readable sentence. Some errors carry more than a sentence — when your organisation requires end-to-end encryption and a request is sent without it, for example, the error also says why in a machine-readable type. The message stays readable, and the original error is on err.detail (the whole response body is on err.body), so your code can branch on it:

try:
    client.chat.completions.create(model="ornith-1.5:9b", messages=messages)
except pendra.APIStatusError as err:
    print(err.message)   # "This organisation requires end-to-end encrypted requests. …"
    if isinstance(err.detail, dict) and err.detail.get("error", {}).get("type") == "e2e_required":
        client = pendra.Pendra(private_inference=True)   # retry encrypted

Failures during a stream

A streaming reply has to send its status code before the model has generated anything, so if something goes wrong once generation is under way — the GPU worker crashing, the request timing out, your fleet being busy — it can't come back as an HTTP error code. The SDK raises APIStreamError from the iterator instead, at the point the failure happened. Anything the model already sent you is yours to keep:

import pendra

answer = ""
try:
    for chunk in client.chat.completions.create(
        model="ornith-1.5:9b",
        messages=[{"role": "user", "content": "Write me a limerick."}],
        stream=True,
    ):
        if chunk.choices and chunk.choices[0].delta.content:
            answer += chunk.choices[0].delta.content
except pendra.APIStreamError as err:
    # `answer` holds the partial reply that arrived before the failure.
    print(f"Stream failed: {err.message}")
    print(f"Quote this to support: {err.request_id}")

err.request_id and err.worker_id identify the exact inference that failed, so quote them if you report the problem. APIStreamError subclasses APIStatusError, so code that already catches that keeps working. If none of your workers are online when you start a stream, you get a plain APIStatusError with a 503 before any tokens arrive, not an APIStreamError.

err.status_code helps you decide whether retrying is worth it — but check it for None first. Some failures, including a timeout, arrive without a status code at all:

except pendra.APIStreamError as err:
    if err.status_code in (429, 503):
        retry_after_backoff()      # fleet busy, or its worker went offline
    elif err.status_code is None:
        # Timed out, or a failure the server couldn't classify. Retrying once
        # is usually reasonable; read err.message to see what happened.
        print(err.message)
    else:
        raise                      # e.g. 502 — a worker failed, retrying won't help

Note that a stream reports being at capacity this way rather than as a RateLimitError — by the time the queue wait is over, the 200 has already been sent — so back off on the status code, not on the exception class.

The same applies to a non-streaming request that runs for more than about 80 seconds — a long embeddings batch, an image, a long JSON-format transcription. Pendra sends the 200 early to keep the connection open, so a failure after that point also arrives as APIStreamError, raised by the create call itself, with the same fields. You never get an empty result in its place.

Replies that don't parse

APIResponseValidationError means the reply reached you but couldn't be read as a chat completion — usually a proxy between you and Pendra returning its own error page. It is deliberately not an APIConnectionError: retrying won't help, because the same body will fail to parse again. If you retry on connection failures, let this one through. The offending body is quoted in the message (and in full on err.body) so you can see what actually came back.

Links

• PyPI
• Get an API key
• API reference