Graph API Reference¶

StateGraphBuilder¶

Builds a state graph with Python callables as nodes.

from flowgentra_ai import StateGraphBuilder

StateGraphBuilder()

Method	Description
`add_node(name, func)`	Add a node. `func(state: SharedState) -> SharedState`
`add_edge(from_node, to_node)`	Add a fixed edge. Use `END` as `to_node` to terminate
`add_conditional_edge(from_node, router)`	Add dynamic routing. `router(state) -> str`
`set_entry_point(name)`	Set the first node to execute
`set_max_steps(n)`	Max execution steps (default: 1000)
`interrupt_before(name)`	Pause before this node (human-in-the-loop)
`interrupt_after(name)`	Pause after this node
`add_subgraph(name, graph)`	Embed a compiled `StateGraph` as a node
`set_checkpointer(base_dir)`	Enable file-based checkpointing
`compile()`	Compile into a runnable `StateGraph`

A compiled, runnable state graph. Created by StateGraphBuilder.compile().

from flowgentra_ai import StateGraph

Method	Returns	Description
`invoke(state)`	`SharedState`	Execute the graph with initial state
`invoke_with_thread(thread_id, state)`	`SharedState`	Execute with checkpointing
`resume(thread_id)`	`SharedState`	Resume an interrupted graph
`resume_with_state(thread_id, updates)`	`SharedState`	Resume with state modifications
`node_names()`	`list[str]`	List all node names
`entry_point()`	`str`	Get entry point node name
`to_dot()`	`str`	Export as Graphviz DOT
`to_mermaid()`	`str`	Export as Mermaid diagram
`to_json()`	`dict`	Export structure as JSON

Sentinel constant for terminating edge targets.

from flowgentra_ai import END

builder.add_edge("last_node", END)

Convenience builder for chat-focused workflows with automatic message accumulation.

from flowgentra_ai import MessageGraphBuilder

MessageGraphBuilder()

Method	Description
`add_node(name, func)`	Add a node. `func(messages: list[Message]) -> list[Message]`
`add_edge(from_node, to_node)`	Add a fixed edge
`set_entry_point(name)`	Set entry point
`compile()`	Compile into a `MessageGraph`

A compiled message-centric graph.

Method	Returns	Description
`invoke(messages)`	`list[Message]`	Execute with initial messages