Metadata-Version: 2.5
Name: statewire
Version: 0.13.0
Summary: Replicate one JSON object over an SSE op stream + a command endpoint
Project-URL: Repository, https://github.com/assistant-ui/harness-sdk
License-Expression: MIT
Requires-Python: <4.0,>=3.12
Requires-Dist: fastapi>=0.115
Requires-Dist: httpx>=0.27
Requires-Dist: pinned<0.16,>=0.15.0
Requires-Dist: statepatch<0.3,>=0.2.0
Provides-Extra: langgraph
Requires-Dist: langchain-core>=0.3; extra == 'langgraph'
Provides-Extra: orjson
Requires-Dist: orjson>=3.9; extra == 'orjson'
Description-Content-Type: text/markdown

# statewire

The Statewire protocol for Python: a `Statewire` host owns one JSON state dict, streams every change to attached clients as ops, and dispatches typed commands back. Part of [harness-sdk](https://github.com/assistant-ui/harness-sdk), with `harness-sdk` and `pinned`.

## Install

```sh
uv add statewire
```

Extras: `uv add "statewire[langgraph]"`, `uv add "statewire[orjson]"`.

## Use

```python
from fastapi import FastAPI
from pinned import PinnedHost, PinnedWorker
from statewire import Statewire, StatewireReject, command

class Counter(Statewire):
    async def lifespan(self):
        self.state = {"count": 0}
        yield

    @command
    async def increment(self):
        if self.state["count"] >= 10:
            raise StatewireReject("the counter is full (max 10)")
        self.state["count"] += 1

worker = PinnedWorker()
app = FastAPI()
app.include_router(worker)
app.include_router(PinnedHost(Counter, worker=worker, namespace="/threads"))
```

## Docs

- Get started: https://github.com/assistant-ui/harness-sdk/blob/main/apps/docs/app/docs/get-started/statewire/page.mdx
- Guide: https://github.com/assistant-ui/harness-sdk/blob/main/apps/docs/app/docs/guides/statewire/host/page.mdx
- Reference: https://github.com/assistant-ui/harness-sdk/blob/main/apps/docs/app/docs/reference/statewire-python/page.mdx
- Spec: https://github.com/assistant-ui/harness-sdk/blob/main/apps/docs/app/docs/spec/statewire-protocol/page.mdx
