HOPSKIP
Start a run

SDKs

Python SDK

The Python SDK runs ordinary async/await workflow code on Pyodide, a prebuilt CPython for WebAssembly, hosted alongside the TypeScript runtime. Awaits are real quiescence boundaries, and the annotations you use (@durable, Searchable, @entity) reach the same workflow ABI host imports as every other launch language.

Install

pip install hopskip-worker hopskip-client

During the preview, build guests with sdk/py/pyodide-guest/build.sh from the dev shell. The first build fetches and hash-checks the pinned Pyodide distribution; after that it needs no network.

Starting

hop init ./greetings --lang python

This writes a project in the shape this SDK uses, already pointing at the guest SDK, and prints the exact hop build and hop run lines for what it wrote.

From there, point the build at the directory and every workflow under it is discovered (a workflow here being a .py with a top-level __hopskip_workflow =), each under a task type derived from its own name:

hop build --manifest-path ./my-workflows
hop dev --watch ./my-workflows

hop dev --watch rebuilds and redeploys what changed on every save and prints the hop run line for it.

A derived task type never contains the implementation language or the source layout. A task type is the contract a client calls, so porting a workflow to another language (or moving its directory) must not rename it. Pin one explicitly when you want a name that outlives the thing it was derived from:

# @hopskip task-type: hopskip:orders.process@2

# @hopskip skip excludes a file the rule would otherwise match, and hop workflows <dir> lists what discovery finds without building anything.

A workflow

The SDK’s names (durable, activity, sleep, …) are ambient in a workflow module - the build embeds the runtime prelude, so there is nothing to import:

def encode_i64(value):
    return value.to_bytes(8, "little", signed=True)

def decode_i64(data):
    return int.from_bytes(data, "little", signed=True)

async def order_processing():
    reserved = await activity("reserve_inventory", b"sku:ABC")
    charged = await activity("charge_card", b"amount:20")
    total = decode_i64(reserved) + decode_i64(charged)
    return encode_i64(total)

__hopskip_workflow = durable(order_processing)

Each compiled Python guest module exposes a single __hopskip_workflow export, so distinct workflows are separate entry-point modules.

A workflow module is self-contained: the build bundles it with the SDK prelude, and the interpreter’s own standard library is what it can import - pulling pip packages into a workflow is not supported yet. Activities are ordinary Python processes and install anything they like.

Typed payloads

The SDK ships no serialization format, by design - but it carries the one you choose. Supply a Codec (any object with content_type, encode, decode) and wrap the entry point with encoded(...): the input is decoded before your function sees it, the output encoded after, and the result’s content-type header declares the encoding without the workflow mentioning it:

class OrderCodec(Codec):
    content_type = "application/vnd.acme.order+json"
    def encode(self, value):
        return json.dumps(value).encode("utf-8")
    def decode(self, data):
        return json.loads(data.decode("utf-8"))

__hopskip_workflow = durable(encoded(OrderCodec(), ReceiptCodec(), process))

Pass None for either side to keep it raw bytes. Decoding is permissive about the declared content-type; content_type_matches(codec) is there for a workflow that wants to check and decide for itself. The same codec contract exists in all five SDKs.

Signals, queries & updates

A workflow’s externally callable operations are plain top-level functions marked # @handler, with the kind - @query (read-only), @signal (fire-and-forget), @update (mutate and return) - declared in a WIT contract hop build --wit contract.wit checks and wires:

# @handler
def get_status():
    return b"pending"

# @handler
def cancel():
    pass

# @handler
def add_item(item):
    return b"item-added:" + item

A @query handler that emits a command - an activity, a timer, a searchable mark - raises ContractViolation at the call site, and the engine’s own history check backstops it.

Handler payloads take the same codecs workflow inputs do - contract_encoded is the contract-side counterpart of encoded(), sitting between the marker and the def:

# @handler
@contract_encoded(order_codec, receipt_codec)
def add_item(order):
    return receipt_for(order)

Clients call the handlers with the matching verbs, bytes-shaped or carrying the same codec choice; a contract dispatch has no header channel, so no content-type travels with it - the codec is the contract, on both ends of the call:

client.signal_workflow("prod", "order-7", "cancel")
status = client.query_workflow("prod", "order-7", "get-status")
receipt = client.update_workflow(
    "prod", "order-7", "add-item", order,
    input_codec=order_codec, output_codec=receipt_codec,
)

Generated activity names

The activity name is not checked. Misspell it and everything still compiles: the dispatch is accepted, durably recorded under a task type no worker is registered for, and never picked up. The run blocks forever, with every component behaving exactly as designed.

hop typegen py removes that by generating the names from what is deployed:

hop typegen py --out workflows/hopskip_deployed.py
from hopskip_deployed import RESERVE_INVENTORY, CHARGE_CARD

reserved = await activity(RESERVE_INVENTORY, b"sku:ABC")
charged = await activity(CHARGE_CARD, b"amount:20")

A name that is not deployed is not in the generated file, so calling one stops compiling. Each constant carries its whole task type, including the version, so importing one is how you choose which version to call. See Which version you call.

hop typegen py --check fails a build whose committed copy no longer matches the plan it was generated from. Put it in CI, and regenerate after a deploy that changes the set.

Searchable state

async def order_processing():
    status = Searchable("status", "pending")     # re-extracted at each checkpoint
    mark_searchable("customer_tier", "gold")     # one-shot tag

    reserved = await activity("reserve_inventory", b"sku:ABC")
    status.set("reserved")

    charged = await activity("charge_card", b"amount:20")
    status.set("charged")
    return charged

__hopskip_workflow = durable(order_processing)

Both entry points (the Searchable wrapper and the bare mark_searchable free function) reach the emit_searchable host import.

Dataset lineage

Declare what a run reads and writes and it joins your OpenLineage graph. See the lineage guide:

dataset_input("orders_source", "postgres://db:5432", "shop.public.orders")
dataset_output("rollup", "s3://warehouse", "orders/2026-07-30.parquet")

Perpetual entities

An @entity is a workflow that carries state across event horizons. When Core seals and reopens the log, the host hands the next generation the sealed snapshot; resume_entity reconstructs it, or starts fresh:

@entity
class CounterEntity:
    def __init__(self, total):
        self.total = total

    def to_snapshot(self):
        return self.total.to_bytes(8, "little", signed=True)

    @classmethod
    def resume_from_snapshot(cls, snapshot):
        return cls(int.from_bytes(snapshot, "little", signed=True))

async def counter_workflow():
    state = resume_entity(lambda: CounterEntity(0))
    tick = await activity("tick", b"")
    state.total += int.from_bytes(tick, "little", signed=True)
    return state.total.to_bytes(8, "little", signed=True)

__hopskip_workflow = durable(counter_workflow)

The snapshot encoding is identical across languages, so a snapshot written by the Rust or TypeScript version resumes the Python guest identically. That is the interchangeable-history requirement.

Streaming

async def stream_processing():
    for chunk in [b"chunk001", b"chunk002", b"chunk003"]:
        try:
            await yield_chunk(chunk)
        except StreamDisconnectedError:
            await activity("reroute-store", b"disconnected")
            return b"rerouted"
    return b"streamed:3"

__hopskip_workflow = durable(stream_processing)

What’s enforced for you

Non-deterministic operations are unavailable inside the sandbox. Wall-clock time, entropy, and network I/O are either rebound to deterministic host imports or trap. Do all I/O in activities. See Determinism & the sandbox.