Interrupts

from langgraph.types import interrupt
 
def approval_node(state: State):
    # Pause and ask for approval
    approved = interrupt("Do you approve this action?")
 
    # When you resume, Command(resume=...) returns that value here
    return {"approved": approved}

from typing import Any
 
from pydantic import BaseModel
from typing_extensions import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
from langgraph.types import Command, interrupt
 
 
class Decision(BaseModel):
    approved: bool
    note: str | None = None
 
 
class State(TypedDict):
    answer: Any
 
 
def node(state: State) -> State:
    answer = interrupt({"question": "approve?"}, response_schema=Decision)
    return {"answer": answer}
 
 
graph = (
    StateGraph(State)
    .add_node("node", node)
    .add_edge(START, "node")
    .compile(checkpointer=InMemorySaver())
)
config = {"configurable": {"thread_id": "1"}}
 
result = graph.invoke({"answer": None}, config)
pending = result["__interrupt__"][0]
assert pending.response_schema == Decision.model_json_schema()
 
result = graph.invoke(Command(resume={"approved": True}), config)
assert result["answer"] == Decision(approved=True)

from typing import Literal
from langgraph.types import interrupt, Command
 
def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # Pause execution; payload shows up on stream.interrupts (with stream_events) or result["__interrupt__"] (with invoke)
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })
 
    # Route based on the response
    if is_approved:
        return Command(goto="proceed")  # Runs after the resume payload is provided
    else:
        return Command(goto="cancel")

# To approve
graph.stream_events(Command(resume=True), config=config, version="v3").output
 
# To reject
graph.stream_events(Command(resume=False), config=config, version="v3").output

:::python



:::

:::js

```typescript
import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  actionDetails: z.string(),
  status: z.enum(["pending", "approved", "rejected"]).nullable(),
});

const graphBuilder = new StateGraph(State)
  .addNode("approval", async (state) => {
    // Expose details so the caller can render them in a UI
    const decision = interrupt({
      question: "Approve this action?",
      details: state.actionDetails,
    });
    return new Command({ goto: decision ? "proceed" : "cancel" });
  }, { ends: ['proceed', 'cancel'] })
  .addNode("proceed", () => ({ status: "approved" }))
  .addNode("cancel", () => ({ status: "rejected" }))
  .addEdge(START, "approval")
  .addEdge("proceed", END)
  .addEdge("cancel", END);

// Use a more durable checkpointer in production
const checkpointer = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer });

const config = { configurable: { thread_id: "approval-123" } };
const initial = await graph.invoke(
  { actionDetails: "Transfer $500", status: "pending" },
  config,
);
console.log(initial.__interrupt__);
// [{ value: { question: ..., details: ... } }]

// Resume with the decision; true routes to proceed, false to cancel
const resumed = await graph.invoke(new Command({ resume: true }), config);
console.log(resumed.status); // -> "approved"
```
:::

from langgraph.types import interrupt
 
def review_node(state: State):
    # Pause and show the current content for review (payload surfaces on stream.interrupts)
    edited_content = interrupt({
        "instruction": "Review and edit this content",
        "content": state["generated_text"]
    })
 
    # Update the state with the edited version
    return {"generated_text": edited_content}

graph.stream_events(
    Command(resume="The edited and improved text"),  # Value becomes the return from interrupt()
    config=config,
    version="v3",
).output

:::python



:::

:::js

```typescript
import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  generatedText: z.string(),
});

const builder = new StateGraph(State)
  .addNode("review", async (state) => {
    // Ask a reviewer to edit the generated content
    const updated = interrupt({
      instruction: "Review and edit this content",
      content: state.generatedText,
    });
    return { generatedText: updated };
  })
  .addEdge(START, "review")
  .addEdge("review", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const config = { configurable: { thread_id: "review-42" } };
const initial = await graph.invoke({ generatedText: "Initial draft" }, config);
console.log(initial.__interrupt__);
// [{ value: { instruction: ..., content: ... } }]

// Resume with the edited text from the reviewer
const finalState = await graph.invoke(
  new Command({ resume: "Improved draft after review" }),
  config,
);
console.log(finalState.generatedText); // -> "Improved draft after review"
```
:::

from langchain.tools import tool
from langgraph.types import interrupt
 
@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""
 
    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?"
    })
 
    if response.get("action") == "approve":
        # Resume value can override inputs before executing
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)
        return f"Email sent to {final_to} with subject '{final_subject}'"
    return "Email cancelled by user"

:::python

```python
import sqlite3
import operator
from typing import TypedDict, Annotated, Literal
from langchain.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import AnyMessage, SystemMessage, ToolMessage


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]


@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?",
    })

    if response.get("action") == "approve":
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)

        # Actually send the email (your implementation here)
        print(f"[send_email] to={final_to} subject={final_subject} body={final_body}")
        return f"Email sent to {final_to}"

    return "Email cancelled by user"


model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email])
tools_by_name = {"send_email": send_email}


def agent_node(state: AgentState):
    # LLM may decide to call the tool; interrupt pauses before sending
    result = model.invoke(state["messages"])
    return {"messages": [result]}

def tool_node(state: AgentState):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

def should_continue(state: AgentState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    if last_message.tool_calls:
        return "tool_node"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tool_node", tool_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tool_node", END])  # Routes to "tools" or END
builder.add_edge("tool_node", "agent")  # Loop back after tools

checkpointer = SqliteSaver(
    sqlite3.connect("tool-approval.db", check_same_thread=False)
)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "email-workflow"}}
initial = graph.stream_events(
    {
        "messages": [
            {"role": "user", "content": "Send an email to [email protected] about the meeting"}
        ]
    },
    config=config,
    version="v3",
)
initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'action': 'send_email', ...}),)

# Resume with approval and optionally edited arguments
resumed = graph.stream_events(
    Command(resume={"action": "approve", "subject": "Updated subject"}),
    config=config,
    version="v3",
)
print(resumed.output["messages"][-1])  # -> Tool result returned by send_email
```
:::

:::js

```typescript
import { tool } from "@langchain/core/tools";
import { ChatAnthropic } from "@langchain/anthropic";
import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  MessagesValue,
  GraphNode,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const sendEmailTool = tool(
  async ({ to, subject, body }) => {
    // Pause before sending; payload surfaces in result.__interrupt__
    const response = interrupt({
      action: "send_email",
      to,
      subject,
      body,
      message: "Approve sending this email?",
    });

    if (response?.action === "approve") {
      const finalTo = response.to ?? to;
      const finalSubject = response.subject ?? subject;
      const finalBody = response.body ?? body;
      console.log("[sendEmailTool]", finalTo, finalSubject, finalBody);
      return `Email sent to ${finalTo}`;
    }
    return "Email cancelled by user";
  },
  {
    name: "send_email",
    description: "Send an email to a recipient",
    schema: z.object({
      to: z.string(),
      subject: z.string(),
      body: z.string(),
    }),
  },
);

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" }).bindTools([sendEmailTool]);

const State = new StateSchema({
  messages: MessagesValue,
});

const agent: typeof State.Node = async (state) => {
  // LLM may decide to call the tool; interrupt pauses before sending
  const response = await model.invoke(state.messages);
  return { messages: [response] };
};

const graphBuilder = new StateGraph(State)
  .addNode("agent", agent)
  .addEdge(START, "agent")
  .addEdge("agent", END);

const checkpointer = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer });

const config = { configurable: { thread_id: "email-workflow" } };
const initial = await graph.invoke(
  {
    messages: [
      { role: "user", content: "Send an email to [email protected] about the meeting" },
    ],
  },
  config,
);
console.log(initial.__interrupt__); // -> [{ value: { action: 'send_email', ... } }]

// Resume with approval and optionally edited arguments
const resumed = await graph.invoke(
  new Command({
    resume: { action: "approve", subject: "Updated subject" },
  }),
  config,
);
console.log(resumed.messages.at(-1)); // -> Tool result returned by send_email
```
:::

:::python



:::

:::js

```typescript
import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  age: z.number().nullable(),
  pendingQuestion: z.string().nullable(),
});

const builder = new StateGraph(State)
  .addNode("collectAge", (state) => {
    const question = state.pendingQuestion ?? "What is your age?";
    const answer = interrupt(question); // called exactly once per invocation

    if (typeof answer === "number" && answer > 0) {
      return { age: answer, pendingQuestion: null };
    }
    return { pendingQuestion: `'${answer}' is not a valid age. Please enter a positive number.` };
  })
  .addEdge(START, "collectAge")
  .addConditionalEdges("collectAge", (state) =>
    state.age !== null ? END : "collectAge"
  );

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const config = { configurable: { thread_id: "form-1" } };
const first = await graph.invoke({ age: null, pendingQuestion: null }, config);
console.log(first.__interrupt__); // -> [{ value: "What is your age?", ... }]

// Provide invalid data; the node re-prompts via the conditional edge
const retry = await graph.invoke(new Command({ resume: "thirty" }), config);
console.log(retry.__interrupt__); // -> [{ value: "'thirty' is not a valid age...", ... }]

// Provide valid data; route returns END and the graph finishes
const final = await graph.invoke(new Command({ resume: 30 }), config);
console.log(final.age); // -> 30
```
:::

def node_a(state: State):
# ✅ Good: interrupting first, then handling
# error conditions separately
interrupt("What's your name?")
try:
    fetch_data()  # This can fail
except Exception as e:
    print(e)
return state

def node_a(state: State):
# ✅ Good: catching specific exception types
# will not catch the interrupt exception
try:
    name = interrupt("What's your name?")
    fetch_data()  # This can fail
except NetworkException as e:
    print(e)
return state

def node_a(state: State):
    # ❌ Bad: wrapping interrupt in bare try/except
    # will catch the interrupt exception
    try:
        interrupt("What's your name?")
    except Exception as e:
        print(e)
    return state

def node_a(state: State):
    # ✅ Good: interrupt calls happen in the same order every time
    name = interrupt("What's your name?")
    age = interrupt("What's your age?")
    city = interrupt("What's your city?")
 
    return {
        "name": name,
        "age": age,
        "city": city
    }

def node_a(state: State):
# ❌ Bad: conditionally skipping interrupts changes the order
name = interrupt("What's your name?")
 
# On first run, this might skip the interrupt
# On resume, it might not skip it - causing index mismatch
if state.get("needs_age"):
    age = interrupt("What's your age?")
 
city = interrupt("What's your city?")
 
return {"name": name, "city": city}

def node_a(state: State):
# ❌ Bad: looping based on non-deterministic data
# The number of interrupts changes between executions
results = []
for item in state.get("dynamic_list", []):  # List might change between runs
    result = interrupt(f"Approve {item}?")
    results.append(result)
 
return {"results": results}

def node_a(state: State):
# ✅ Good: passing simple types that are serializable
name = interrupt("What's your name?")
count = interrupt(42)
approved = interrupt(True)
 
return {"name": name, "count": count, "approved": approved}

def node_a(state: State):
# ✅ Good: passing dictionaries with simple values
response = interrupt({
    "question": "Enter user details",
    "fields": ["name", "email", "age"],
    "current_values": state.get("user", {})
})
 
return {"user": response}

def validate_input(value):
return len(value) > 0
 
def node_a(state: State):
# ❌ Bad: passing a function to interrupt
# The function cannot be serialized
response = interrupt({
    "question": "What's your name?",
    "validator": validate_input  # This will fail
})
return {"name": response}

class DataProcessor:
def __init__(self, config):
    self.config = config
 
def node_a(state: State):
processor = DataProcessor({"mode": "strict"})
 
# ❌ Bad: passing a class instance to interrupt
# The instance cannot be serialized
response = interrupt({
    "question": "Enter data to process",
    "processor": processor  # This will fail
})
return {"result": response}

def node_a(state: State):
# ✅ Good: using upsert operation which is idempotent
# Running this multiple times will have the same result
db.upsert_user(
    user_id=state["user_id"],
    status="pending_approval"
)
 
approved = interrupt("Approve this change?")
 
return {"approved": approved}

def node_a(state: State):
# ✅ Good: placing side effect after the interrupt
# This ensures it only runs once after approval is received
approved = interrupt("Approve this change?")
 
if approved:
    db.create_audit_log(
        user_id=state["user_id"],
        action="approved"
    )
 
return {"approved": approved}

def approval_node(state: State):
# ✅ Good: only handling the interrupt in this node
approved = interrupt("Approve this change?")
 
return {"approved": approved}
 
def notification_node(state: State):
# ✅ Good: side effect happens in a separate node
# This runs after approval, so it only executes once
if (state.approved):
    send_notification(
        user_id=state["user_id"],
        status="approved"
    )
 
return state

def node_a(state: State):
# ❌ Bad: creating a new record before interrupt
# This will create duplicate records on each resume
audit_id = db.create_audit_log({
    "user_id": state["user_id"],
    "action": "pending_approval",
    "timestamp": datetime.now()
})
 
approved = interrupt("Approve this change?")
 
return {"approved": approved, "audit_id": audit_id}

def node_a(state: State):
# ❌ Bad: appending to a list before interrupt
# This will add duplicate entries on each resume
db.append_to_history(state["user_id"], "approval_requested")
 
approved = interrupt("Approve this change?")
 
return {"approved": approved}

def node_in_parent_graph(state: State):
    some_code()  # <-- This will re-execute when resumed
    # Invoke a subgraph as a function.
    # The subgraph contains an `interrupt` call.
    subgraph_result = subgraph.invoke(some_input)
    # ...
 
def node_in_subgraph(state: State):
    some_other_code()  # <-- This will also re-execute when resumed
    result = interrupt("What's your name?")
    # ...

    :::python
    ```python
    graph = builder.compile(
        interrupt_before=["node_a"],  # [!code highlight]
        interrupt_after=["node_b", "node_c"],  # [!code highlight]
        checkpointer=checkpointer,
    )

    # Pass a thread ID to the graph
    config = {
        "configurable": {
            "thread_id": "some_thread"
        }
    }

    # Run the graph until the breakpoint
    graph.invoke(inputs, config=config)  # [!code highlight]

    # Resume the graph
    graph.invoke(None, config=config)  # [!code highlight]
    ```

    1. The breakpoints are set during `compile` time.
    2. `interrupt_before` specifies the nodes where execution should pause before the node is executed.
    3. `interrupt_after` specifies the nodes where execution should pause after the node is executed.
    4. A checkpointer is required to enable breakpoints.
    5. The graph is run until the first breakpoint is hit.
    6. The graph is resumed by passing in `None` for the input. This will run the graph until the next breakpoint is hit.
    :::
    :::js
    ```typescript
    const graph = builder.compile({
        interruptBefore: ["node_a"],  // [!code highlight]
        interruptAfter: ["node_b", "node_c"],  // [!code highlight]
        checkpointer,
    });

    // Pass a thread ID to the graph
    const config = {
        configurable: {
            thread_id: "some_thread"
        }
    };

    // Run the graph until the breakpoint
    await graph.invoke(inputs, config);# [!code highlight]

    await graph.invoke(null, config);  # [!code highlight]
    ```

    1. The breakpoints are set during `compile` time.
    2. `interruptBefore` specifies the nodes where execution should pause before the node is executed.
    3. `interruptAfter` specifies the nodes where execution should pause after the node is executed.
    4. A checkpointer is required to enable breakpoints.
    5. The graph is run until the first breakpoint is hit.
    6. The graph is resumed by passing in `null` for the input. This will run the graph until the next breakpoint is hit.
    :::

    :::python
    ```python
    config = {
        "configurable": {
            "thread_id": "some_thread"
        }
    }

    # Run the graph until the breakpoint
    graph.invoke(
        inputs,
        interrupt_before=["node_a"],  # [!code highlight]
        interrupt_after=["node_b", "node_c"],  # [!code highlight]
        config=config,
    )

    # Resume the graph
    graph.invoke(None, config=config)  # [!code highlight]
    ```

    1. `graph.invoke` is called with the `interrupt_before` and `interrupt_after` parameters. This is a run-time configuration and can be changed for every invocation.
    2. `interrupt_before` specifies the nodes where execution should pause before the node is executed.
    3. `interrupt_after` specifies the nodes where execution should pause after the node is executed.
    4. The graph is run until the first breakpoint is hit.
    5. The graph is resumed by passing in `None` for the input. This will run the graph until the next breakpoint is hit.
    :::
    :::js
    ```typescript
    // Run the graph until the breakpoint
    graph.invoke(inputs, {
        interruptBefore: ["node_a"],  // [!code highlight]
        interruptAfter: ["node_b", "node_c"],  // [!code highlight]
        configurable: {
            thread_id: "some_thread"
        }
    });

    // Resume the graph
    await graph.invoke(null, config);  // [!code highlight]
    ```

    1. `graph.invoke` is called with the `interruptBefore` and `interruptAfter` parameters. This is a run-time configuration and can be changed for every invocation.
    2. `interruptBefore` specifies the nodes where execution should pause before the node is executed.
    3. `interruptAfter` specifies the nodes where execution should pause after the node is executed.
    4. The graph is run until the first breakpoint is hit.
    5. The graph is resumed by passing in `null` for the input. This will run the graph until the next breakpoint is hit.
    :::