Documentation Index
Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
Use this file to discover all available pages before exploring further.
Workflows can maintain typed state for the execution.
Define a state schema
from polos import WorkflowState
class ShoppingCartState(WorkflowState):
items: list[dict] = []
total: float = 0.0
Create a stateful workflow
from polos import workflow, WorkflowContext
@workflow(id="shopping_cart", state_schema=ShoppingCartState)
async def shopping_cart(ctx: WorkflowContext, payload: CartPayload):
if payload.action == "add" and payload.item:
ctx.state.items.append(payload.item.model_dump())
ctx.state.total += payload.item.price * payload.item.quantity
elif payload.action == "remove" and payload.item_id:
for i, item in enumerate(ctx.state.items):
if item.get("id") == payload.item_id:
ctx.state.total -= item["price"] * item["quantity"]
ctx.state.items.pop(i)
break
elif payload.action == "clear":
ctx.state.items = []
ctx.state.total = 0.0
return {"items": ctx.state.items, "total": ctx.state.total}
Initialize with state
# Start workflow with initial state
handle = await polos.invoke(
"shopping_cart",
payload={"action": "add", "item": {...}},
initial_state={"items": [], "total": 0.0},
)
Run it
git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/10-state-persistence
cp .env.example .env # Add your POLOS_PROJECT_ID and API key
uv sync
python main.py
