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".

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. 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 are truncated_during_structured_output 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.

Environment variables

Error handling

All exceptions inherit from pendra.APIError.

Links

• PyPI
• Get an API key
• API reference