LangGraph: prebuilt nodes that are Restate steps, and langgraph.agent(graph=...) by igalshilman · Pull Request #288 · restatedev/sdk-python

igalshilman · 2026-11-06T18:22:12Z

This PR adds two things for LangGraph graphs of your own.

Prebuilt nodes that are Restate steps

from restate.agents.langgraph import nodes

graph.add_node("read", nodes.subagent(clerk, message=asked, output="refund"))
graph.add_node("approve", nodes.wait_for_signal("approval", input=lambda s: s["refund"], output="approval"))
graph.add_node("pay", nodes.workflow(pay, input=lambda s: s["refund"], output="receipt"))  # a dict, validated into Refund

Node	What it does
`nodes.wait_for_signal(signal, output=, input=)`	Waits durably for a signal. The agent's `pending` view lists the wait, with `input` converted to JSON, and its `signal` view resolves it.
`nodes.workflow(handler, output=, input=, key=)`	Runs a workflow to completion. By default the workflow's id is the node's own.
`nodes.call(handler, output=, input=, key=)`	Calls a handler of a service, a virtual object or a workflow.
`nodes.send(handler, input=, key=, delay=)`	Sends to a handler, as a one-way call.
`nodes.subagent(agent, message=, output=, key=, handler="chat")`	Asks another agent, by default at the key `<this key>.<node name>`.

How the nodes behave:

Each step happens once. A node's idempotency key is LangGraph's task namespace. LangGraph derives it from the checkpoint the step started from, so a node re-run after a crash joins the step it started before. A node that runs again on a later pass through the graph, such as in a loop, is a new task and makes a new step.
Plain data is validated into the handler's input. A dict from the state is validated into the handler's input type, such as a Pydantic model, a dataclass or a TypedDict, with Pydantic's TypeAdapter. A value that already has that type passes through. A type Pydantic doesn't know, such as a msgspec Struct, goes to the handler's serde unchanged. Input that doesn't validate fails the node with a terminal error, instead of an attempt that Restate retries forever.
Mistakes are caught early. key must be given exactly when the handler is keyed, and input exactly when it takes one; otherwise building the node raises a TypeError.
The nodes refuse an unsafe setup. They won't run without a checkpointer, without the handler's context (ainvoke(context=ctx)), or with any durability other than "sync". With other durabilities, a crash could recreate a step's checkpoint under a new id, and so a new key.

`langgraph.agent(name, graph=...)`: `chat` runs your graph

refunds = langgraph.agent("refunds", graph=graph)  # a StateGraph on MessagesState, or a compiled graph
answer = await langgraph.prompt(ctx, text, graph=graph)  # the same, from a handler of your own

Each message is a run of the graph on the key's thread, with the message as a new HumanMessage. The graph has its own model and tools, so model, system and tools are refused.
Durability is built in. The checkpoints go in the key's database, each step is stored before it runs, and the run's context is the handler's, so the prebuilt nodes work without any setup. A retried attempt resumes the run it cut off, or takes the answer of a run that had already finished.
The conversation views follow the run. After each step, the AI and tool messages it added are copied into the agent's conversation by id, so messages, sync, watch and the chat CLI show the run as it goes.
The checkpoints and the conversation stay consistent.
- The conversation never gets ahead of the checkpoints: with durability="sync", LangGraph streams a step's state only once its checkpoint is stored. A test checks this ordering against a deliberately slow checkpointer, and the test fails under "async".
- If a crash cuts off a copy, the next step, or the end of the run, makes it.
- The conversation is a log: a message the graph later removes or edits stays as it was copied.
What ended the last turn carries over. After a stopped, refused or failed turn, the next run's input starts with what closed that turn: the results of its cut-off tool calls, and its closing note, such as "Cancelled…". So the graph's history never has a tool call without a result, which model providers refuse, and the graph's model knows how the turn ended, as the built-in agent's does. That turn's message is carried too, if the graph never stored it. Each item is carried once.
The answer is the run's last AI message. output= validates it; with your own graph, it's up to the graph's own prompts to ask for the JSON.

Example

examples-agents/intermediate/refund_graph.py is a refund desk you chat with:

A clerk agent, a langgraph.agent with output=Refund, reads the request.
A policy agent checks it.
A manager approves through a signal; an ordinary conditional edge routes on the answer.
The payout workflow runs.

tests/examples_agents.py serves it with scripted agents, and checks both the approved and the rejected path.

Tests

tests/agents_langgraph_nodes.py and tests/agents_langgraph_graph.py run against a Restate server:

Nodes:
- a graph that waits for a signal, runs a workflow, and asks another agent;
- call and send;
- dict-to-model validation, and invalid input;
- argument checks, msgspec types, and the durability guard.
Crashes: a parallel branch crashes while a node's step is under way, both during a subagent call and during a signal wait. Both tests fail if the key is made random.
The graph agent:
- the conversation going on, a compiled graph, prompt(graph=), and output=;
- a signal wait inside chat;
- crash and resume, with each message copied once;
- stop, and what carries over after it, both its note and a cut-off tool call. These tests fail if the carry-over is removed;
- the ordering test.

The helpers that the tools and the nodes share in _tools.py drop their leading underscore.

nodes.wait_for_signal, nodes.workflow and nodes.subagent make nodes for a graph of your own, as the tools do for an agent: each reads its input from the state and writes what it gets back to the state field `output`. The graph runs in a handler of an AgentObject, with ainvoke(context=ctx) and SqlCheckpointSaver(ctx.db). Each node's idempotency key is LangGraph's task namespace, which LangGraph derives from the checkpoint the step started from: a node re-run after a crash gets the same one, so it joins the wait, the workflow or the other agent's answer it started before. A wait is listed by the agent's pending view, and its signal view resolves it. The helpers the tools and the nodes share in _tools.py lose their leading underscore (handler_of, input_type_of, inputs_of, untyped).

pullfrog

ℹ️ No critical issues — one suggestion inline.

Reviewed changes

Reviewed the whole PR: the new restate.agents.langgraph.nodes module, the helper renames in _tools.py, and the Restate-server tests.

Prebuilt nodes: wait_for_signal, workflow and subagent each make one Restate call. The idempotency key comes from LangGraph's task namespace, <node>:<task_id>, and the task id is derived from the checkpoint id.
Guards: _step refuses to run without the handler's context (ainvoke(context=ctx)) or without a checkpointer.
Shared helpers: _untyped, _inputs, _handler and _input_type in _tools.py drop their leading underscore. No old names are left anywhere.
Tests: the end-to-end graph, plus two crash-in-a-parallel-branch tests that check the step is joined rather than repeated. ruff and pyright pass locally on the touched files.

^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

github-actions · 2026-11-06T18:30:18Z

Test Results

8 files ±0 8 suites ±0 3m 50s ⏱️ +15s
71 tests ±0 71 ✅ ±0 0 💤 ±0 0 ❌ ±0
322 runs ±0 322 ✅ ±0 0 💤 ±0 0 ❌ ±0

Results for commit 669b763. ± Comparison against base commit 0d06f2c.

♻️ This comment has been updated with latest results.

LangGraph's default durability, "async", only submits a step's checkpoint save, and the next step's tasks start right away ("exit" saves nothing between steps). A crash before the save makes the resumed run create the checkpoint again with a new id, so the node gets a new task id and a new idempotency key, and makes its step twice. Only "sync" stores the checkpoint before the step runs.

pullfrog

✅ No new issues found. The durability concern from the last review is fixed.

Reviewed changes

Reviewed the one commit since the last Pullfrog review, which makes the nodes require durability="sync".

Durability guard: _step now raises TypeError unless __pregel_durability is "sync". It rejects a missing value (LangGraph defaults to "async"), "async" and "exit". Subgraphs get the parent's value through configurable, so the guard also applies inside them.
Docs: the module docstring explains that durability="sync" stores each step's checkpoint before the step runs.
Test: the new test_a_node_refuses_to_run_unless_each_step_is_stored_before_it_runs runs the graph with no durability, "async" and "exit". It passes locally, as does the workflow-input test.

^{｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

examples-agents/intermediate/refund_graph.py runs a graph of your own in a handler of an AgentObject: it asks a policy agent about the refund (nodes.subagent), waits for a manager's approval (nodes.wait_for_signal), routes on the answer with an ordinary conditional edge, and runs the payout workflow (nodes.workflow). tests/examples_agents.py serves it with a scripted policy agent, and checks both an approved refund, which is paid, and a rejected one, which ends unpaid.

pullfrog

✅ No new issues found.

Reviewed changes

Reviewed the one commit since the last Pullfrog review. It adds an example that uses the nodes in a graph, plus a test for it.

refund_graph.py example: a graph with three steps. check asks a policy subagent, approve waits for an approval signal, and pay runs the payouts workflow only if the refund is approved. The handler resumes a run that a crash cut off, using aget_state(config).next, and runs the graph with durability="sync", as the durability guard requires.
Example test: test_intermediate_refund_graph puts a scripted VirtualObject("policy") in place of the model-backed agent before serving. This works because nodes.subagent finds its target by agent.name when it runs. The test checks what pending shows, then sends the signal, for one approved refund and one rejected refund.
README: added a row for the new example.

ruff check and ruff format --check pass on both files. Pyright could not check them here because this runner doesn't have langgraph or pytest installed.

^{｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

The graph's state has LangGraph's messages (MessagesState); a StateGraph or a compiled graph (such as LangChain's create_agent makes) is accepted, with no model, system or tools of ours. Each message is a run of the graph on the key's thread, with the message as a new HumanMessage whose id is the prompt's: - The checkpoints are in the key's database (SqlCheckpointSaver, history "latest"), stored before each step (durability "sync"), and the run's context is the handler's, so the prebuilt nodes work in it. A retried attempt resumes the run it cut off, or takes the finished run's answer. - After each step, the AI and tool messages it added are copied into the agent's conversation by id, so the messages, sync and watch views show the run as it goes. LangGraph streams a step's state only once its checkpoint is stored, so the conversation never gets ahead of the checkpoints; a copy a crash cut off is made at the next step or at the end. A test checks that ordering. - The answer is the run's last AI message; output= validates it. - After a stopped run, the next message starts a run of its own. The refund example becomes a desk you chat with: a clerk agent (output=Refund) reads the request, a policy agent checks it, a manager approves, and the payout workflow runs.

… input Graph state is plain data, while a workflow's handler often takes a Pydantic model (or a dataclass, or a TypedDict). A dict from the state was passed through as it was, so the handler's serde failed on every attempt and Restate retried forever. The node now validates the value with Pydantic's TypeAdapter into the handler's input type; a value of that type already is passed through. A value that does not validate fails the node with a terminal error naming the handler and the field, since a retry would read the same state. The refund example passes the state's dict again.

pullfrog

Important

One problem to fix before merging: after a stop, a create_agent-style graph can be left unable to answer any later message on that key. Details inline.

Reviewed changes

Reviewed the one commit since the last Pullfrog review. It adds langgraph.agent(name, graph=...) and changes the refund example into a chat agent.

Graph agent: _graph.py's run_of runs the graph on the key's "chat" thread. Checkpoints go in SqlCheckpointSaver(ctx.db, history="latest") and the run uses durability="sync". The prompt goes in as a HumanMessage whose id is prompt:<prompt id>, so a retried attempt can tell whether to resume the run, reuse its finished answer, or start a new one.
Conversation copy: after each streamed state, _copied adds to the conversation, by id, the AI and tool messages after the prompt that it doesn't have yet, then publishes them. The answer is the last AI message, and output= validates it through answer_with.
Argument checks: agent() raises TypeError if graph is passed with model, system or tools, or if the graph's state has no messages channel.
Example and tests: refund_graph.py is now a chat desk with a clerk subagent (output=Refund) and an answer node. tests/agents_langgraph_graph.py covers the graph agent. The two in-process tests pass locally (-k "streams or own_model"). ruff and pyright are clean on the touched files.

^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

pullfrog

✅ No new issues in this commit. The finding from the previous review, about unanswered tool calls after a stop (_graph.py:121), still applies.

Reviewed changes

Reviewed the one commit since the last Pullfrog review. It makes nodes.workflow validate the input it takes from the state into the handler's input type.

Input validation: _validator builds a TypeAdapter for the handler's input type when the node is built, and skips it for None and Any. When the node runs, it validates input(state). An instance of the target type passes through unchanged. A ValidationError becomes a TerminalError that names the handler, so the node fails instead of retrying on the same state. Building a TypeAdapter up front is what Restate tools already do (_tools.py:345).
Example: refund_graph.py passes the refund dict from the state directly into pay.
Tests: the new invoice and bad_invoice graphs check that a dict is validated into Invoice, and that an invalid amount fails with the handler's name without running the workflow.

^{｜ Fix it ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

…nodes The graph agent: when a message starts a run after one that was stopped, refused or failed, the run's input starts with what closed that run in the conversation: the results of its cut-off tool calls, so that the graph's history has no tool call without a result (which a model provider refuses), and the note that ends it, so that the graph's model knows how the turn ended, as the built-in agent's does. The previous message comes too, if the graph never stored it. Each carried item has an id of its own, so it is carried once. nodes.call and nodes.send call or send to a handler, as the tools do. A key is required exactly for a keyed handler (a virtual object's or a workflow's), and plain data from the state is validated into the handler's input, as for nodes.workflow. _tools.keyed is shared with them.

- nodes: a handler input type Pydantic does not know (a msgspec Struct) no longer makes building the node fail: the value goes to the handler's serde as it is. - nodes.wait_for_signal: what the graph waits for is turned into JSON, so a model in the state (a workflow's result, say) no longer fails each attempt. - prompt(ctx, text, graph=...): a handler of your own prompts the graph as chat does; agent(graph=...) answers through it. - The graph agent copies a run's messages looking at the run's own entries only, not the whole conversation at each step. - Docs: nodes lead with langgraph.agent(graph=...); output= with a graph is the graph's own prompts' to ask for; the example does not claim the chat shows steps that add no messages. - Tests wait at most 20 seconds for their events, so a broken one fails rather than hangs; new tests for the above.

pullfrog

ℹ️ No critical issues. The earlier finding about unanswered tool calls after a stop is fixed. One minor suggestion inline.

Reviewed changes

Reviewed the one commit since the last Pullfrog review, 203c090.

Closing the previous run: _closing puts what close() added to the conversation at the start of the next run's input. That is the error results of cut-off tool calls and the closing note, each with its own closed:<n> id so it is only carried once. When the graph never stored the previous run's prompt, that prompt is carried as well. With these results in the graph's state, every tool call has a result, so the model provider no longer refuses the history after a stop. LangGraph's _first drops the pending tasks when it gets this input.
nodes.call and nodes.send: these call or send to a handler through the shared _target check. key= is required exactly when the handler is keyed. The input is validated the same way as in nodes.workflow. send also takes a delay. _tools._keyed is renamed to keyed.
Tests: new tests cover the graph seeing CANCELLED after a stop, a cut-off tool call getting a result exactly once, a call to an object followed by a send to a service, and the key argument checks. ruff passes, and the in-process tests pass locally (-k "takes_a_key or takes_input or refuses or streams or own_model").

^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

A run was found by its prompt's place in the graph's state, and its messages and answer were those after it. A graph that trims its history (a summarizing one) removes the prompt as well, and the run then copied nothing and answered with nothing; and the next message brought the trimmed prompt back. Each checkpoint of a run now names its prompt in its metadata, which is how a retried attempt finds the run it cut off. The messages copied are the graph's AI and tool messages that the conversation does not have (not those it gave the graph), and the answer is the last AI message the run copied, read from the conversation, which is a log. The previous prompt is carried into a new run only if its run copied nothing, so a prompt the graph removed is not brought back.

pullfrog

✅ No new issues in this commit. One earlier suggestion is still open. It is about _closing carrying back a previous prompt that the graph removed (_graph.py:136), and that code hasn't changed.

Reviewed changes

Reviewed the one commit since the last Pullfrog review, f7b518c, which applies fixes from a review of the whole PR.

prompt(graph=...): langgraph.prompt now takes a graph. It runs run_of through answer_once, and answer_with when output is given. The graph agent's chat now just calls it. _check_graph holds the model, system and tools check, so agent() and prompt() share it.
_copied scope: to find the messages already copied, _copied now looks only at the conversation entries after the running prompt. A run's copies are only ever added after its own prompt, and the closed:<n> items carried from the last turn come before the prompt, so _after never returns them.
Input types Pydantic doesn't know: _validator catches PydanticSchemaGenerationError (from a msgspec Struct, for example) and passes the value through unchanged, so the handler's serde handles it.
wait_for_signal input as JSON: the input value goes through to_jsonable_python before the signals companion stores it. Models in the state now show in pending as JSON.
Tests and docs: the new tests cover a custom handler that prompts the graph, a model as the signal's input, and a msgspec handler input. A new until helper puts a 20s limit on the crash tests' busy-waits. The nodes docstring now starts with langgraph.agent(graph=...). The new in-process msgspec test passes locally, along with the argument-check tests. ruff is clean, and pyright reports 0 errors on agents/langgraph/.

^{｜ Fix it ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

pullfrog

✅ No new issues found. The earlier suggestion about _closing bringing back a prompt that the graph removed is fixed. I resolved that thread.

Reviewed changes

Reviewed the one commit since the last Pullfrog review, 73c9121. With it, a graph that trims its history, including the run's input and answer, keeps working.

Finding a run by checkpoint metadata: run_of puts restate_prompt: prompt:<id> in the run's config["metadata"]. asked now reads that key from the latest checkpoint's metadata, instead of looking for the prompt's HumanMessage in the state. I checked locally that LangGraph's get_checkpoint_metadata adds the key to every checkpoint, the input checkpoint included, so a retried attempt still recognizes the run it cut off.
Copying by id across the whole state: _copied now looks at every message in the state, not only those after the prompt (_after is gone). It skips ids already in the conversation and ids with the prompt: or closed: prefix. The ids it copies go into known as it goes. clear empties the key's database, checkpoints included, so known and the graph's state stay in step.
Answer from the conversation: _answer returns the last AI message with an id that the run copied after its prompt. The closing notes from close() have no id, so they are never taken as an answer.
_closing: the previous prompt is carried back only when its run copied nothing. A run that copied something had stored its prompt.
Test: test_a_message_the_graph_trimmed_is_not_brought_back uses a trim node that removes every message, then a humans node that checks the old prompt didn't come back. The in-process tests pass locally (the others need Docker). ruff check, ruff format --check and pyright are clean on the touched files.

^{｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

The test assumed that b ran in the first attempt, beside c, which fails. When c fails before b starts, b runs in the retry instead (seen on CI, Python 3.13). Either way a and b run once, c twice, and d last, which is what the test now checks.