pendra-python

Official Python SDK for Pendra — UK-based, privacy-first LLM inference.

Your data is processed in the UK, never stored, never shared with US cloud providers.

Installation

pip install pendra

Quick Start

import pendra

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

response = client.chat.completions.create(
    model="qwen3.5:0.8b",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of the UK?"},
    ],
)

print(response.choices[0].message.content)
# → London is the capital of the United Kingdom.

Your first request — full sequence

Pendra serves inference from workers you (or your org) run, so a brand-new account needs three things in place before that chat.completions.create() call returns a 200:

1. A worker connected. Install Pendra on any host with a GPU or CPU and run pendra setup. The wizard walks through pasting a worker key from console.pendra.ai/workers and connecting to the API.
2. A model on disk. The worker only serves models it has locally. From the worker host run, e.g., pendra models install qwen3.5:0.8b. Browse console.pendra.ai/models for the full catalogue.
3. An API key. Create one at console.pendra.ai/api-keys and pass it as api_key= above.

If your call returns 404 Model 'X' is in the catalogue but no connected worker has it installed yet, skip back to step 2 — that's the API telling you the model is known but hasn't been pulled onto a worker yet.

Streaming

with client.chat.completions.create(
    model="qwen3.5:0.8b",
    messages=[{"role": "user", "content": "Write me a short poem about London."}],
    stream=True,
) as stream:
    for chunk in stream:
        # The last chunks of a stream carry token usage and per-request timings
        # instead of text, with no choices at all — so check before indexing.
        if not chunk.choices:
            continue
        print(chunk.choices[0].delta.content or "", end="", flush=True)

Async

import asyncio
import pendra

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

        # Streaming
        stream = await client.chat.completions.create(
            model="qwen3.5:0.8b",
            messages=[{"role": "user", "content": "Count to 5"}],
            stream=True,
        )
        async for chunk in stream:
            if not chunk.choices:  # usage / timing chunk — no text
                continue
            print(chunk.choices[0].delta.content or "", end="", flush=True)

asyncio.run(main())

Notices, timings and other response extras

Replies 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 — truncated_during_reasoning is the model saying it spent the whole max_tokens budget thinking and never reached an answer, which otherwise just looks like an empty message.content:

if response.notice and response.notice.code == "truncated_during_reasoning":
    print(response.notice.message)
    print(response.usage.reasoning_tokens)  # where the budget went
    # Retry with a bigger max_tokens, or with enable_thinking=False.

The other codes are truncated_during_structured_output and strict_schema_not_enforced. The same field carries response.pendra.web_tool_steps when the serving worker has web tools enabled.

The chunks with no choices that the streaming loop above skips are where a stream reports on the request rather than continuing the answer — so read them instead of only skipping them:

for chunk in stream:
    if chunk.notice:
        print(chunk.notice.message)
    if chunk.timing:
        print(f"{chunk.timing.tokens_per_second} tok/s, {chunk.timing.ttft_ms}ms to first token")
    if chunk.usage:
        print(f"{chunk.usage.total_tokens} tokens")
    if not chunk.choices:
        continue
    print(chunk.choices[0].delta.content or "", end="", flush=True)

Which worker served a request

Every response says which of your workers handled it, under pendra.worker. id is the worker's id as shown in the console (the same one worker_id= pins a request to), name is the name you gave it, and version is the worker release that served the request (either can be None). pendra.request_id is Pendra's id for the request, which is worth logging alongside it:

response = client.chat.completions.create(
    model="qwen3.5:0.8b",
    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 have iterated to the end. Both are None until then:

with client.chat.completions.create(
    model="qwen3.5:0.8b",
    messages=[{"role": "user", "content": "Count to 5"}],
    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 "unknown worker", stream.request_id)

The async stream has the same worker and request_id properties.

Embeddings, rerank, image generation, transcription and speech responses carry the same pendra field:

embeddings = client.embeddings.create(model="nomic-embed-text:latest", input="Hello")
print(embeddings.pendra.worker.id if embeddings.pendra else None)

It works with private inference too: the worker is named in the clear beside the encrypted reply, never inside it.

List Models

models = client.models.list()
for model in models:
    print(model.id)

Image Generation

Generate images from a text prompt. Returns base64-encoded PNGs by default.

import base64

response = client.images.generations.create(
    model="x/z-image-turbo",
    prompt="A red London double-decker bus at sunset",
    size="1024x1024",
)

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

Async usage mirrors the sync API:

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

Image generation is non-streaming — the response is returned as a single JSON payload once the worker finishes.

Text to Speech

Turn text into spoken audio. The response holds a WAV file you can save or read as bytes:

speech = client.audio.speech.create(
    model="qwen3-tts",
    input="Good morning from London.",
    voice="Serena",  # optional: omit it to use the model's default voice
)

speech.write_to_file("speech.wav")  # or speech.read() for the raw bytes
print(speech.pendra.worker.id if speech.pendra else None)

speech.duration_ms is the length of the audio, when the worker reports it. Pass worker_id= to send the request to a particular worker. The async client has the same await client.audio.speech.create(...).

Speech isn't available with private inference yet. On a client with private inference turned on, create raises an error before sending anything, so your text is never sent unencrypted.

Environment Variables

OpenAI Compatibility

The Pendra SDK is fully compatible with the OpenAI Python SDK interface. To migrate:

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

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

The client.chat.completions.create() interface is identical.

Self-Hosted Workers

Run inference on your own GPUs with a single command. Your prompts and completions never leave your infrastructure.

curl -fsSL https://get.pendra.ai/worker | bash

See the Workers documentation for full setup instructions.

Licence

Apache-2.0