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)

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.

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