How To Create a Node

How To Create A Node

Purpose

Use a Node when you want to define one executable unit of behavior inside GraphK.

A Node is responsible for:

• reading from the active Session
• writing results back into the active Session
• exposing a small amount of metadata through info()
• yielding execution steps from step()

What Must Be Implemented?

graphk.Node is abstract. A custom node must implement:

• ping()
• info()
• step()

ping()

Purpose:

• report whether the node is structurally healthy and ready to execute

Typical pattern:

• return True for simple nodes
• return a validation result for nodes that depend on callables, resources, or configuration

info()

Purpose:

• expose lightweight metadata about the node

Typical pattern:

• return a dictionary with at least a stable human-readable name

Example:

def info(self) -> dict:
    return {"name": self._name}

step()

Purpose:

• define the actual executable behavior of the node

Important rule:

• step() must be a generator

Typical pattern:

• read current values from self.session
• compute one change
• write updated values with self.session.set(...)
• yield each step result

Minimal Example

import graphk


class IncrementNode(graphk.Node):

    def __init__(self, name: str, increment: int = 1, **kwargs) -> None:
        super().__init__(**kwargs)
        self._name = name
        self._increment = increment

    def ping(self) -> bool:
        return True

    def info(self) -> dict:
        return {"name": self._name}

    def step(self):
        current = self.session.get("counter", 0)
        current += self._increment
        self.session.set("counter", current)
        yield current

How Is A Node Used?

The low-level lifecycle is:

1. create a Session
2. call session.start()
3. create the node
4. call node.start(session)
5. iterate or step the node
6. call node.complete()

Example:

import graphk


session = graphk.Session(counter=0)
session.start()

node = IncrementNode("A", increment=2)
node.start(session)

for value in node.step():
    print(value)

node.complete()
print(session.to_dict())

What Should A Node Write To?

Write execution results into self.session, not into self.context.

Use Context for:

• scoped configuration
• stage names
• routing values
• execution settings

Use Session for:

• runtime state
• outputs
• accumulated results
• final response content

Entry And Exit Policies

Every node already carries:

• context
• entry_policy
• exit_policy

You do not need to implement policy logic inside step() unless the node has its own internal rule separate from GraphK policy handling.

In normal package usage, entry and exit validation are enforced by Runner.

Common Mistakes

• Do not return a plain value from step(); it must yield.
• Do not treat Context as the final result store.
• Do not forget info(); pipelines and logs often rely on node metadata.
• Do not bypass Session.get() and Session.set() when scoped access matters.

Learn More

Recommended references:

• tests/node/
• tests/sequence/test_sequence_1.py
• examples/example_4.py
• src/graphk/pipes/demos.py

The demo nodes in src/graphk/pipes/demos.py are especially useful as reusable examples of how simple nodes are implemented in practice.