Skip to content

Event Callbacks

Hook into node lifecycle events for monitoring, logging, and custom behaviors. Grafo provides callbacks at key lifecycle points:

  • on_connect - When a node connects to a child
  • on_disconnect - When a node disconnects from a child
  • on_before_run - Before node execution
  • on_after_run - After node execution
  • on_before_forward - Before forwarding data (during connect())

Callbacks are tuples of (async_function, kwargs_dict). The kwargs dict can contain lambdas for dynamic evaluation.

Basic Usage

async def log_start(node: Node):
    print(f"Starting {node.uuid}")

async def log_complete(node: Node):
    print(f"Completed {node.uuid} in {node.metadata.runtime}s")

node = Node(coroutine=my_task, uuid="task")
node.on_before_run = (
    log_start,
    dict(node=node)
)
node.on_after_run = (
    log_complete,
    dict(node=node)
)

Modifying Tree Structure Dynamically

A node's coroutine can dynamically connect and disconnect other nodes during execution. This enables adaptive tree structures that change based on runtime conditions.

async def process_success(data: str):
    return f"Success: {data}"

async def process_failure(error: str):
    return f"Failure: {error}"

async def router_coroutine(data: str):
    condition = len(data) > 5
    return condition

async def redirect_callback(
    router_node: Node,
    success_node: Node,
    failure_node: Node
):
    condition = router_node.output
    if condition:
        await router_node.redirect([success_node])
    else:
        await router_node.redirect([failure_node])

success_node = Node(coroutine=process_success, uuid="success")
failure_node = Node(coroutine=process_failure, uuid="failure")

router = Node(
    coroutine=router_coroutine,
    uuid="router",
    kwargs=dict(data="input"),
)

router.on_after_run=(redirect_callback, dict(
    router_node=lambda: router,
    success_node=success_node,
    failure_node=failure_node
))

await router.connect(success_node, forward=Node.AUTO)
await router.connect(failure_node, forward=Node.AUTO)

Next Steps