# 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. Source: https://docs.getrunstate.com/sdk/python/ ```bash 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`). ```python from runstate import AsyncRunstate, Runstate, ConflictError, LeaseLostError ``` The Python SDK has the same object graph as the [TypeScript SDK](https://docs.getrunstate.com/sdk/typescript/), 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`](https://docs.getrunstate.com/sdk/low-level-client/) | ## AsyncRunstate The primary client. Constructing it validates configuration and makes no network calls. ```python 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](https://docs.getrunstate.com/configuration/). ## 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. ```python 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: ... ``` ```python 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. ```python 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 ```python 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 ```python 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](https://docs.getrunstate.com/guides/work-once/#own-a-key-with-a-claim). ```python 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](https://docs.getrunstate.com/guides/work-once/#workers-and-crash-takeover). ```python 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 ```python 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](https://docs.getrunstate.com/guides/completion-and-cancellation/#task-groups). ```python 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](https://docs.getrunstate.com/guides/share-capacity/#concurrency-pools). ```python 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](https://docs.getrunstate.com/guides/share-capacity/#shared-quotas). ```python 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](https://docs.getrunstate.com/guides/budgets/). ```python 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 ```python 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](https://docs.getrunstate.com/guides/observability/). ```python 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 ```python 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: ... ``` ```python 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](https://docs.getrunstate.com/errors/). `Config` and `resolve_config` are also exported; they hold the resolved client configuration.