Python SDK

Reference for runstate-sdk on PyPI, the Python SDK for Python 3.10 and later. AsyncRunstate, the blocking Runstate client, every method and default, and where behaviour differs from TypeScript.

pip install runstate-sdk

The package is runstate-sdk on PyPI and imports as runstate (Python 3.10 or later, MIT licensed, depends on httpx).

from runstate import AsyncRunstate, Runstate, ConflictError, LeaseLostError

The Python SDK has the same object graph as the TypeScript SDK, with snake_case names and keyword-only options. This page lists every signature and calls out where the two SDKs behave differently.

Differences from TypeScript

Area	Python	TypeScript
Lease lost during `claim.run()` / `pool.run()`	The handler is cancelled and `run()` raises `LeaseLostError`.	The handler keeps running; `lease.signal` aborts and `lease.isLost` becomes true.
Checking a lease	`lease.is_lost`, `lease.lost_event` (`asyncio.Event`), `lease.raise_if_lost()`	`lease.isLost`, `lease.signal`
Stopping `consume()` / `watch()`	`stop_event=` (an `asyncio.Event`)	`signal:` (an `AbortSignal`)
Cancelling other waits	Cancel the asyncio task; there is no `signal` parameter.	`signal` on `acquire`, `take`, `result`, `wait`
Handlers	`async def` or plain functions (plain ones run in a thread pool).	`async` functions
Return values	Most results are `dict`s with the API’s camelCase keys (`view["storedState"]`).	Typed objects
`delivery.complete()`	`complete(result={"payload": ...})`	`complete({ payload })`
`timer.wait()` result	`{"fired_at": datetime}`	`{ firedAt: Date }`
Closing	`await rs.aclose()` or `async with`; blocking client: `rs.close()` or `with`	`rs.close()` (no-op)
Low-level client	none	`createRunstateClient`

AsyncRunstate

The primary client. Constructing it validates configuration and makes no network calls.

class AsyncRunstate:
    def __init__(
        self,
        *,
        api_key: str | None = None,  # default: RUNSTATE_API_KEY (required)
        space_id: str | None = None,  # default: RUNSTATE_SPACE_ID (required)
        base_url: str | None = None,  # default: RUNSTATE_BASE_URL, else "http://localhost:8080"
        request_timeout_ms: int | None = None,  # default 10_000, per attempt
        retries: int | None = None,  # default 2, transport failures only
        holder: str | None = None,  # default: random "worker-xxxxxx"
    ) -> None: ...

    space_id: str
    scopes: ScopesAPI
    mailboxes: MailboxesAdminAPI
    pools: PoolsAdminAPI
    quotas: AllowancesAdminAPI
    work_limits: WorkLimitsAdminAPI
    budgets: BudgetsAdminAPI
    events: EventsAPI
    diagnostics: DiagnosticsAPI

    def scope(self, scope_id: str) -> ScopeHandle: ...  # local handle, no request
    def with_space(self, space_id: str) -> AsyncRunstate: ...  # shares transport and identity
    async def aclose(self) -> None: ...
    async def __aenter__(self) -> AsyncRunstate: ...
    async def __aexit__(self, *exc) -> None: ...

Missing api_key or space_id raises ConfigError. See Configuration & auth.

Blocking client

Runstate wraps AsyncRunstate for synchronous code. It starts one background thread running an event loop, forwards every call to it, and blocks until the result is ready. It takes the same keyword arguments.

class Runstate:
    def __init__(self, **kwargs) -> None: ...  # same keywords as AsyncRunstate
    # Every attribute and method of AsyncRunstate, called without await.
    def watch(self, **kwargs) -> Iterator[dict]: ...  # blocking iterator over events.watch
    def new_event(self) -> ThreadSafeEvent: ...  # stop flag for consume(stop_event=...)
    def close(self) -> None: ...
    def __enter__(self) -> Runstate: ...
    def __exit__(self, *exc) -> None: ...

from runstate import Runstate

with Runstate(holder="nightly-report") as rs:
    rs.mailboxes.ensure("work", mode="WORK")
    run = rs.scopes.create()
    ticket = run.mailbox("work").submit({"report": "daily"}, key="daily-2026-09-15")
    print(ticket.result(timeout_ms=120_000)["state"])

Handles returned by the blocking client (runs, queues, tickets) are blocking too. Two things to know when you pass a handler to consume() or run():

The handler receives the async objects. A Delivery or ClaimLease passed to your handler has coroutine methods. Write the handler as async def if it needs await delivery.complete(result=...). A plain def handler runs in a thread pool; its delivery is completed automatically (without a payload) when it returns and retried when it raises.
Don’t call the blocking client from an async def handler. That handler runs on the client’s own event loop, so a blocking call would wait on itself. Use the async objects you were given, or a plain def handler, which may call blocking methods freely.

import os

from runstate import Runstate

rs = Runstate(holder="worker-1")
queue = rs.scope(os.environ["RUNSTATE_SCOPE_ID"]).mailbox("work")
stop = rs.new_event()  # call stop.set() from a signal handler or another thread


async def handle(data, delivery):
    await delivery.complete(result={"payload": {"doubled": data["n"] * 2}})


queue.consume(handle, lease_seconds=30, stop_event=stop)  # blocks until stopped
rs.close()

Runs

class ScopesAPI:
    async def create(
        self,
        *,
        name: str | None = None,  # label, 1–256 chars; returned by status(), not unique
        parent_id: str | None = None,
        deadline: str | None = None,  # ISO 8601
        child_limit: int | None = None,  # 1–10000, server default 1000
    ) -> ScopeHandle: ...


class ScopeHandle:
    id: str
    timers: TimersAPI
    barriers: BarriersAPI
    groups: GroupsAPI

    async def status(self) -> dict: ...  # {"id", "name", "parentId", "storedState", "effective", "deadline", ...}
    async def child(
        self, *, name: str | None = None, deadline: str | None = None, child_limit: int | None = None
    ) -> ScopeHandle: ...
    async def cancel(self) -> dict: ...  # {"id", "state", "version"}
    async def complete(self) -> dict: ...

    def claim(self, key: str) -> ClaimRef: ...
    def mailbox(self, name: str) -> MailboxRef: ...
    def mailbox_by_id(self, mailbox_id: str) -> MailboxRef: ...
    def pool(self, name: str) -> PoolRef: ...
    def pool_by_id(self, permit_id: str) -> PoolRef: ...
    def quota(self, name: str) -> AllowanceRef: ...
    def quota_by_id(self, allowance_id: str) -> AllowanceRef: ...
    def budget(self, name: str) -> BudgetRef: ...
    def task(self, task_id: str) -> TaskTicket: ...  # re-attach by id, no request

Space-level resources

class MailboxesAdminAPI:  # rs.mailboxes
    async def ensure(self, name: str, *, mode: str | None = None, backlog_limit: int | None = None) -> dict: ...

class PoolsAdminAPI:  # rs.pools
    async def ensure(self, name: str, *, capacity: int) -> dict: ...

class AllowancesAdminAPI:  # rs.quotas
    async def ensure(self, name: str, *, units_per_window: int, window_seconds: int) -> dict: ...

class WorkLimitsAdminAPI:  # rs.work_limits
    async def ensure(self, name: str, *, scope_id: str, max_outstanding: int) -> dict: ...

class BudgetsAdminAPI:  # rs.budgets
    async def ensure(self, name: str, *, currency: str, scale: int, limit: str) -> dict: ...

ensure creates the resource or returns the existing one when its settings match, and raises ConflictError (budgets: RunstateError with code CONFLICT) when they don’t. With a key that lacks resource_config, ensure returns a matching existing pool, quota, work limit or budget and raises AuthenticationError (FORBIDDEN) only when there is nothing to verify. Name lookups resolve on first use and raise NotFoundError if the resource doesn’t exist.

Claims

Guide: Claims, shared tasks and takeover.

class ClaimRef:
    scope_id: str
    key: str
    async def acquire(self, *, lease_seconds: int = 30, wait: bool = False, timeout_ms: int = 30_000) -> ClaimLease: ...
    async def try_acquire(self, **kwargs) -> dict: ...  # {"kind": "acquired", "lease": ClaimLease} | {"kind": "busy"}
    async def run(self, fn, **kwargs): ...  # fn(lease), async or plain; kwargs go to acquire()


class ClaimLease:
    key: str
    generation: str
    expires_at: datetime
    lost_event: asyncio.Event
    @property
    def is_lost(self) -> bool: ...
    def token(self) -> str: ...
    def raise_if_lost(self) -> None: ...  # raises LeaseLostError
    async def renew(self, lease_seconds: int | None = None) -> None: ...
    async def release(self, observation=None) -> None: ...
    async def aclose(self) -> None: ...  # stop renewing without releasing

acquire() raises ConflictError (CLAIM_HELD) when the key is owned and wait is false. With wait=True it retries CLAIM_HELD every ~300 ms and CONCURRENCY_LIMITED every ~1 s until timeout_ms, then raises WaitTimeoutError.
Renewal runs about every lease_seconds / 3 (±10%, at least 1 s) and stops at the first failure, setting lost_event.
run() races your handler against lost_event. If the lease is lost first, the handler task is cancelled and run() raises LeaseLostError. A plain function running in a thread can’t be stopped; only the wait for it is abandoned. The lease is released in all cases.

Work queues

Guide: Claims, shared tasks and takeover.

class MailboxRef:
    scope_id: str
    async def send(self, data, *, work_key: str | None = None, deadline: str | None = None) -> dict: ...  # {"id", "replay"}
    async def submit(
        self,
        input,
        *,
        key: str,
        deadline: str | None = None,
        subscriber_id: str | None = None,
        requirements: TaskRequirements | None = None,  # {"pools": [{"name", "units"}], "quotas": [...], "budgets": [{"name", "reserveMinor"}]}
        work_limit: str | None = None,
    ) -> TaskTicket: ...
    async def receive(self, *, wait_ms: int = 0) -> Delivery | None: ...  # 30 s lease, not renewed; wait_ms long-polls
    async def admit(self, *, lease_seconds: int = 30, wait_ms: int = 0) -> AdmittedWork | None: ...
    async def consume(
        self,
        handler,  # handler(data, delivery), async or plain
        *,
        concurrency: int = 1,
        lease_seconds: int = 30,
        poll_ms: int = 250,
        wait_ms: int = 20_000,  # server-side long poll while the queue is empty; 0 disables
        shutdown_timeout: float = 30.0,  # seconds
        stop_event: asyncio.Event | None = None,
        on_error=None,  # on_error(err, delivery)
    ) -> None: ...


class Delivery:
    data: object
    id: str
    attempt: int
    scope_id: str
    async def complete(self, result: dict | None = None) -> None: ...  # {"payload": ..., "mailboxId"?, "scopeId"?, "workKey"?}
    async def retry(self) -> None: ...
    async def reject(self) -> None: ...
    async def renew(self, lease_seconds: int | None = None) -> None: ...


class AdmittedWork:
    task: TaskTicket
    delivery: Delivery
    resources: dict  # {"grants": [...], "consumptions": [...], "reservations": [...]}
    attempt: int


class TaskRequirements(TypedDict, total=False):  # from runstate import TaskRequirements
    pools: list[UnitsRequirement]  # up to 4; {"name": str, "units": int}
    quotas: list[UnitsRequirement]  # up to 4
    budgets: list[BudgetRequirement]  # up to 2; {"name": str, "reserveMinor": str}

consume() behaves like the TypeScript version: success completes, an exception calls on_error and retries the delivery, AuthenticationError / ConfigError / NotFoundError are re-raised, ScopeCancelledError ends the loop, and other errors are retried after poll_ms. When stop_event is set it stops taking work, waits up to shutdown_timeout seconds, and retries unfinished deliveries.

Tasks

class TaskTicket:
    id: str
    async def status(self) -> dict: ...  # the task view, see the TypeScript TaskView
    async def result(self, *, timeout_ms: int = 30_000, poll_ms: int = 250) -> dict: ...  # {"state", "outcome", "reason"}
    async def await_admission(self, *, timeout_ms: int = 30_000, poll_ms: int = 250) -> dict: ...
    async def cancel(self, reason: str | None = None) -> dict: ...
    async def join(self, subscriber_id: str, scope_id: str) -> dict: ...
    async def detach(self, subscriber_id: str) -> dict: ...

Task groups

Guide: Task groups and cancellation.

class GroupsAPI:  # run.groups
    async def create(
        self,
        *,
        mailbox: str,
        condition: str,  # "FIRST_ACCEPTED" | "N_ACCEPTED" | "ALL_TERMINAL"
        threshold: int | None = None,
        expected_members: int | None = None,  # 1-100; close automatically once this many have joined
        deadline: str | None = None,
        parent_scope_id: str | None = None,
    ) -> GroupHandle: ...
    def by_id(self, group_id: str) -> GroupHandle: ...


class GroupHandle:
    id: str
    async def status(self) -> dict: ...  # includes "scopeId", "state", "members"
    async def submit(self, input, *, key: str, deadline: str | None = None) -> TaskTicket: ...
    async def close(self) -> dict: ...  # no more members; evaluates now; {"state", "outcome", "outcomeReason"}
    async def seal(self) -> dict: ...  # deprecated: close(), but ConflictError once finalized
    async def accept(self, task_id: str) -> dict: ...  # {"verdict", "late", "groupState", "outcome", "outcomeReason"}
    async def reject(self, task_id: str) -> dict: ...
    async def join_task(self, task_id: str) -> dict: ...
    async def wait(self, *, timeout_ms: int = 30_000, poll_ms: int = 250) -> dict: ...  # {"outcome", "outcomeReason"}

Concurrency pools

Guide: Quotas and concurrency pools.

class PoolRef:
    scope_id: str
    async def status(self) -> dict | None: ...  # {"id", "name", "unitsTotal"}
    async def acquire(self, *, units: int = 1, lease_seconds: int = 30, wait: bool = True, timeout_ms: int = 60_000) -> PoolLease: ...
    async def try_acquire(self, **kwargs) -> dict: ...  # {"kind": "acquired", "lease"} | {"kind": "capacity-unavailable"}
    async def run(self, fn, **kwargs): ...  # fn(lease); cancelled with LeaseLostError on lease loss


class PoolLease:
    grant_id: str
    expires_at: datetime
    lost_event: asyncio.Event
    @property
    def is_lost(self) -> bool: ...
    def token(self) -> str: ...
    def raise_if_lost(self) -> None: ...
    async def renew(self, lease_seconds: int | None = None) -> None: ...
    async def release(self, observation=None) -> None: ...
    async def aclose(self) -> None: ...

try_acquire() always sets wait=False.

Shared quotas

Guide: Quotas and concurrency pools.

class AllowanceRef:  # run.quota(name)
    scope_id: str
    async def status(self) -> dict | None: ...  # {"unitsLimit", "windowStart", "consumed", "remaining", "cooldownUntil", ...}
    async def try_take(self, *, units: int = 1) -> dict: ...  # {"kind": "consumed", "consumption"} | {"kind": "exhausted"}
    async def take(self, *, units: int = 1, wait: bool = True, timeout_ms: int = 30_000) -> dict: ...  # consumption
    async def cooldown(self, seconds: int, reason: str | None = None) -> dict: ...  # {"cooldownUntil"}

Budgets

Guide: Spend budgets.

class BudgetRef:  # run.budget(name)
    scope_id: str
    async def status(self) -> dict: ...  # balances in minor units plus "reservations"
    async def reserve(self, *, amount: str, settle_by: str | None = None) -> BudgetReservation: ...


class BudgetReservation:
    id: str
    reserved_minor: str  # as of reserve()
    state: str  # as of reserve()
    @property
    def reserved(self) -> str: ...  # major units
    async def settle(self, *, amount: str, usage_key: str) -> dict: ...  # {"entryId", "reservedMinor", "state", "replay"}
    async def void(self, *, usage_key: str) -> dict: ...


def decimal_to_minor(amount: str, scale: int) -> str: ...  # "1.25", 2 -> "125"
def minor_to_decimal(minor: str, scale: int) -> str: ...  # "125", 2 -> "1.25"

Barriers and timers

class BarriersAPI:  # run.barriers
    async def create(self, *, target: int) -> BarrierHandle: ...
    def get(self, barrier_id: str) -> BarrierHandle: ...


class BarrierHandle:
    id: str
    async def arrive(self, *, key: str) -> dict: ...  # {"state", "arrivals"}
    async def status(self) -> dict: ...  # {"id", "state", "epoch", "arrivals", "target"}
    async def wait(self, *, timeout_ms: int = 60_000, poll_ms: int = 250) -> dict: ...


class TimersAPI:  # run.timers
    async def create(self, *, after_seconds: int | None = None, at: str | None = None) -> TimerHandle: ...
    def get(self, timer_id: str) -> TimerHandle: ...  # fire_at unknown on a re-attached handle


class TimerHandle:
    id: str
    scope_id: str
    fire_at: datetime
    async def wait(self, *, timeout_ms: int = 30_000, poll_ms: int = 250) -> dict: ...  # {"fired_at": datetime}
    async def cancel(self) -> dict: ...

Events and diagnostics

Guide: Events and monitoring.

class EventsAPI:  # rs.events
    async def list(self, *, cursor: str | None = None, limit: int | None = None) -> dict: ...  # {"events", "nextCursor"}
    async def watch(
        self,
        *,
        cursor: str | None = None,
        types: list | None = None,
        aggregate: dict | None = None,  # {"type": "task", "id": task_id}
        scope_id: str | None = None,
        limit: int | None = None,
        wait_seconds: int | None = None,  # 0–30
        stop_event: asyncio.Event | None = None,
    ) -> AsyncIterator[dict]: ...  # each event dict has an added "cursor"


class DiagnosticsAPI:  # rs.diagnostics
    async def run(self, scope_id: str) -> dict: ...
    async def waits(self) -> dict: ...
    async def workers(self, scope_id: str) -> list: ...

Errors

class RunstateError(Exception):
    code: str  # e.g. "CLAIM_HELD"
    status: int  # HTTP status, 0 for client-side errors
    message: str
    request_id: str | None
    retry_after_ms: int | None

class ConfigError(RunstateError): ...
class AuthenticationError(RunstateError): ...
class NotFoundError(RunstateError): ...
class ConflictError(RunstateError): ...
class ScopeCancelledError(RunstateError): ...
class LeaseLostError(RunstateError): ...
class RateLimitError(RunstateError): ...
class UnavailableError(RunstateError): ...
class WaitTimeoutError(RunstateError): ...
class CursorExpiredError(RunstateError): ...  # reserved; not emitted today

def map_error(code: str, status: int, message: str, request_id: str | None = None, retry_after_ms: int | None = None) -> RunstateError: ...

from runstate import ConflictError, RateLimitError, RunstateError

try:
    await run.claim("company:acme").acquire()
except ConflictError as err:
    if err.code == "CLAIM_HELD":
        print("someone else owns it")
    else:
        raise
except RateLimitError as err:
    print(f"rate limited: {err.code}, retry after {err.retry_after_ms} ms")
    raise
except RunstateError as err:
    print(err.code, err.status, err.request_id)
    raise

Codes map to the same classes as in TypeScript; both SDKs are tested against one shared error-mapping fixture. The full table is on Errors & retries.

Config and resolve_config are also exported; they hold the resolved client configuration.