Flow & Workflow Engine

Use Flow when in-process Python methods should form a graph (@start / @listen / @router). Use WorkflowEngine when you want a configuration-driven graph. Neither replaces a Near chat turn — that is AgentRuntime.

Worked example: a three-node class

Scene. Fetch → analyze → route to publish or review, inside your own process.

1. Subclass Flow, mark the entry with @start, the next method with @listen, the branch with @router.
2. Call kickoff() / kickoff_async(). This never starts Near and never writes messages.json.
3. If you need pause/resume, use ExecutionPlan (to_mermaid(), pause(), resume()).

What you should see. In a script: returned state. In docs: a pipeline you can point at. In Near: nothing unless you wrap the Flow yourself.

Overview

Flow system

Flow base class

• Generic over state type T (dict or Pydantic model).
• Entry points: kickoff() (sync) and kickoff_async().

Decorators

Example: simple data pipeline

1from agenticx.flow import Flow, start, listen, router
2
3class DataPipeline(Flow[dict]):
4    @start()
5    def fetch_data(self):
6        return {"data": [1, 2, 3]}
7
8    @listen("fetch_data")
9    def process_data(self, result):
10        return {"processed": [x * 2 for x in result["data"]]}
11
12    @router("process_data")
13    def branch(self, result):
14        values = result.get("processed", [])
15        return "NONEMPTY" if values else "EMPTY"
16
17    @listen("NONEMPTY")
18    def on_success(self):
19        self.state["status"] = "ok"
20
21    @listen("EMPTY")
22    def on_empty(self):
23        self.state["status"] = "empty"
24
25flow = DataPipeline()
26flow.kickoff()

Execution plan

ExecutionPlan

ExecutionStage and Subtask

• ExecutionStage: name, subtasks, status
• Subtask: id, name, query, status, result/error

Core capabilities

• ExecutionPlan.to_mermaid(): returns a Mermaid diagram string
• pause() / resume() / reset_node(subtask_id)
• overall_progress: fraction of subtasks completed

Graph engine

Module: agenticx.core.graph

1from agenticx.core.graph import Graph, BaseNode, End, GraphRunContext
2
3class StartNode(BaseNode):
4    async def run(self, ctx: GraphRunContext) -> ProcessNode | End[str]:
5        # Do work
6        return ProcessNode()
7
8class ProcessNode(BaseNode):
9    async def run(self, ctx: GraphRunContext) -> End[str]:
10        return End(result="done")
11
12graph = Graph(nodes=[StartNode, ProcessNode])
13result = await graph.run(StartNode())

WorkflowEngine

Module: agenticx.core.workflow_engine

• WorkflowEngine runs a Workflow model or WorkflowGraph
• Concurrency: entry nodes and downstream fan-out run via asyncio.gather
• Condition routing: edges may encode JSON condition_config
• Observability: ExecutionContext holds event_log

Quick reference imports

1# Flow
2from agenticx.flow import (
3    Flow,
4    FlowState,
5    start,
6    listen,
7    router,
8    or_,
9    and_,
10)
11
12# Execution plan
13from agenticx.flow import (
14    ExecutionPlan,
15    ExecutionStage,
16    Subtask,
17    ExecutionPlanManager,
18)
19
20# Graph
21from agenticx.core.graph import Graph, BaseNode, End, GraphRunContext
22
23# Workflow
24from agenticx.core.workflow_engine import WorkflowEngine, WorkflowGraph