Are node-level callbacks / plugin hooks planned for graph Workflow?

[Koichi73]

Is there a plan to add node-level callbacks (or plugin hooks) for graph Workflow?

I'd like to log each node execution from a Plugin, but NodeRunner doesn't dispatch plugin callbacks, and on_event_callback only shows what a node emitted, not when it ran.

Something like before_node_callback / after_node_callback on BasePlugin would be enough for my case.

Is this planned, or is the intended approach something else?

[llalitkumarrr]

Hello @Koichi73,

Could you please provide a few more details about your question? If possible sharing some details of your use case and a minimal code example would help us better understand and analyze the issue. Also, could you please elaborate on the gaps you found with the current callbacks and plugins?

Additionally, would the BigQueryAgentAnalyticsPlugin be helpful for your use case?

[Koichi73]

@llalitkumarrr
Thanks for the follow-up. My original post framed this as logging, which was too narrow — here is the actual use case.

Use case

A node reads its configuration from an environment variable. The node author hardcodes the variable name:

def notify(node_input: Report) -> Sent:
    webhook = read_env("SLACK_WEBHOOK_URL")  # hardcoded here
    ...

I want to place the same node twice in one graph and have one instance read ALERT_WEBHOOK and the other read INFO_WEBHOOK. The node's input comes from the upstream node's output, so the caller cannot pass this through the node's signature — it has to arrive out of band.

For tools this already works — a plugin callback sets the mapping, and the tool body reads it:

_ENV_MAP: ContextVar[dict[str, str]] = ContextVar("env_map")  # read_env() consults this

class EnvPlugin(BasePlugin):
    async def before_tool_callback(self, *, tool, tool_args, tool_context):
        _ENV_MAP.set({"SLACK_WEBHOOK_URL": "ALERT_WEBHOOK"})

The callback runs in the same task as the tool, so the value reaches the tool body and does not leak into sibling calls.

The gap

For graph Workflow function nodes there is no equivalent place to call that same set(). NodeRunner does not dispatch plugin callbacks, and nodes have no callbacks of their own. The hook I want is the exact counterpart of the one above:

class EnvPlugin(BasePlugin):
    async def before_node_callback(self, *, node, ctx):
        _ENV_MAP.set({"SLACK_WEBHOOK_URL": "ALERT_WEBHOOK"})

On BigQueryAgentAnalyticsPlugin

As far as I can tell it is a BasePlugin implementing the existing callbacks (before_agent_callback, before_tool_callback, on_event_callback, ...), so it has the same gap — node executions are not visible to it. Please correct me if it hooks in somewhere else.

The ask

before_node_callback / after_node_callback on BasePlugin, invoked in the node's execution task. Logging is one use of it; the injection above is the one I actually need.

If there is a different intended way to get caller-supplied configuration into a node body, I would be glad to hear it.