Python SDK

pip install runstate-sdk

The package is runstate-sdk on PyPI and imports as runstate (Python 3.10 or later, MIT licensed). You decorate a function once where you define it and call it exactly as before. runstate does the coordination inside the call.

import runstate


@runstate.once
def research_company(name: str) -> dict: ...


@runstate.limit("search-api", max=60, per="minute")
async def search_web(query: str) -> list: ...


@runstate.slots("browser", max=5)
def open_page(url: str) -> str: ...


@runstate.budget("model-usd", reserve=0.40, cost=lambda r: r.usd, limit=50)
def call_model(prompt: str): ...


research_company("acme")  # looks like a normal call
runstate.stop()  # every worker stops taking new work

The module functions use a default client. The package also exports the Runstate class, is_runstate_error, the error classes and the result types (RunContext, RunEvent, RunStatus, Job, Queue). The concepts and defaults are the same as in the TypeScript SDK, with snake_case names and keyword options.

The HTTP API is still available if you work in another language. Everything on this page maps onto it.

Sync and async

Every decorator works on both plain functions and async def functions:

• a decorated plain function stays a plain, blocking function;
• a decorated async def function stays a coroutine function, and you await it.

Every module function that talks to the API has an a-prefixed async form:

Blocking	Async
runstate.setup(...)	await runstate.asetup(...)
runstate.run_id()	await runstate.arun_id()
runstate.stop()	await runstate.astop()
runstate.finish()	await runstate.afinish()
runstate.take(name)	await runstate.atake(name)
runstate.sleep(name, duration)	await runstate.asleep(name, duration)
runstate.together(name, count)	await runstate.atogether(name, count)
runstate.status()	await runstate.astatus()
for e in runstate.events()	async for e in runstate.aevents()
runstate.serve()	await runstate.aserve()
queue(x) (producer-only queue)	await queue.acall(x)
queue.send(x)	await queue.asend(x)
job.result()	await job.aresult()
queue.gather(xs)	await queue.agather(xs)

Use the blocking forms in plain code and the a forms inside async def.

Configuration

Environment variables

Variable	Required	Default	Meaning
RUNSTATE_API_KEY	yes	none	The API key from the console.
RUNSTATE_SPACE_ID	yes	none	The space the key works in.
RUNSTATE_BASE_URL	no	https://api.getrunstate.com	API address. Use http://localhost:8080 for a local kernel.
RUNSTATE_RUN	no	default	The run name every process joins.

The default client reads these on first use. Importing runstate makes no network calls. Missing configuration raises ConfigError on the first call.

configure and Runstate

import runstate

runstate.configure(api_key=..., space_id=..., base_url=..., run="nightly")  # change the default client
rs = runstate.Runstate(api_key=..., space_id=..., run="nightly")  # or make your own

Option	Default	Notes
api_key	RUNSTATE_API_KEY	Required.
space_id	RUNSTATE_SPACE_ID	Required.
base_url	RUNSTATE_BASE_URL, else https://api.getrunstate.com
run	RUNSTATE_RUN, else "default"	The run this client joins.
worker	<hostname>-<pid>	A label for this process, shown in the console.
timeout	"30s"	Per request.
retries	3	Transport retries.

A Runstate instance has exactly the same methods as the module. Everything below written as runstate.x works on an instance too, for example @rs.once.

Durations and amounts

• Durations are a number of seconds or a string: "500ms", "30s", "10m", "2h", "1d".
• per (for limit) also accepts "second", "minute", "hour" or "day".
• Amounts are numbers or decimal strings in major units (0.40, "12.5"). The SDK converts them to exact integer minor units using the budget’s scale. Numbers are rounded half up to the scale, so float noise like 0.1 + 0.2 becomes 0.30. A string with more decimal places than the scale is a ConfigError.

Runs

Every call belongs to a run. By default all processes join the run named by RUNSTATE_RUN (default). Joining is get-or-create: the first call creates the run, and every other process with the same name joins it.

runstate.run(name: str, *, deadline: float | str | None = None) -> Runstate
runstate.child(name: str | None = None, *, deadline: float | str | None = None) -> Runstate
runstate.stop() -> None       # await runstate.astop()
runstate.finish() -> None     # await runstate.afinish()
runstate.run_id() -> str      # await runstate.arun_id()

nightly = runstate.run("nightly-2026-09-17", deadline="2h")  # a client bound to that run


@nightly.once
def fetch_prices(currency: str) -> dict: ...


child = runstate.child("retry-batch", deadline="10m")  # a child of the current run
runstate.stop()  # stop the current run: nothing new starts anywhere
runstate.finish()  # mark it finished

• run(name) returns a client bound to the named root run. The run is created on the first call that needs it.
• After a run is stopped or finished, or its deadline passes, the next call with the same name starts a fresh run. That is how once results reset: they belong to one run.
• child(name) returns a client bound to a new child run of the current one. Stopping a run stops its children.
• stop() and finish() are idempotent.
• run_id() returns the id of the client’s run, joining or creating the run if needed.
• A client that has seen its run end keeps raising StoppedError, so a loop in a stopped worker can’t quietly start the next run. A new process, or a fresh runstate.run(name) client, joins the next run.
• When a run stops, every wrapped call in flight sees context().signal set, serve() stops taking work and drains, and new calls raise StoppedError.

context

ctx = runstate.context()
ctx.run.id  # the run's id
ctx.run.name  # the run's name
ctx.signal  # set when the work should stop: ctx.signal.is_set()
ctx.attempt  # 1 for the first try
ctx.key  # the work key for once calls and queue jobs, otherwise None

Inside a decorated function or queue handler, context() tells you which run you are in, which attempt this is and the key of the work. It returns None outside a wrapped call.

signal works like a threading.Event. It is set when the run stops, its deadline passes, or this process loses its hold on the work. Long work should check it:

@runstate.once
def build_report(name: str) -> dict:
    sections = []
    for part in PARTS:
        if runstate.context().signal.is_set():
            raise RuntimeError("stopping")
        sections.append(write_section(name, part))
    return {"sections": sections}

When decorators are nested, context() describes the innermost wrapper that holds something (once, slots or a queue job). Otherwise it describes the run.

once

@runstate.once
@runstate.once(name: str | None = None, key: Callable[..., str] | None = None, attempts: int = 3, lease: float | str = "30s")

@runstate.once(attempts=3)
def research_company(name: str) -> dict: ...


summary = research_company("acme")

• The first caller with given arguments runs the function. Callers with the same arguments, in any process, wait and get the same result.
• If the function raises, the next waiting caller tries again. After attempts failures every caller gets OnceFailedError with the last error message.
• If the process dies, its lease lapses and another caller takes over. The lease is renewed automatically while the function runs.
• The result must be JSON-serializable and at most 64 KB (ResultTooLargeError). Store big outputs elsewhere and return a reference.
• Results belong to the run. A fresh run runs the function again.
• name defaults to the function name. key receives the same arguments as the function and returns a string.

Keys

The key is <name>:<hash>, where hash is the first 32 hex characters of the SHA-256 of the canonical JSON of the argument list:

• JSON with object keys sorted, no whitespace, UTF-8, lists kept in order.
• Tuples become arrays. Dataclasses and pydantic models are dumped to dicts first. None is null.
• Integers print as integers; floats use the shortest round-trip form (1.5, not 1.50). nan and inf are a ConfigError.
• Anything else (functions, class instances, datetimes) is a ConfigError. Pass key instead.
• A lambda without name is a ConfigError.

The TypeScript SDK produces identical keys, so a Python process and a TypeScript process calling the same once with the same arguments share one result.

limit and take

@runstate.limit(name: str, *, max: int | None = None, per: float | str | None = None, units: int = 1, timeout: float | str = "60s")
runstate.take(name: str, *, units: int = 1, timeout: float | str = "60s") -> None   # await runstate.atake(...)

@runstate.limit("search-api", max=60, per="minute", units=1)
def search_web(q: str) -> list: ...


search_web("fusion")  # waits for a unit, then calls search_web

runstate.take("search-api")  # just take a unit, no function

• Every worker draws from the same limit. When the window is used up, the call waits in a fair queue on the server until the next window.
• max and per declare the limit. If it doesn’t exist, it is created (the key needs the resource_config permission). If it exists with different sizes, the call raises ConflictError. Omit both to use a limit created in the console.
• If no unit arrives within timeout, the call raises TimeoutError.
• Units are spent even if the function then fails.

slots

@runstate.slots(name: str, *, max: int | None = None, units: int = 1, timeout: float | str = "60s", lease: float | str = "30s")
with runstate.slots(name, ...): ...
async with runstate.slots(name, ...): ...

@runstate.slots("browser", max=5)
async def open_page(url: str) -> str: ...


with runstate.slots("browser"):  # also a context manager
    render_pdf()

• Waits for a free slot, runs the function or the with block, and always gives the slot back, also on error.
• The slot is kept alive automatically while the work runs. If the process dies, the slot returns after its lease lapses.
• max declares the slots the same way as limit: created if missing, ConflictError if the size differs.
• If no slot frees up within timeout, the call raises TimeoutError.
• Use async with in async code.

budget

@runstate.budget(
    name: str,
    *,
    reserve: float | str,              # held before the call
    cost: Callable[[R], float | str] | None = None,  # the real cost, read from the result; default: the reserve
    limit: float | str | None = None,  # declare the budget
    currency: str = "USD",
    scale: int = 2,                    # decimal places
)

@runstate.budget("model-usd", reserve=0.40, cost=lambda r: r.usd, limit=50)
def call_model(prompt: str): ...

• Before the call, reserve is held. If the budget can’t cover it, the function is not called and BudgetExceededError is raised.
• After the call, cost(result) is charged and the rest of the reservation is released. If the function raises, the whole reservation is released.
• If the real cost is higher than the reservation, the extra is charged from what is left. If the budget can’t cover it, what it can is charged, the result is still returned, and the overrun is reported in the run’s events as budget.overrun. Pick reserve as your worst case to avoid this.
• cost may return a number or a decimal string. Returning 0 releases everything. Without cost, the full reserve is charged when the call succeeds.
• limit declares the budget: created if missing, ConflictError if it differs.

queue

A queue hands work to worker processes. Define it once in a module that both producers and workers import. The handler takes one argument, the input, and returns the result. Both must be JSON-serializable and at most 64 KB.

@runstate.queue(
    name: str,
    *,
    slots: str | dict | None = None,   # "browser" or {"name": "browser", "units": 2}
    limit: str | dict | None = None,   # "model-rpm" or {"name": ..., "units": ...}
    budget: dict | None = None,        # {"name": "model-usd", "reserve": 0.40, "cost": lambda r: r["usd"]}
    attempts: int = 5,
    lease: float | str = "30s",
    concurrency: int | None = None,    # handlers at once in this process when serving, default 1
    key: Callable[[Any], str] | None = None,  # default: hash of the input
)
runstate.queue(name: str, **options) -> Queue   # producer-only: no handler

queue(input) -> R                    # submit and wait; await queue.acall(input) in async code
queue.send(input) -> Job             # await queue.asend(input)
queue.gather(inputs, *, need="all", accept=None, timeout=None) -> list   # await queue.agather(...)
queue.name -> str
job.id -> str
job.result(timeout: float | str | None = None) -> R   # await job.aresult(timeout=...)

runstate.serve(*, concurrency: int | None = None, queues: list[Queue] | None = None, signal=None) -> None
# await runstate.aserve(...)

# review.py, imported by producers and workers
import runstate


@runstate.queue(
    "review",
    slots="browser",
    budget={"name": "model-usd", "reserve": 0.40, "cost": lambda r: r["usd"]},  # cost defaults to the reserve
    attempts=5,
    concurrency=4,
)
def review(doc: dict) -> dict:
    return score_doc(doc)


# producer
score = review(doc)  # submit and wait for the result
job = review.send(doc)  # submit only; await review.asend(doc)
print(job.id)
later = job.result(timeout="5m")  # await job.aresult(timeout="5m")

# worker process
runstate.serve()  # runs every queue defined in this process; await runstate.aserve()

• With a plain handler, calling the queue blocks until the result is ready. With an async def handler, await review(doc).
• Submitting the same input twice in a run returns the same job. The key is the input hash; pass key to choose it yourself.
• A worker takes a job together with everything the queue declared (slots, limit units, budget reservation) in one step, so a job never starts half-ready.
• If the handler raises, the job is retried with backoff until attempts is reached. Then callers get TaskFailedError with the last error message.
• If the worker dies, the job goes back to the queue after its lease lapses.
• slots and limit take a name or {"name": ..., "units": ...}. budget takes {"name": ..., "reserve": ..., "cost": ...}, where cost is optional. Without cost, the job is charged its full reserve when it succeeds and nothing when it fails.
• The limit, slots and budget a queue names must already exist. Declare them with setup, with limit, slots or budget elsewhere, or in the console.
• job.result() waits without a limit by default. With timeout, it raises TimeoutError when the time passes. The job keeps going on the server.
• If a worker loses its hold on a job because another worker took it over, its result is dropped quietly. The caller gets the result of the worker that finished the job.
• review.name is the queue’s name.

Producer-only queues

runstate.queue("review") without a decorated function is producer-only. You can call it, send to it and gather on it. serve() skips it. Use it when the workers run in another process or language.

review = runstate.queue("review")

score = review(doc)  # in plain code
score = await review.acall(doc)  # in async code

Decorating a function with @runstate.queue("review") registers it as the handler instead.

The queue remembers the slots, limit units and budget its workers declared, so a producer-only queue doesn’t repeat them. Jobs sent through it need the same resources as jobs sent by the workers. When workers are redeployed with different resources, the latest definition wins.

A producer-only queue uses the queue a worker has declared. Until a worker that defines the queue has started, calls raise a RunstateError with code NOT_FOUND and the message queue review not found. Start the workers first, or pass the same attempts the workers use so the producer declares the queue itself (the key then needs the resource_config permission if the queue doesn’t exist yet).

serve

serve() runs every queue with a handler defined through this client, or through any client made from it with run() or child(). Pass queues=[review] to run only some. It takes jobs from the run and from every run under it, including child runs and gathers.

It runs until the run stops, signal is set, or the process gets SIGTERM or SIGINT. Then it stops taking work and waits up to 30 seconds for running handlers before it returns. concurrency overrides each queue’s own setting. In async code, await runstate.aserve(); cancelling that task stops it the same way.

runstate.serve(queues=[review], concurrency=8)

gather

accepted = review.gather(docs, need=5, accept=lambda result, doc: result["score"] > 0.8, timeout="10m")
# accepted = await review.agather(docs, need=5, ...)

• Submits every input as one group, at most 100 inputs per call. need is "all" or a whole number from 1 to the number of inputs. accept runs in the producer as each result arrives. By default every result is accepted.
• Returns the accepted results, in completion order, once need of them are accepted, and cancels the rest.
• With need="all" (the default) it returns when every job has finished, with the accepted results only.
• If need can no longer be reached, it raises TaskFailedError.

setup

runstate.setup(*, limits: dict | None = None, slots: dict | None = None, budgets: dict | None = None) -> None
# await runstate.asetup(...)

setup() creates or verifies resources without calling anything or spending units. It is the simplest way to declare the resources a queue names. Each entry follows the same rules as max/per on limit, max on slots and limit on budget: created if missing, ConflictError if the sizes differ.

runstate.setup(
    limits={"search-api": {"max": 60, "per": "minute"}},
    slots={"browser": {"max": 5}},
    budgets={"model-usd": {"limit": 50, "currency": "USD", "scale": 2}},
)

sleep and together

runstate.sleep(name: str, duration: float | str | None = None, *, until: datetime | None = None) -> None   # await runstate.asleep(...)
runstate.together(name: str, count: int, *, timeout: float | str | None = None) -> None   # await runstate.atogether(...)

from datetime import datetime, timezone

runstate.sleep("cooldown:acme", "10m")
runstate.sleep("report-window", until=datetime(2026, 9, 18, 9, 0, tzinfo=timezone.utc))
runstate.together("stage-2", 5, timeout="5m")

• sleep(name, duration) is a durable timer in the run. Pass until= with a datetime to sleep until a point in time instead. A process that restarts and calls the same name waits only for what is left. Raises StoppedError if the run stops first.
• Sleep names are per run. A name that already fired returns at once when used again in the same run, so give each distinct wait its own name.
• together(name, count) returns once count different workers have called it with that name. Workers are counted by their worker label (default <hostname>-<pid>), so a restarted process with the same label counts once. A worker that arrives after the group already went through continues immediately.
• together waits without a limit unless you pass timeout. When the timeout passes, it raises TimeoutError.

events and status

for event in runstate.events(types=["task.failed"]):
    print(event.type, event.data)

async for event in runstate.aevents(types=["task.failed"]):
    ...

status = runstate.status()  # await runstate.astatus()
print(status.run.state, [w.worker for w in status.workers])

• events(types=...) streams the current run’s events and resumes where it left off after network errors. Break out of the loop to stop it.
• Each event has id, type, at (a datetime) and data (the event’s payload).
• status() returns the run’s state and deadline, the workers currently holding work, and what is waiting, as attributes:
  • status.run: id, name, state ("running", "stopped", "finished" or "expired") and deadline;
  • status.workers: each with worker, holding (how many things it holds) and lease_expires_at;
  • status.waiting: each with kind ("slots"), name and units.
• workers and waiting are empty when the key lacks the usage permission.

See Events and monitoring for event types.

Errors

Every error is a runstate.RunstateError with code, status, request_id (a string, or None), retry_after (seconds, or None) and retryable (true for contention the SDK normally waits out). runstate.is_runstate_error(err) tells you whether an exception came from runstate.

Class	When
ConfigError	Missing key or space, bad durations or amounts, arguments that can’t be hashed.
AuthError	Bad key, missing permission, suspended space, plan limit reached.
StoppedError	The run was stopped, finished or passed its deadline.
TimeoutError	A wait (limit, slots, take, together, a job result) ran out. Also a subclass of the built-in TimeoutError.
OnceFailedError	A once function failed attempts times. Has attempts and message.
TaskFailedError	A queue job failed, or a gather can’t reach need. Has job_id and reason.
BudgetExceededError	A reservation doesn’t fit in the budget.
ResultTooLargeError	An input or result is over 64 KB.
ConflictError	Sizes in code don’t match the existing resource, or a queue’s attempts differ between processes.
LeaseLostError	The process lost its hold on the work because someone else took it over. For queue jobs, the result is dropped quietly instead.
UnavailableError	runstate couldn’t be reached after retries.

Busy resources, rate limits and plan concurrency limits are retried inside the SDK, honouring Retry-After. They never surface as errors unless a timeout runs out. See Errors & retries.

import runstate

try:
    answer = call_model(prompt)
except runstate.BudgetExceededError:
    answer = None  # the call never started
except runstate.StoppedError:
    answer = None  # the run is over