LangGraph: wrap_node

Wraps a LangGraph node so that, after it runs, each event you extract from its result is sent to Aether with emit_delta. The node’s result is returned unchanged.

When to use it

Use wrap_node to report Deltas from LangGraph nodes without adding reporting code inside each node.

Signature

def wrap_node(
    fn,
    *,
    extract,
    workspace_revision,
    api_key: str,
) -> Callable[[Any], Awaitable[Any]]

Argument	Type	Description
`fn`	`(state) -> result`	The node to wrap. Sync or async.
`extract`	`(state, result) -> list[tuple[str, dict]]`	Returns the events to report, as `(event, payload)` pairs. Return `[]` to report nothing.
`workspace_revision`	`(state) -> str`	Returns the workspace revision from the state.
`api_key`	`str`	API key of the system that owns the node.

What you implement

wrap_node only sends what you give it. You write extract and workspace_revision for your use case.

`extract`

You decide, for each node:

Which events to report, and their names, e.g. "dependency.scan.completed".
What goes in each payload. Only what extract returns is sent.
How many Deltas to send: one, several (e.g. one per change applied), or none ([]), e.g. when the node produced nothing worth reporting.

Write extract defensively. An error raised inside it fails the node. For example, use result.get("findings", []), not result["findings"].

`workspace_revision`

You decide what “revision” means for your workspace: a git commit SHA, a document version, a run id, and so on. It is called once per node run, and all Deltas from that run share it.

The revision is read after the node runs. If the node changes the workspace (e.g. commits a fix), you get the new revision. To report the revision the node started from, store it in the state before the node runs and return it from there.

Returns

An async node that you add to the graph in place of fn.

Behavior

Runs fn, then calls extract(state, result).
Sends one Delta per (event, payload) pair.
Adds wall_ms (the node’s run time in milliseconds) to every payload.
Converts dataclasses in the payload to plain values with to_jsonable.
Delta failures never break the node: emit_delta fails open.
Errors raised by fn or extract still propagate.

Example

import os

from aether_sdk import wrap_node
from langgraph.graph import StateGraph


def summarize(state: dict) -> dict:
    return {"summary": f"{len(state['docs'])} documents"}


def extract(state: dict, result: dict) -> list[tuple[str, dict]]:
    return [("summary.created", {"summary": result["summary"]})]


graph = StateGraph(dict)
graph.add_node(
    "summarize",
    wrap_node(
        summarize,
        extract=extract,
        workspace_revision=lambda state: state["revision"],
        api_key=os.environ["AETHER_API_KEY"],
    ),
)

Running the node reports a summary.created Delta with the payload {"summary": "...", "wall_ms": 12}.

Getting started

Important concepts

Aether

When to use it

Signature

What you implement

`extract`

`workspace_revision`

Returns

Behavior

Example

​When to use it

​Signature

​What you implement

​extract

​workspace_revision

​Returns

​Behavior

​Example

When to use it

Signature

What you implement

`extract`

`workspace_revision`

Returns

Behavior

Example