Metadata-Version: 2.4
Name: runstate-sdk
Version: 0.3.0
Summary: runstate SDK: wrap a function once, runstate coordinates it across your agents
License: MIT
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-asyncio; extra == "test"
Requires-Dist: anyio; extra == "test"

# runstate (Python SDK)

Python SDK for [runstate](https://getrunstate.com), 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

```sh
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

```python
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

```python
# 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/
