某些 tool operations 可能比较敏感，需要在 execution 前获得 human approval。Deep Agents 通过 LangGraph 的 interrupt capabilities 支持 human-in-the-loop workflows。你可以使用 interrupt_on parameter 配置哪些 tools 需要 approval。

​
Basic configuration

interrupt_on parameter 接受一个 dictionary，将 tool names 映射到 interrupt configurations。每个 tool 都可配置为：
  • True：使用默认 behavior 启用 interrupts（允许 approve、edit、reject、respond）
  • False：为该 tool 禁用 interrupts
  • {"allowed_decisions": [...]}：使用指定 allowed decisions 的 custom configuration
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver


@tool
def remove_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"


@tool
def fetch_file(path: str) -> str:
    """Read a file from the filesystem."""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    return f"Sent email to {to}"


# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

​
Decision types

allowed_decisions list 控制 human 在 review tool call 时可以采取哪些 actions：
  • "approve"：使用 agent 提议的 original arguments 执行 tool
  • "edit"：在 execution 前修改 tool arguments
  • "reject"：完全跳过该 tool call 的 execution，并向 agent 返回 rejection feedback
  • "respond"：将 human message 直接作为 tool result 返回，跳过 execution，适用于 “ask user” 风格的 tools
当 human 拒绝 proposed action 时，请使用 reject。只有当 human 扮演 tool 本身时才使用 respond，例如回答 ask_user prompt。不要用 respond 拒绝有 side effects 的 tools，因为它的 message 可能会被 model 当作 successful tool result。 你可以自定义每个 tool 可用的 decisions： 
interrupt_on = {
    # Sensitive operations: allow all options
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},

    # Moderate risk: approval or rejection only
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # Must approve (no rejection allowed)
    "critical_operation": {"allowed_decisions": ["approve"]},
}

​
Handle interrupts

触发 interrupt 时，agent 会暂停 execution 并返回 control。检查 result 中是否有 interrupts，并相应处理。如果 user rejects an action，请包含清晰的 message，告诉 agent tool 未执行，以及下一步应该怎么做。 
from langchain_core.utils.uuid import uuid7
from langgraph.types import Command

# Create config with thread_id for state persistence
config = {"configurable": {"thread_id": str(uuid7())}}

# Invoke the agent
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Delete the file temp.txt"}]},
    config=config,
    version="v2",
)

# Check if execution was interrupted
if result.interrupts:
    # Extract interrupt information
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["action_requests"]
    review_configs = interrupt_value["review_configs"]

    # Create a lookup map from tool name to review config
    config_map = {cfg["action_name"]: cfg for cfg in review_configs}

    # Display the pending actions to the user
    for action in action_requests:
        review_config = config_map[action["name"]]
        print(f"Tool: {action['name']}")
        print(f"Arguments: {action['args']}")
        print(f"Allowed decisions: {review_config['allowed_decisions']}")

    # Get user decisions (one per action_request, in order)
    decisions = [
        {
            "type": "reject",
            "message": "User rejected deleting temp.txt. Do not retry deletion.",
        }
    ]

    # Resume execution with decisions
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,  # Must use the same config!
        version="v2",
    )

# Process final result
print(result.value["messages"][-1].content)

​
Multiple tool calls

当 agent 调用多个需要 approval 的 tools 时，所有 interrupts 都会 batch 到一个 interrupt 中。你必须按顺序为每个 action 提供 decisions。 
config = {"configurable": {"thread_id": str(uuid7())}}

result = agent.invoke(
    {"messages": [{
        "role": "user",
        "content": "Delete temp.txt and send an email to admin@example.com"
    }]},
    config=config,
    version="v2",
)

if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["action_requests"]

    # Two tools need approval
    assert len(action_requests) == 2

    # Provide decisions in the same order as action_requests
    decisions = [
        {"type": "approve"},  # First tool: delete_file
        {
            "type": "reject",
            "message": "User rejected this action. Do not retry this tool call.",
        }  # Second tool: send_email
    ]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
        version="v2",
    )

​
Rejection messages

当 reviewer 返回 reject decision 时，Deep Agents 会跳过 tool call，并将 rejection feedback 发送回 agent。如果你省略 message，default feedback 会告诉 model tool 未执行，且除非 user 要求，否则不要 retry 相同 tool call。 对于 sensitive 或 side-effecting tools，请随 decision 传入 domain-specific message。明确说明 agent 应该 abandon action、ask a follow-up question，还是尝试 safer alternative。 
decisions = [
    {
        "type": "reject",
        "message": "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead.",
    }
]

​
Edit tool arguments

当 allowed decisions 中包含 "edit" 时，你可以在 execution 前修改 tool arguments： 
if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_request = interrupt_value["action_requests"][0]

    # Original args from the agent
    print(action_request["args"])  # {"to": "everyone@company.com", ...}

    # User decides to edit the recipient
    decisions = [{
        "type": "edit",
        "edited_action": {
            "name": action_request["name"],  # Must include the tool name
            "args": {"to": "team@company.com", "subject": "...", "body": "..."}
        }
    }]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
        version="v2",
    )

​
Subagent interrupts

使用 subagents 时，你可以在 tool calls 上 和 tool calls 内 使用 interrupts。

​
Interrupts on tool calls

每个 subagent 都可以拥有自己的 interrupt_on configuration，用来 override main agent 的 settings： 
agent = create_deep_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[delete_file, read_file],
    interrupt_on={
        "delete_file": True,
        "read_file": False,
    },
    subagents=[{
        "name": "file-manager",
        "description": "Manages file operations",
        "system_prompt": "You are a file management assistant.",
        "tools": [delete_file, read_file],
        "interrupt_on": {
            # Override: require approval for reads in this subagent
            "delete_file": True,
            "read_file": True,  # Different from main agent!
        }
    }],
    checkpointer=checkpointer
)
当 subagent 触发 interrupt 时，handling 相同：检查 result 上的 interrupts，并使用 Command resume。

​
Interrupts within tool calls

Subagent tools 可以直接调用 interrupt() 来 pause execution 并等待 approval： 
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

from deepagents.graph import create_deep_agent
from deepagents.middleware.subagents import CompiledSubAgent


@tool(description="Request human approval before proceeding with an action.")
def request_approval(action_description: str) -> str:
    """Request human approval using the interrupt() primitive."""
    # interrupt() pauses execution and returns the value passed to Command(resume=...)
    approval = interrupt({
        "type": "approval_request",
        "action": action_description,
        "message": f"Please approve or reject: {action_description}",
    })

    if approval.get("approved"):
        return f"Action '{action_description}' was APPROVED. Proceeding..."
    else:
        return f"Action '{action_description}' was REJECTED. Reason: {approval.get('reason', 'No reason provided')}"


def main():
    checkpointer = InMemorySaver()
    model = ChatAnthropic(
        model_name="claude-sonnet-4-6",
        max_tokens=4096,
    )

    compiled_subagent = create_agent(
        model=model,
        tools=[request_approval],
        name="approval-agent",
    )

    parent_agent = create_deep_agent(
        model="google_genai:gemini-3.5-flash",
        checkpointer=checkpointer,
        subagents=[
            CompiledSubAgent(
                name="approval-agent",
                description="An agent that can request approvals",
                runnable=compiled_subagent,
            )
        ],
    )

    thread_id = "test_interrupt_directly"
    config = {"configurable": {"thread_id": thread_id}}

    print("Invoking agent - sub-agent will use request_approval tool...")

    result = parent_agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content="Use the task tool to launch the approval-agent sub-agent. "
                    "Tell it to use the request_approval tool to request approval for 'deploying to production'."
                )
            ]
        },
        config=config,
        version="v2",
    )

    # Check for interrupt
    if result.interrupts:
        interrupt_value = result.interrupts[0].value  
        print(f"\nInterrupt received!")
        print(f"  Type: {interrupt_value.get('type')}")
        print(f"  Action: {interrupt_value.get('action')}")
        print(f"  Message: {interrupt_value.get('message')}")

        print("\nResuming with Command(resume={'approved': True})...")
        result2 = parent_agent.invoke(
            Command(resume={"approved": True}),
            config=config,
            version="v2",
        )

        if not result2.interrupts:
            print("\nExecution completed!")
            # Find the tool response
            tool_msgs = [m for m in result2.value.get("messages", []) if m.type == "tool"]
            if tool_msgs:
                print(f"  Tool result: {tool_msgs[-1].content}")
        else:
            print("\nAnother interrupt occurred")
    else:
        print("\n  No interrupt - the model may not have called request_approval")


if __name__ == "__main__":
    main()
运行后会产生以下 output： 
Invoking agent - sub-agent will use request_approval tool...

Interrupt received!
  Type: approval_request
  Action: deploying to production
  Message: Please approve or reject: deploying to production

Resuming with Command(resume={'approved': True})...

Execution completed!
  Tool result: Great! The approval request has been processed. The action **"deploying to production"** was **APPROVED**. You can now proceed with the production deployment.

​
Filesystem permission interrupts

Filesystem permission interrupts 需要 deepagents>=0.6.8。
除 interrupt_on 外，你还可以通过将 permission rule 标记为 mode="interrupt" 来暂停 built-in filesystem tools。当 agent 在匹配 interrupt-mode rule 的 path 上调用 write_file 或 edit_file 时，create_deep_agent 会发出与 configured tool 相同的 human-in-the-loop interrupt，并使用 filesystem tool name 作为 action name。 
from deepagents import FilesystemPermission, create_deep_agent
from langgraph.checkpoint.memory import MemorySaver


agent = create_deep_agent(
    model=model,
    permissions=[
        FilesystemPermission(
            operations=["write"],
            paths=["/secrets/**"],
            mode="interrupt",
        ),
    ],
    checkpointer=MemorySaver(),  # Required to pause and resume
)
以与 tool-call interrupt 相同的方式 handle 和 resume interrupt：运行到 pause、inspect request，然后用 decision resume。 
from langgraph.types import Command

config = {"configurable": {"thread_id": "fs-thread-1"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Save the API key to /secrets/key.txt"}]},
    config=config,
    version="v2",
)

if result.interrupts:
    action = result.interrupts[0].value["action_requests"][0]
    print(f"Approve {action['name']} on {action['args']}?")

    # Resume with the human decision (approve, edit, or reject).
    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,  # Same thread ID
        version="v2",
    )
Filesystem-permission interrupts 会与你传入的任何 interrupt_on 合并，因此单个 review step 可以同时覆盖 custom tools 和 protected filesystem paths。

​
Best practices

​
Always use a checkpointer

Human-in-the-loop 需要 checkpointer 在 interrupt 和 resume 之间持久化 agent state： 
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[...],
    interrupt_on={...},
    checkpointer=checkpointer  # Required for HITL
)

​
Use the same thread ID

Resuming 时，必须使用相同 config 和相同 thread_id： 
# First call
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config, version="v2")

# Resume (use same config)
result = agent.invoke(Command(resume={...}), config=config, version="v2")

​
Match decision order to actions

Decisions list 必须与 action_requests 的顺序匹配： 
if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["action_requests"]

    # Create one decision per action, in order
    decisions = []
    for action in action_requests:
        decision = get_user_decision(action)  # Your logic
        decisions.append(decision)

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
        version="v2",
    )

​
Tailor configurations by risk

根据 tools 的 risk level 配置不同策略： 
interrupt_on = {
    # High risk: full control (approve, edit, reject)
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
    "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},

    # Medium risk: no editing allowed
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # Low risk: no interrupts
    "read_file": False,
    "list_files": False,
}
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.
Edit this page on GitHub or file an issue.