Docs

Python SDK

alienplatform is the Python package for code that runs in a deployment: async access to linked resources, plus queue handlers for Python Workers.

app.py
import asyncio
from alienplatform import kv, storage

async def main() -> None:
    await storage("files").put("reports/2026-10.json", b"{}")
    await kv("cache").put("status", b"ready", ttl_seconds=60)

asyncio.run(main())

This page documents version 3.3.41. The package wraps the same Rust core as the TypeScript SDK, so bindings behave the same on every platform.

Install

pip install alienplatform

Requires Python 3.10 or newer. Install the worker extra for a Python Worker, which adds the gRPC runtime that receives events:

pip install "alienplatform[worker]"

Build Python workloads with the python toolchain in alien.ts ({ type: "python", command: [...] }). See Worker and Container.

Bindings

Call the factory with the resource ID from alien.ts. Every method is async. Factories do no I/O; the first call reads the binding Alien injects for the link.

FactoryMethods
storage(name)get(path), get_optional(path), put(path, data, *, condition=None), delete(path), list(prefix=None), signed_url(method, path, expires_in)
kv(name)get(key), put(key, value, *, ttl_seconds=None, condition=None, version=None), delete(key, *, if_version=None), exists(key), scan(prefix, *, limit=None, cursor=None)
queue(name)send(message), send_text(message), send_batch(messages), send_batch_text(messages), receive(max_messages=1), ack(receipt_handle), ack_batch(receipts), renew_lease(receipt_handle, duration_seconds), nack(receipt_handle), purge()
vault(name)get(name), put(name, value), delete(name), list()
key(name)encrypt(plaintext, context=None), decrypt(ciphertext, context=None)
postgres(name)connection() returns a PostgresConnection with sqlalchemy_async_url, sqlalchemy_async_engine_kwargs(), and ssl_context
container(name)internal_url(), internal_urls(), public_url()
worker(name)invoke(*, method, path, headers=None, body=b"", timeout_ms=None), public_url()
ai(name)connection() returns an AiConnection with base_url, openai_base_url, and api_key
sandbox(name)create, get, get_or_create, list, run_command, start_job, poll_job, cancel_job, read_file, write_file, preview, update_egress, pause, resume, terminate, capabilities

worker(name) calls a linked Worker over HTTP. Use it to call one workload from another.

The following code connects SQLAlchemy and the OpenAI client to linked resources:

from openai import AsyncOpenAI
from sqlalchemy.ext.asyncio import create_async_engine
from alienplatform import ai, postgres

async def clients():
    db = await postgres("db").connection()
    engine = create_async_engine(db.sqlalchemy_async_url, **db.sqlalchemy_async_engine_kwargs())

    llm = await ai("llm").connection()
    client = AsyncOpenAI(base_url=llm.openai_base_url, api_key=llm.api_key or "unused")
    return engine, client

The AI connection points at an embedded gateway that signs requests with the workload's cloud identity, so you need no provider key.

Queue handlers

A Python Worker registers queue handlers, then starts the runtime with run_worker():

worker.py
import asyncio
import json
from alienplatform import QueueEvent, kv, on_queue_message, run_worker

async def handle(event: QueueEvent) -> None:
    job = json.loads(event.payload)
    await kv("cache").put(f"seen:{job['key']}", b"1")

on_queue_message("jobs", handle)
asyncio.run(run_worker())

on_queue_batch(source, handler) receives a QueueBatch and returns one QueueOutcome(delivery_id, disposition) per delivery, with disposition set to "success" or "retry".

Not yet in Python

These Worker features exist in the TypeScript and Rust SDKs but not in Python:

  • command() handlers. A Python workload can't receive commands today.
  • Storage event and cron handlers (onStorageEvent, onCronEvent).
  • wait_until() for background work after a response.

There is no Python client for the REST API. Call it over HTTPS, or use the TypeScript client.

Remote bindings

Bindings lets your backend use a resource in a customer's environment. Add the resource with { remoteAccess: true } in alien.ts.

import os
from alienplatform import Bindings

async def enqueue() -> None:
    bindings = await Bindings.for_remote_customer(
        project="acme",
        external_id="customer-1",
        token=os.environ["ALIEN_API_KEY"],
    )
    await bindings.queue("jobs").send({"type": "reindex"})
    await bindings.kv("cache").put("last-sync", b"2026-10-24")

Bindings.for_remote_deployment(deployment_id=..., token=...) targets one deployment. In Python, remote bindings cover kv and queue (send only).

Errors

Binding failures raise alienplatform.AlienError, which carries the error code and message from the Rust core.

Next

On this page