Workflow Module¶

`workflow` ¶

Workflow graph types and decorator — the public facade.

The vocabulary itself lives in functualize._types.workflow so that boot and discovery can read declarations without an internal layer importing the public surface; this module is the user-facing re-export, mirroring how functualize.job fronts _types.job_declaration.

Public API

workflow: Decorator for declaring multi-step workflows.
Step: A node that runs a registered job.
Gate: A node that pauses for input.
AgentStep: A node performed by an agent, via a registered executor.
Tool: A job a gate offers, with gate-fixed arguments narrowed away.
Edge: Directed connection between two workflow nodes.
ConditionalEdge: Branching connection based on runtime condition.
END: Sentinel marking workflow termination.
FromStep: A read of this walk's recorded result for one step, used to bind a gate tool's argument (Tool(read_file, allowed=FromStep(...))).

`END = _EndSentinel()` `module-attribute` ¶

`all = ['workflow', 'AgentStep', 'ConditionalEdge', 'Edge', 'END', 'FromStep', 'Gate', 'Loop', 'Notification', 'Notify', 'OnFailure', 'Step', 'Tool', '_EndSentinel']` `module-attribute` ¶

`FromStep(step)` ¶

A read of this walk's recorded result for one step (resolved Q20).

FromStep and :class:FromJob answer different questions, and the difference is why this is a separate name rather than a flag:

============ =============================== ========================= FromJob in a signature FromStep ============ =============================== ========================= may run it yes — it is a dependency edge never resolves in fingerprints, or a scope this scope's steps only upstream may not have run yet has already run ============ =============================== =========================

Used where a value is read from inside a walk that has already produced it — a gate tool's bound argument, and the epilogue body::

Gate(
    name="review",
    awaits=Decision,
    tools=[Tool(read_file, allowed=FromStep("setup-vfs"))],
)

The agent may call read_file, but allowed is fixed to whatever setup-vfs returned in this scope — so the tool is scoped to exactly those files and a call outside them is inexpressible rather than refused.

Why not FromJob here. In a gate-tool binding run=True is not merely unused, it is unmeaningful: the agent invokes the tool on demand, outside the graph's ordering, and the referenced step has already run because the graph ordered it before the gate. Running it from here would execute it outside the walk's step recording, so the walker would not know it happened. Reusing FromJob would have made its natural spelling mean one thing in a signature and another in a binding — the "same declaration, different path" divergence this codebase has repeatedly paid for.

Parameters:

Name	Type	Description	Default
`step`	`str \| Callable[..., Any]`	The step's registered job name, or the decorated function. Positional-only, matching :class:`FromJob` and :class:`~functualize._types.workflow.Tool`.	required

Source code in src/functualize/_types/from_job.py

def __init__(self, step: str | Callable[..., Any], /) -> None:
    if isinstance(step, str):
        if not step.strip():
            raise ValueError("FromStep reference must not be empty")
    elif not callable(step):
        raise TypeError(
            f"FromStep must reference a step name or a callable, "
            f"got {type(step).__name__}"
        )
    object.__setattr__(self, "_step", step)

`slots = ('_step',)` `class-attribute` `instance-attribute` ¶

`step` `property` ¶

The referenced step, as written.

`name` `property` ¶

The referenced step's canonical name.

`setattr(name, value)` ¶

Source code in src/functualize/_types/from_job.py

def __setattr__(self, name: str, value: Any) -> None:
    raise AttributeError("FromStep is immutable")

`repr()` ¶

Source code in src/functualize/_types/from_job.py

def __repr__(self) -> str:
    return f"FromStep({self.name!r})"

`eq(other)` ¶

Source code in src/functualize/_types/from_job.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, FromStep):
        return NotImplemented
    return self.name == other.name

`hash()` ¶

Source code in src/functualize/_types/from_job.py

def __hash__(self) -> int:
    return hash(("FromStep", self.name))

`AgentStep(name, instructions, executor=None, tools=(), requires=frozenset(), time_budget_s=None)` `dataclass` ¶

A workflow node performed by an agent, not by a registered job.

The third node kind, and the first whose execution is not a local function call: it runs somewhere else, it can take arbitrarily long, it can fail in ways a try/except around a callable does not describe, and it can be refused — which is the point. A step that declares something its executor cannot honour refuses before the walk starts, rather than running with the constraint silently dropped.

Nothing about the agent is declared here. The port says what the engine needs; how an implementation talks to a model, an MCP client or a terminal is the implementation's business.

Attributes:

Name	Type	Description
`name`	`str`	Graph key for this node — the address it is recorded under, and the name a refusal reports.
`instructions`	`str`	What the step asks the agent to do. Required, and not allowed to be blank: an agent step with nothing to ask cannot be serviced by anyone.
`executor`	`str \| None`	The registered executor that services this step, by name. `None` means the only registered executor, which is a unique answer — a step declaring `None` where two are registered is refused rather than guessed at, because handing it to one of them would be substituting an executor for the one the step meant.
`tools`	`Sequence[str]`	The tool allowlist this step declares. Declaring any tool implies :attr:`AgentCapability.ENFORCES_TOOL_ALLOWLIST` (:data:`IMPLIED_CAPABILITIES`). An empty sequence declares no constraint, which is not the same statement as "no tools".
`requires`	`frozenset[AgentCapability]`	Capabilities the step needs honoured, on top of the implied ones. Widening a declaration is explicit; narrowing it is not possible — `tools=[…]` implies its capability either way.
`time_budget_s`	`float \| None`	The step's active-time budget in seconds, when it declares one. Declaring one implies :attr:`AgentCapability.PRESERVES_ACTIVE_TIME_BUDGET`: an executor that cannot honour a budget must refuse the step, not ignore it.

`name` `instance-attribute` ¶

`instructions` `instance-attribute` ¶

`executor = None` `class-attribute` `instance-attribute` ¶

`tools = ()` `class-attribute` `instance-attribute` ¶

`requires = frozenset()` `class-attribute` `instance-attribute` ¶

`time_budget_s = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    if not isinstance(self.name, str) or not self.name.strip():
        raise ValueError("AgentStep name must be a non-empty string")
    # A node name is a graph address, so it canonicalizes like every other
    # address (`Step`, `Gate`, `Edge`) — otherwise an edge written from the
    # name the author typed would not find this node.
    object.__setattr__(self, "name", _job_ref_name(self.name))
    if not isinstance(self.instructions, str) or not self.instructions.strip():
        raise ValueError(
            f"AgentStep '{self.name}' must declare non-empty instructions"
        )
    if self.executor is not None and (
        not isinstance(self.executor, str) or not self.executor.strip()
    ):
        raise ValueError(
            f"AgentStep '{self.name}' executor must be a non-empty string or "
            f"None, got {self.executor!r}"
        )
    tools = tuple(self.tools)
    if any(not isinstance(tool, str) or not tool.strip() for tool in tools):
        raise ValueError(
            f"AgentStep '{self.name}' tools must be non-empty strings, "
            f"got {tools!r}"
        )
    object.__setattr__(self, "tools", tools)
    if self.time_budget_s is not None and self.time_budget_s <= 0:
        raise ValueError(
            f"AgentStep '{self.name}' time_budget_s must be positive, "
            f"got {self.time_budget_s!r}"
        )
    # Folded in rather than left to the checker: anything reading
    # `step.requires` gets the set the engine will actually check against,
    # so an implication cannot be missed by reading the declaration.
    object.__setattr__(
        self, "requires", frozenset(self.requires) | implied_capabilities(self)
    )

`ConditionalEdge(source, condition, targets)` `dataclass` ¶

Branching connection where the target depends on a runtime condition.

The chosen key is recorded per scope on first evaluation and replayed on resume, so a walk that pauses cannot resume down a different branch than the one it paused on (§D.7).

Attributes:

Name	Type	Description
`source`	`str`	Name of the source node.
`condition`	`Callable[..., str]`	Callable returning a key into `targets`.
`targets`	`dict[str, str \| _EndSentinel]`	Mapping of condition keys to node names or `END`.

`source` `instance-attribute` ¶

`condition` `instance-attribute` ¶

`targets` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    # Same reason as `Edge`: every endpoint here names a node, and node
    # names are canonical.
    object.__setattr__(self, "source", _job_ref_name(self.source))
    object.__setattr__(
        self,
        "targets",
        {
            key: _job_ref_name(target) if isinstance(target, str) else target
            for key, target in self.targets.items()
        },
    )

`Edge(source, target=(lambda: END)())` `dataclass` ¶

Unconditional directed connection between two workflow nodes.

Attributes:

Name	Type	Description
`source`	`str`	Name of the source node.
`target`	`str \| _EndSentinel`	Name of the target node, or `END` to terminate the walk.

`source` `instance-attribute` ¶

`target = field(default_factory=(lambda: END))` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    # Endpoints name nodes, and node names are canonical — so an edge
    # written `Edge("travel_plan", "book")` must land on the same strings
    # `Step.name` produces, or validation rejects a graph that is correct.
    object.__setattr__(self, "source", _job_ref_name(self.source))
    if isinstance(self.target, str):
        object.__setattr__(self, "target", _job_ref_name(self.target))

`Gate(name, awaits, tools=(), strategy=None)` `dataclass` ¶

A workflow node that pauses for input.

When the walker reaches a gate it records a BLOCKED position, publishes awaits's JSON schema, and stops. Depositing valid input resumes the walk; the deposited payload is available to the epilogue body.

Attributes:

Name	Type	Description
`name`	`str`	Graph key for this node, and the address used to resume it.
`awaits`	`type[BaseModel]`	Pydantic model describing the input the gate requires.
`tools`	`Sequence[ToolRef]`	Jobs an external agent may run while resolving this gate — a permission, not a hint, enforced at MCP dispatch. Each entry is a job name, a decorated function, or a :class:`Tool` when the gate needs to pin some of that job's arguments. Capped at 50: a longer list is a sign the gate is being used as a general-purpose agent handoff rather than an input request. An empty list asks for no restriction.
`strategy`	`str \| None`	Preferred resolution strategy. One of `"resolve"` (config chain), `"prompt"` (interactive surface), `"ai_inbound"` (LLM generation), or `"ai_outbound"` (external AI via MCP). `None` (default) defers to the walker's policy (block unless a CLI flag overrides).

`name` `instance-attribute` ¶

`awaits` `instance-attribute` ¶

`tools = ()` `class-attribute` `instance-attribute` ¶

`strategy = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    if not isinstance(self.name, str) or not self.name.strip():
        raise ValueError("Gate name must be a non-empty string")
    # A gate name is a node address — the string used to resume it — so it
    # canonicalizes like every other address.
    object.__setattr__(self, "name", _job_ref_name(self.name))
    if self.strategy is not None and self.strategy not in _VALID_GATE_STRATEGIES:
        raise ValueError(
            f"Gate strategy must be one of {sorted(_VALID_GATE_STRATEGIES)}, "
            f"got {self.strategy!r}"
        )
    if len(self.tools) > 50:
        raise ValueError(
            f"Gate tools must have at most 50 entries, got {len(self.tools)}"
        )
    object.__setattr__(self, "tools", tuple(self.tools))
    names = [_as_tool(ref).name for ref in self.tools]
    duplicated = {name for name in names if names.count(name) > 1}
    if duplicated:
        # Two entries for one job cannot both be honored — the second's
        # bindings would silently lose to the first at call time.
        raise ValueError(
            f"Gate '{self.name}' lists these tools more than once: "
            f"{', '.join(sorted(duplicated))}"
        )
    from pydantic import BaseModel as _BaseModel

    if not (isinstance(self.awaits, type) and issubclass(self.awaits, _BaseModel)):
        raise TypeError(
            f"Gate awaits must be a BaseModel subclass, got {self.awaits!r}"
        )

`tool_specs()` ¶

Every offered tool, normalized to :class:Tool.

Source code in src/functualize/_types/workflow.py

def tool_specs(self) -> tuple[Tool, ...]:
    """Every offered tool, normalized to :class:`Tool`."""
    return tuple(_as_tool(ref) for ref in self.tools)

`Loop(source, target, max_iterations, condition=None)` `dataclass` ¶

A back-edge that closes a cycle, with the bound that makes it legal.

A cycle declared with ordinary :class:Edge\ s is refused at decoration time (workflow-graph-semantics/T1), because the walk prunes nodes it has already visited and such a graph therefore ran its cycle exactly once, in silence. Loop is the declaration that says how many times, which is the one thing the graph could not previously express.

max_iterations has no default on purpose. Every value anyone would pick as one is wrong for somebody: too low silently truncates work, too high turns a runaway condition into a long outage instead of a quick refusal. Writing the bound is the point of the type.

Attributes:

Name	Type	Description
`source`	`str`	The node the back-edge leaves — the end of the repeated body.
`target`	`str`	The node it returns to — the start of the repeated body.
`max_iterations`	`int`	How many times the body may run in total, counting the first pass. `1` is a body that never repeats, which is legal and occasionally what a caller wants while they are switching it off.
`condition`	`Callable[..., bool] \| None`	Called with the source node's return value; going round again requires a true answer. `None` means "always, until the bound", which is the honest spelling of a fixed repeat.

`source` `instance-attribute` ¶

`target` `instance-attribute` ¶

`max_iterations` `instance-attribute` ¶

`condition = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    # Same canonicalization as `Edge`: endpoints name nodes, and node names
    # are canonical.
    object.__setattr__(self, "source", _job_ref_name(self.source))
    object.__setattr__(self, "target", _job_ref_name(self.target))
    if not isinstance(self.max_iterations, int) or isinstance(
        self.max_iterations, bool
    ):
        raise TypeError(
            f"Loop max_iterations must be an int, got "
            f"{type(self.max_iterations).__name__}"
        )
    if self.max_iterations < 1:
        raise ValueError(
            f"Loop max_iterations must be at least 1, got "
            f"{self.max_iterations}. A bound of 0 would declare a body that "
            f"cannot run, which is a graph with the edge deleted."
        )

`Notification(to, scope_id, workflow, status, node=None)` `dataclass` ¶

What a notifier is handed. A target, and what happened.

Five fields and no envelope. There is no message id, no correlation key, no priority and no retry count, because each of those is the first field of a broker and this is not one (N8). A provider that needs an id makes one; a provider that needs a retry owns it.

Attributes:

Name	Type	Description
`to`	`str`	The declaration's `to`, verbatim. Never parsed here.
`scope_id`	`str`	The walk this is about.
`workflow`	`str \| None`	The job name, when the scope records one.
`status`	`str`	The scope status that fired it.
`node`	`str \| None`	Where the walk stopped, when it stopped somewhere — the failed node, or the gate it is blocked at. None for a walk that finished.

`to` `instance-attribute` ¶

`scope_id` `instance-attribute` ¶

`workflow` `instance-attribute` ¶

`status` `instance-attribute` ¶

`node = None` `class-attribute` `instance-attribute` ¶

`Notify(on, to, provider=None)` `dataclass` ¶

Tell somebody when a walk ends in a given state.

An effect, and it rides the same outbox rule as Step(effecting=True): the record that it fired is committed before the provider is called, so a crash can lose a notification but can never send it twice. At-most-once, in the direction that matters — a resumed workflow must not page the on-call again for a failure they have already seen.

Not a bus and not a broker. to is opaque to the engine: it is handed to the provider verbatim and nothing here parses, matches or routes on it. The moment to becomes load-bearing routing, you own a broker — and then retries, fan-out and a dead-letter queue follow, none of which this is. A target, and an effect.

Attributes:

Name	Type	Description
`on`	`str`	The scope status to fire on — one of `_VALID_NOTIFY_STATES`.
`to`	`str`	Where the notification goes, in whatever spelling the provider understands. An address, a channel, a task list, a URL.
`provider`	`str \| None`	Which registered notifier delivers it. None means the single registered one, and is an error when there is more than one — a notification that silently picked a deliverer would be the worst kind of working.

`on` `instance-attribute` ¶

`to` `instance-attribute` ¶

`provider = None` `class-attribute` `instance-attribute` ¶

`key` `property` ¶

What "this notification already fired" is recorded under.

The declaration's own content, not its position in a list: a workflow that gains a second Notify must not make the first one fire again by shifting its index.

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    if self.on not in _VALID_NOTIFY_STATES:
        raise ValueError(
            f"Notify `on` must be one of {sorted(_VALID_NOTIFY_STATES)}, "
            f"got {self.on!r}"
        )
    if not isinstance(self.to, str) or not self.to.strip():
        raise ValueError("Notify `to` must be a non-empty string")
    if self.provider is not None and not self.provider.strip():
        raise ValueError("Notify `provider` must be a non-empty string or None")

`OnFailure(source, target=(lambda: END)(), when=None)` `dataclass` ¶

Where control goes when a step raises.

Without one, a raising step stops the walk and the scope is marked failed — that is the behaviour every workflow has today and OnFailure does not change it. It adds a declared alternative, and only for the node it names.

Attributes:

Name	Type	Description
`source`	`str`	The node whose failure this routes.
`target`	`str \| _EndSentinel`	Where to continue, or `END` to finish the walk without marking it failed — a cleanup path that succeeds is a success.
`when`	`Callable[[BaseException], bool] \| None`	Called with the exception the step raised; routing requires a true answer. `None` routes every failure, which is the honest spelling of a catch-all.

The chosen route is recorded and read back on replay, never re-evaluated — the property ConditionalEdge already has, for a sharper reason. _choice_for puts it as "calling it and discarding the answer would still run whatever side effects it has", and a failure predicate is exactly the kind that pages somebody.

`source` `instance-attribute` ¶

`target = field(default_factory=(lambda: END))` `class-attribute` `instance-attribute` ¶

`when = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    object.__setattr__(self, "source", _job_ref_name(self.source))
    if isinstance(self.target, str):
        object.__setattr__(self, "target", _job_ref_name(self.target))

`Step(job, effecting=False)` `dataclass` ¶

A workflow node that runs a registered job.

Attributes:

Name	Type	Description
`job`	`str \| Callable[..., Any]`	The job to run — its registered name, or the decorated function itself. Nothing else: a step does not define behavior, it points at behavior that is already declared and independently runnable.
`effecting`	`bool`	This step does something the world remembers — charges a card, sends a mail, files a ticket. It must run exactly once across a crash and a resume. Default `False`, and that default is the honest one: the framework cannot tell an effecting step from a pure one by looking at it, and guessing wrong in this direction re-runs a refund. A step that says nothing is replayed, which is the behaviour every workflow has had until now. What the flag buys is an outbox: the step's completion record is committed in the same locked batch as the walk's position, so a crash can leave the effect done and the record absent only if it lands between the effect and a write that is itself atomic. See `_engine/frontier.complete`. It does not make the effect itself transactional — nothing here can. It makes the record of the effect commit with the walk's progress, which is what a resume reads to decide whether to run the step again.

`job` `instance-attribute` ¶

`effecting = False` `class-attribute` `instance-attribute` ¶

`name` `property` ¶

The graph key for this node (the referenced job's name).

`__post_init__()` ¶

Source code in src/functualize/_types/workflow.py

def __post_init__(self) -> None:
    if isinstance(self.job, str):
        if not self.job.strip():
            raise ValueError("Step job reference must not be empty")
        return
    if not callable(self.job):
        raise TypeError(
            f"Step job must be a registered job name or a callable, "
            f"got {type(self.job).__name__}"
        )

`Tool(job, /, **bound)` ¶

A job offered at a gate, with some of its arguments fixed by the gate.

A tool is a registered job — there is no second kind of callable thing in this framework. What a Tool adds is narrowing: arguments pinned here are removed from the schema the agent is shown, so a forbidden call is not merely refused, it is inexpressible::

Gate(
    name="approval",
    awaits=RefundDecision,
    tools=[order_history, Tool(issue_refund, cap_cents=5_000)],
)

The agent sees issue_refund(order_id, amount_cents) and never learns cap_cents exists. Passing it anyway is an error rather than a silent override: an agent that believes it set a value and did not is worse off than one told no.

This has to live at the gate rather than on the job, because it is a property of the usage. The same issue_refund may be capped at $50 in a self-serve workflow and uncapped in a supervisor one, and its signature cannot know which workflow is calling.

A bare job reference stays legal wherever no narrowing is wanted — tools=[order_history] is not worth a wrapper.

Parameters:

Name	Type	Description	Default
`job`	`str \| Callable[..., Any]`	The job to offer — its registered name or the decorated function. Positional-only, so a job with a parameter named `job` can still have it bound.	required
`**bound`	`Any`	Arguments fixed by this gate.	`{}`

Source code in src/functualize/_types/workflow.py

def __init__(self, job: str | Callable[..., Any], /, **bound: Any) -> None:
    if isinstance(job, str):
        if not job.strip():
            raise ValueError("Tool job reference must not be empty")
    elif not callable(job):
        raise TypeError(
            f"Tool job must be a registered job name or a callable, "
            f"got {type(job).__name__}"
        )
    from functualize._types.from_job import FromJob

    for arg, value in bound.items():
        if isinstance(value, FromJob):
            raise TypeError(
                f"Tool({_job_ref_name(job)!r}, {arg}=FromJob(...)) is not "
                f"valid — use FromStep({value.name!r}) instead. A gate "
                f"tool's argument is read from this walk's recorded "
                f"results and can never trigger a job: the agent calls "
                f"the tool on demand, outside the graph's ordering, and "
                f"the step has already run because the graph ordered it "
                f"before the gate."
            )
    object.__setattr__(self, "_job", job)
    object.__setattr__(self, "_bound", dict(bound))

`slots = ('_bound', '_job')` `class-attribute` `instance-attribute` ¶

`job` `property` ¶

The referenced job.

`bound` `property` ¶

Arguments this gate fixes, as a copy.

`name` `property` ¶

The referenced job's name.

`setattr(name, value)` ¶

Source code in src/functualize/_types/workflow.py

def __setattr__(self, name: str, value: Any) -> None:
    raise AttributeError("Tool is immutable")

`repr()` ¶

Source code in src/functualize/_types/workflow.py

def __repr__(self) -> str:
    if not self._bound:
        return f"Tool({self.name!r})"
    pinned = ", ".join(f"{k}={v!r}" for k, v in self._bound.items())
    return f"Tool({self.name!r}, {pinned})"

`eq(other)` ¶

Source code in src/functualize/_types/workflow.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, Tool):
        return NotImplemented
    return self.name == other.name and self.bound == other.bound

`hash()` ¶

Source code in src/functualize/_types/workflow.py

def __hash__(self) -> int:
    return hash((self.name, tuple(sorted(self._bound))))

`workflow(*, steps, edges, notify=())` ¶

Decorator registering a function as a declarative workflow.

Validates the workflow graph at decoration time and attaches the frozen declaration to the decorated function. The function's own body is the workflow's epilogue: it runs when the walk reaches END.

Parameters:

Name	Type	Description	Default
`steps`	`Sequence[Step \| Gate \| AgentStep]`	Workflow nodes — `Step` (runs a registered job), `Gate` (pauses for input), or `AgentStep` (delegates to a registered agent executor).	required
`edges`	`Sequence[Edge \| ConditionalEdge]`	List of Edge or ConditionalEdge objects defining connections.	required
`notify`	`Sequence[Notify]`	Notifications to fire when the walk ends in a declared state. A separate argument rather than an entry in `edges`: a `Notify` has no source and no target in the graph — it is about the walk's outcome, not about control moving between nodes — and putting it there would make every edge consumer test for a kind that has neither.	`()`

Returns:

Type	Description
`Callable[[Callable[..., Any]], Callable[..., Any]]`	A decorator that attaches the workflow definition to the function.
`Callable[[Callable[..., Any]], Callable[..., Any]]`	Identity-preserving: `decorated is original` always holds.

Raises:

Type	Description
`TypeError`	If a list entry is not a workflow node or edge type.
`ValueError`	If the graph contains duplicate node names or unknown node references in edges.

Source code in src/functualize/workflow/_decorator.py

def workflow(
    *,
    steps: Sequence[Step | Gate | AgentStep],
    edges: Sequence[Edge | ConditionalEdge],
    notify: Sequence[Notify] = (),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator registering a function as a declarative workflow.

    Validates the workflow graph at decoration time and attaches the frozen
    declaration to the decorated function. The function's own body is the
    workflow's epilogue: it runs when the walk reaches ``END``.

    Args:
        steps: Workflow nodes — `Step` (runs a registered job), `Gate`
            (pauses for input), or `AgentStep` (delegates to a registered
            agent executor).
        edges: List of Edge or ConditionalEdge objects defining connections.
        notify: Notifications to fire when the walk ends in a declared state.
            A separate argument rather than an entry in ``edges``: a `Notify`
            has no source and no target in the graph — it is about the walk's
            outcome, not about control moving between nodes — and putting it
            there would make every edge consumer test for a kind that has
            neither.

    Returns:
        A decorator that attaches the workflow definition to the function.
        Identity-preserving: ``decorated is original`` always holds.

    Raises:
        TypeError: If a list entry is not a workflow node or edge type.
        ValueError: If the graph contains duplicate node names or unknown
            node references in edges.
    """
    _validate_workflow_graph(steps, edges, notify)
    declaration = WorkflowDeclaration(
        nodes=tuple(steps), edges=tuple(edges), notify=tuple(notify)
    )

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        fn.__functualize_workflow__ = declaration  # type: ignore[attr-defined]
        return fn

    return decorator

Overview¶

The functualize.workflow module provides graph-based workflow execution with multi-step job orchestration, branching logic, and conditional execution paths.

Module location: src/functualize/workflow/

from functualize.workflow import (
    workflow,
    Step,
    Gate,
    AgentStep,
    Edge,
    ConditionalEdge,
    OnFailure,
    END,
    Notify,
    Notification,
)

`@workflow` Decorator¶

The @workflow decorator transforms a function into a multi-step workflow. The decorated function should return a workflow graph description.

from functualize.workflow import workflow, Step, Edge, END

@workflow(
    steps=[
        Step(step1),
        Step(step2),
    ],
    edges=[
        Edge(source="step1", target="step2"),
        Edge(source="step2", target=END),
    ],
)
def multi_step_job(config, rc):
    """Define a workflow with multiple steps."""
    rc.log("Workflow complete")

`Step`¶

A workflow node that runs a registered job. A step wraps a reference to a job — its registered name or the decorated function itself. It carries no behavior of its own; the job's @job declaration (deps, guards, caching) is what runs.

from functualize.workflow import Step

step = Step("validate_data")
step = Step(validate_data)  # callable reference

Attribute	Type	Description
`job`	`str \\| Callable`	The registered job name or decorated function
`name` (property)	`str`	Graph key for this node (the referenced job's normalized name)

Note

Step only takes a job argument. There are no action or name constructor parameters — the job declaration already describes what runs.

`AgentStep`¶

A workflow node performed by an agent, through a registered executor, rather than by a registered job. The third node kind, and the first whose execution is not a local function call: it runs somewhere else, can take arbitrarily long, and can be refused.

from functualize.workflow import AgentStep

AgentStep(
    "draft",
    instructions="Draft the release notes from the fetched changelog.",
    tools=["read_file"],
    time_budget_s=120,
)

Attribute	Type	Description
`name`	`str`	Graph key for this node — the address it is recorded under, and the name a refusal reports
`instructions`	`str`	What the step asks the agent to do. Required, and may not be blank
`executor`	`str \\| None`	The registered executor that services this step. `None` means the only registered executor; with two registered, a step naming none is refused rather than guessed at
`tools`	`Sequence[str]`	The tool allowlist. Declaring any tool implies `enforces_tool_allowlist`. An empty sequence declares no constraint, which is not the same statement as "no tools"
`requires`	`frozenset[AgentCapability]`	Capabilities needed on top of the implied ones. Widening is explicit; narrowing is not possible
`time_budget_s`	`float \\| None`	Active-time budget in seconds. Declaring one implies `preserves_active_time_budget`

Refused, never degraded

An executor that cannot honour a required capability makes the step refuse before the walk starts. Running it with the constraint dropped would produce a workflow that appears to have restricted tools it left wide open.

See Workflows → Agent steps for executors, capabilities and a worked graph.

`Edge`¶

Represents a directed connection between two workflow steps.

from functualize.workflow import Edge, END

# source and target name workflow nodes
edge = Edge(source="start", target="process")
final_edge = Edge(source="process", target=END)

Attribute	Type	Description
`source`	`str`	Name of the source node
`target`	`str \\| END`	Name of the target node, or `END` to terminate

Note

Edge takes source and target as keyword arguments naming nodes, not Step objects.

`ConditionalEdge`¶

Represents a branching connection based on a runtime condition.

from functualize.workflow import ConditionalEdge, END

# Maps condition keys to target nodes
conditional = ConditionalEdge(
    source="check",
    condition=lambda: "success" if all_green() else "failure",
    targets={
        "success": "success_handler",
        "failure": "failure_handler",
    },
)

Attribute	Type	Description
`source`	`str`	Name of the source node
`condition`	`Callable`	Returns a key into `targets`
`targets`	`dict[str, str \\| END]`	Mapping of condition keys to node names or `END`

`OnFailure`¶

Where control goes when a step raises. Without one, a raising step stops the walk and the scope is marked failed — the behaviour every workflow has by default, and OnFailure does not change it. It adds a declared alternative, and only for the node it names.

from functualize.workflow import OnFailure, END

OnFailure(
    source="deploy",
    target="rollback",
    when=lambda exc: isinstance(exc, DeploymentError),
)

Attribute	Type	Description
`source`	`str`	The node whose failure this routes.
`target`	`str \\| END`	Where to continue, or `END` to finish the walk without marking it failed — a cleanup path that succeeds is a success.
`when`	`Callable[[BaseException], bool] \\| None`	Called with the exception the step raised; routing requires a true answer. `None` routes every failure — the honest spelling of a catch-all.

The chosen route is recorded and read back on replay

Never re-evaluated on resume, for the same reason ConditionalEdge's choice isn't: calling when again and discarding the answer would still run whatever side effects it has, and a failure predicate is exactly the kind that pages somebody.

`END`¶

A sentinel value marking the end of a workflow. Used in workflow graph definitions to indicate the termination point.

from functualize.workflow import workflow, Step, Edge, END

@workflow(
    steps=[Step("work")],
    edges=[Edge(source="work", target=END)],
)
def simple_workflow(config, rc):
    rc.log("Work done")

`Notification`¶

What a registered notifier is handed when a Notify declaration fires — a target, and what happened. Five fields and no envelope: there is no message id, correlation key, priority, or retry count, because Notify declares a target and an effect, not a broker.

from functualize.workflow import Notification

@dataclass(frozen=True)
class Notification:
    to: str
    scope_id: str
    workflow: str | None
    status: str
    node: str | None = None

Attribute	Type	Description
`to`	`str`	The `Notify` declaration's `to`, verbatim — never parsed by the engine.
`scope_id`	`str`	The workflow walk this notification is about.
`workflow`	`str \\| None`	The job name, when the scope records one.
`status`	`str`	The scope status that fired the notification.
`node`	`str \\| None`	Where the walk stopped, when it stopped somewhere — the failed node, or the gate it is blocked at. `None` for a walk that finished.

A plugin implementing a notifier receives one Notification per fired Notify(on=..., to=..., provider=...) declaration on @workflow(..., notify=[...]). Delivery is at-most-once: the record that a notification fired is committed before the provider is called, so a crash can lose a notification but never send it twice.

Internal Location¶

Workflow types live in functualize._types.workflow:

_types/workflow.py — Step, Gate, AgentStep, Edge, ConditionalEdge, OnFailure, Loop, Notify, Notification, END, WorkflowShape, WorkflowDeclaration
workflow/_decorator.py — @workflow decorator and execution engine
workflow/_validation.py — Graph validation and cycle detection

Internal API

Modules under functualize.workflow._* are implementation details. Import types from functualize.workflow instead.

Workflow Module¶

workflow ¶

END = _EndSentinel() module-attribute ¶

__all__ = ['workflow', 'AgentStep', 'ConditionalEdge', 'Edge', 'END', 'FromStep', 'Gate', 'Loop', 'Notification', 'Notify', 'OnFailure', 'Step', 'Tool', '_EndSentinel'] module-attribute ¶

FromStep(step) ¶

__slots__ = ('_step',) class-attribute instance-attribute ¶

step property ¶

name property ¶

__setattr__(name, value) ¶

__repr__() ¶

__eq__(other) ¶

__hash__() ¶

AgentStep(name, instructions, executor=None, tools=(), requires=frozenset(), time_budget_s=None) dataclass ¶

name instance-attribute ¶

instructions instance-attribute ¶

executor = None class-attribute instance-attribute ¶

tools = () class-attribute instance-attribute ¶

requires = frozenset() class-attribute instance-attribute ¶

time_budget_s = None class-attribute instance-attribute ¶

__post_init__() ¶

ConditionalEdge(source, condition, targets) dataclass ¶

source instance-attribute ¶

condition instance-attribute ¶

targets instance-attribute ¶

__post_init__() ¶

Edge(source, target=(lambda: END)()) dataclass ¶

source instance-attribute ¶

target = field(default_factory=(lambda: END)) class-attribute instance-attribute ¶

__post_init__() ¶

Gate(name, awaits, tools=(), strategy=None) dataclass ¶

name instance-attribute ¶

awaits instance-attribute ¶

tools = () class-attribute instance-attribute ¶

strategy = None class-attribute instance-attribute ¶

__post_init__() ¶

tool_specs() ¶

Loop(source, target, max_iterations, condition=None) dataclass ¶

source instance-attribute ¶

target instance-attribute ¶

max_iterations instance-attribute ¶

condition = None class-attribute instance-attribute ¶

__post_init__() ¶

Notification(to, scope_id, workflow, status, node=None) dataclass ¶

to instance-attribute ¶

scope_id instance-attribute ¶

workflow instance-attribute ¶

status instance-attribute ¶

node = None class-attribute instance-attribute ¶

Notify(on, to, provider=None) dataclass ¶

on instance-attribute ¶

to instance-attribute ¶

provider = None class-attribute instance-attribute ¶

key property ¶

__post_init__() ¶

OnFailure(source, target=(lambda: END)(), when=None) dataclass ¶

source instance-attribute ¶

target = field(default_factory=(lambda: END)) class-attribute instance-attribute ¶

when = None class-attribute instance-attribute ¶

__post_init__() ¶

Step(job, effecting=False) dataclass ¶

job instance-attribute ¶

effecting = False class-attribute instance-attribute ¶

name property ¶

__post_init__() ¶

Tool(job, /, **bound) ¶

__slots__ = ('_bound', '_job') class-attribute instance-attribute ¶

job property ¶

bound property ¶

name property ¶

__setattr__(name, value) ¶

__repr__() ¶

__eq__(other) ¶

__hash__() ¶

workflow(*, steps, edges, notify=()) ¶

Overview¶

@workflow Decorator¶

Step¶

AgentStep¶

Edge¶

ConditionalEdge¶

`workflow` ¶

`END = _EndSentinel()` `module-attribute` ¶

`all = ['workflow', 'AgentStep', 'ConditionalEdge', 'Edge', 'END', 'FromStep', 'Gate', 'Loop', 'Notification', 'Notify', 'OnFailure', 'Step', 'Tool', '_EndSentinel']` `module-attribute` ¶

`FromStep(step)` ¶

`slots = ('_step',)` `class-attribute` `instance-attribute` ¶

`step` `property` ¶

`name` `property` ¶

`setattr(name, value)` ¶

`repr()` ¶

`eq(other)` ¶

`hash()` ¶

`AgentStep(name, instructions, executor=None, tools=(), requires=frozenset(), time_budget_s=None)` `dataclass` ¶

`name` `instance-attribute` ¶

`instructions` `instance-attribute` ¶

`executor = None` `class-attribute` `instance-attribute` ¶

`tools = ()` `class-attribute` `instance-attribute` ¶

`requires = frozenset()` `class-attribute` `instance-attribute` ¶

`time_budget_s = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

`ConditionalEdge(source, condition, targets)` `dataclass` ¶

`source` `instance-attribute` ¶

`condition` `instance-attribute` ¶

`targets` `instance-attribute` ¶

`__post_init__()` ¶

`Edge(source, target=(lambda: END)())` `dataclass` ¶

`source` `instance-attribute` ¶

`target = field(default_factory=(lambda: END))` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

`Gate(name, awaits, tools=(), strategy=None)` `dataclass` ¶

`name` `instance-attribute` ¶

`awaits` `instance-attribute` ¶

`tools = ()` `class-attribute` `instance-attribute` ¶

`strategy = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

`tool_specs()` ¶

`Loop(source, target, max_iterations, condition=None)` `dataclass` ¶

`source` `instance-attribute` ¶

`target` `instance-attribute` ¶

`max_iterations` `instance-attribute` ¶

`condition = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

`Notification(to, scope_id, workflow, status, node=None)` `dataclass` ¶

`to` `instance-attribute` ¶

`scope_id` `instance-attribute` ¶

`workflow` `instance-attribute` ¶

`status` `instance-attribute` ¶

`node = None` `class-attribute` `instance-attribute` ¶

`Notify(on, to, provider=None)` `dataclass` ¶

`on` `instance-attribute` ¶

`to` `instance-attribute` ¶

`provider = None` `class-attribute` `instance-attribute` ¶

`key` `property` ¶

`__post_init__()` ¶

`OnFailure(source, target=(lambda: END)(), when=None)` `dataclass` ¶

`source` `instance-attribute` ¶

`target = field(default_factory=(lambda: END))` `class-attribute` `instance-attribute` ¶

`when = None` `class-attribute` `instance-attribute` ¶

`__post_init__()` ¶

`Step(job, effecting=False)` `dataclass` ¶

`job` `instance-attribute` ¶

`effecting = False` `class-attribute` `instance-attribute` ¶

`name` `property` ¶

`__post_init__()` ¶

`Tool(job, /, **bound)` ¶

`slots = ('_bound', '_job')` `class-attribute` `instance-attribute` ¶

`job` `property` ¶

`bound` `property` ¶

`name` `property` ¶

`setattr(name, value)` ¶

`repr()` ¶

`eq(other)` ¶

`hash()` ¶

`workflow(*, steps, edges, notify=())` ¶

`@workflow` Decorator¶

`Step`¶

`AgentStep`¶

`Edge`¶

`ConditionalEdge`¶

`OnFailure`¶

`END`¶