Skip to main content

Quick Start

Build a Hello Flow end to end: install Dex locally, define two Steps with a wait, unblock it with an RPC, and inspect the result in Dex Web.

Choose an SDK once — the preference is stored in this browser (localStorage) and defaults to Python.

TypeScript SDK

Use @superdurable/dex ^0.1.3+. waitFor, execute, and RPC handlers may be async and await Client calls on the same Worker — no sync sidecar. See TypeScript SDK notes.

1. Install Dex server + web​

brew install superdurable/tap/dexcli
brew upgrade superdurable/tap/dexcli   # when you want the latest
dexcli dev --open

dexcli starts Temporal, Dex Server, and Dex Web. Defaults:

ServiceAddress
Dex Webhttp://127.0.0.1:8802
Dex Server127.0.0.1:8801
Temporal Webhttp://127.0.0.1:8233

See the CLI README for flags and persistence options.

2. Hello Flow design​

You will implement:

  1. A Flow with StepA → StepB
  2. StepA uses waitFor with anyOf(timer, channel); StepB has no wait
  3. One Attribute, one Channel, one RPC that updates the attribute and publishes the channel to unblock StepA
  4. A client (and optional REST controller) to start the Flow and invoke the RPC
  5. Inspection in Dex Web
Start → StepA (wait anyOf timer | channel)
              │ channel msg or timer
              ▼
            StepB (execute only) → complete

3. Flow definition​

from datetime import timedelta
from dex import (
    Attribute, Channel, Context, Flow, PersistenceSchema, RPCResult,
    Step, StepDecision, StepList, Timer, Wait, go_to, graceful_complete, rpc,
)

class StepA(Step[str]):
    def __init__(self, unblock: Channel[str]):
        self.unblock = unblock

    def wait_for(self, context: Context, input: str) -> Wait:
        return Wait.any_of(
            self.unblock.for_one(),
            Timer.by_duration(timedelta(seconds=60)),
        )

    def execute(self, context: Context, input: str) -> StepDecision:
        msgs = self.unblock.results(context)
        note = msgs[0] if msgs else "timer"
        return go_to(StepB(), f"{input}:{note}")

class StepB(Step[str]):
    def execute(self, context: Context, input: str) -> StepDecision:
        return graceful_complete(input)

class HelloFlow(Flow[str]):
    status = Attribute("status", str)
    unblock = Channel("unblock", str)

    def get_steps(self) -> StepList[str]:
        return StepList.start_step(StepA(self.unblock)).other_steps(StepB())

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.attributes(self.status).channels(self.unblock)

    @rpc
    def nudge(self, context: Context, message: str) -> RPCResult:
        self.status.set(context, "nudged")
        self.unblock.publish(context, message)
        return RPCResult(output="ok")

4. Start the Flow and invoke the RPC​

Register the Flow on a Worker pointed at dexcli (worker_target / server address 127.0.0.1:8801), then:

# Start
client.start_flow(HelloFlow(), flow_id="hello-1", input="hi")

# Unblock StepA via RPC (also updates attribute + publishes channel)
client.invoke_rpc(flow_id="hello-1", rpc="nudge", input="from-client")

A thin REST controller is optional — useful when a browser or partner system should start / nudge the Flow. The Python basic example exposes /basic routes in the same process as the Worker.

5. View results in Dex Web​

  1. Open http://127.0.0.1:8802
  2. Find Flow hello-1
  3. Observe StepA waiting, then after the RPC, StepB completing and attributes/channel updates

Dex Web Flow list placeholder

TODO: upload Dex Web screenshot

Replace static/img/quick-start/flow-list.svg with a real Dex Web screenshot

Dex Web waiting placeholder

TODO: upload Dex Web screenshot

Replace static/img/quick-start/waiting.svg with StepA waiting

Dex Web completed placeholder

TODO: upload Dex Web screenshot

Replace static/img/quick-start/completed.svg after RPC completes

Next​