runstate (Python SDK)

Python SDK for runstate, the coordination layer for agent swarms. Decorate a function once where you define it, then call it exactly as before. runstate does the coordination inside the call: work done once per run, shared rate limits, slots, budgets, queues and stopping a run.

Install

pip install runstate-sdk

Requires Python 3.10 or later. Imports as runstate.

Configure

Set RUNSTATE_API_KEY and RUNSTATE_SPACE_ID in the environment. Optional: RUNSTATE_RUN (the run every process joins, default default) and RUNSTATE_BASE_URL (default https://api.getrunstate.com). The SDK reads them on the first call. To configure in code, use runstate.configure(api_key=..., space_id=..., run=...) or runstate.Runstate(...).

Use

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

Hand work to workers

# review.py, imported by producers and workers
@runstate.queue("review", slots="browser")
def review(doc: dict) -> dict: ...

# producer
score = review(doc)
accepted = review.gather(docs, need=5, accept=lambda result, doc: result["score"] > 0.8)

# worker
runstate.serve()

Decorators work on plain and async def functions. Every module function has an a-prefixed async form: await runstate.astop(), await runstate.aserve(), await review.asend(doc).

Docs

• Reference: https://docs.getrunstate.com/sdk/python/
• Migrating from 0.2: https://docs.getrunstate.com/changelog/
• HTTP API for other languages: https://docs.getrunstate.com/api/