> ## Documentation Index
> Fetch the complete documentation index at: https://documentation.aibydna.com/llms.txt
> Use this file to discover all available pages before exploring further.

# LangGraph: wrap_node

> Report Deltas automatically after a LangGraph node runs.

Wraps a LangGraph node so that, after it runs, each event you extract from its result is sent to Aether with [`emit_delta`](https://documentation.aibydna.com/desp/aether/python-sdk/reporting-deltas/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

```python theme={null}
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.

> [!NOTE]
>
> 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

```python theme={null}
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}`.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.