Your first flow¶

Build a three-node flow that classifies a support ticket, then either looks up a policy with a tool or escalates. Ten minutes. The whole program is examples/first_flow.py; the test suite runs it.

flowchart LR
  classify -- refund --> policy["policy<br/>Call: lookup_policy"]
  classify -- other --> escalate

1. The state¶

A flow passes one pydantic model from node to node. Each node returns the fields it changes.

class Ticket(BaseModel):
    text: str
    category: str = ""
    reply: str = ""

2. The nodes¶

A node is a plain function of the state (sync or async). It returns a mapping of updates, or None.

def classify(ticket: Ticket) -> dict[str, str]:
    return {"category": "refund" if "refund" in ticket.text.lower() else "other"}


def escalate(ticket: Ticket) -> dict[str, str]:
    return {"reply": "A person will contact you today."}

3. The flow¶

flow = (
    Flow("support", state=Ticket)
    .node("classify", classify)
    .node("policy", Call("lookup_policy", args=lambda t: {"category": t.category}, into="reply"))
    .node("escalate", escalate)
    .route(
        "classify",
        Route(
            lambda t: "policy" if t.category == "refund" else "escalate",
            targets=["policy", "escalate"],
        ),
    )
    .end("policy", "escalate")
)

Line	Effect
`Flow("support", state=Ticket)`	a flow over `Ticket`; names must be identifiers
`.node("classify", classify)`	a function node; the first node added is the start
`Call("lookup_policy", args=…, into="reply")`	runs a registered tool through the executor and writes its value to `reply`
`.route("classify", Route(fn, targets=[…]))`	`fn` returns the name of the next node
`.end("policy", "escalate")`	end nodes have no exit

The flow is checked when it runs (or call flow.validate()): unknown nodes, unreachable nodes, missing exits and unbounded loops are all reported together.

4. Run it¶

runner = FlowRunner(Executor(registry))
result = run_sync(runner.run(flow, {"text": "I want a refund"}))
print(result.status, "|", result.state.reply)
print(dict(result.visits))

other = run_sync(runner.run(flow, {"text": "My parcel is late"}))
print(other.state.category, "|", other.state.reply)

python examples/first_flow.py

done | Our refund policy: full refund within 30 days.
{'classify': 1, 'policy': 1}
other | A person will contact you today.

FlowRunner(executor) runs every tool call through the executor, so the tool's policies (timeouts, retries, approval, cache, ledger) apply. run is async; run_sync calls it from plain code.

Next¶

To	Read
fan out over a list, join, loop with a cap, pause for approval	Flows
see how the runner fits with the executor	Executor and flows
control tool calls inside nodes	Executor