agent-wait

Get a LangGraph interrupt out of the process, and the answer back in.

When a node calls interrupt(), the graph pauses and the interrupt is handed to whatever called invoke() — and that is where LangGraph stops. There is no built-in way to tell anyone else that a question was asked, and no built-in way for anyone else to answer it. The moment the question has to reach a person on Slack, an approvals dashboard, a ticket queue or another service, you are writing that code yourself — whether your agent is a server that runs for a year or a Lambda that is gone in seconds.

agent-wait is that code. It takes the interrupt and puts it somewhere people can see it — a topic, a queue, a webhook, a database row — with everything needed to answer it in one envelope, and documents the shape of the answer so the return leg is one if.

pip install agent-wait langgraph-wait          # core + LangGraph
pip install agent-wait-aws                     # SNS / SQS / EventBridge / DynamoDB announcers

The whole thing

In the graph — one decorator, on the function where the decision belongs:

from agent_wait import WaitPolicy
from langgraph_wait import hitl

FINANCE = WaitPolicy(
    timeout="P3D",
    default={"action": "reject"},
    allowed_actions=("approve", "reject"),
    tags={"approver_group": "finance"},
)


@tool
@hitl(FINANCE)
def issue_refund(order_id: str, amount: int) -> str:
    payments.refund(order_id, amount)  # runs only if the answer is {"action": "approve"}
    return "refunded"

Calling issue_refund(...) inside a run parks the graph on {"function": "issue_refund", "args": {...}}. On resume, {"action": "approve"} runs the body — optionally with edited args — and anything else is returned in its place, so a model sees why it did not run.

A node that needs the answer itself declares a decision parameter and always runs:

@hitl(FINANCE)
def review(state, decision=None):
    return {"decision": decision}  # verbatim: {"action": "approve", "note": "ok"}

After the run — one call:

from agent_wait_aws import SnsAnnounce
from langgraph_wait import publish_interrupts

result = graph.invoke(value, {"configurable": {"thread_id": thread_id}})
publish_interrupts(result, thread_id, announce=[SnsAnnounce(topic_arn)])

It reads result["__interrupt__"] — Interrupt.id and Interrupt.value, nothing else — and hands one envelope per question to the announcers. No graph handle, no get_state(), no checkpointer. If nothing interrupted, it does nothing. A plain interrupt(value) from a node that never heard of this library is published too, with the default policy.

When the answer comes back — the one if you write:

from langgraph.types import Command

if "question_id" in message:  # an answer
    value = Command(resume={message["question_id"]: message["answer"]})
else:  # a start
    value = message["input"]

result = graph.invoke(value, {"configurable": {"thread_id": message["thread_id"]}})
publish_interrupts(result, message["thread_id"], announce)

That is the complete integration. A duplicate answer runs nothing — LangGraph ignores a resume for a question the thread has moved past — so there is nothing to check.

Async mode — the thread does not park

@tool
@hitl(FINANCE, mode="async", announce=[SnsAnnounce(topic_arn), DynamoDbAnnounce(table)])
def issue_refund(order_id: str, amount: int) -> str: ...

The decorator is the publisher: the call announces the question and returns {"status": "pending_approval", "question_id": ...} without running. The graph carries on; nothing is parked; nothing to call after the run. The decision arrives later as a new message and your graph acts on it. This is the mode for a single-thread channel like WhatsApp, where the approver is not the person on the thread.

What goes out

{
  "type": "wait.created",
  "thread_id": "order-4471",
  "question_id": "a1b2c3d4e5f60718",
  "question": { "function": "issue_refund", "args": { "order_id": "order-4471", "amount": 41000 } },
  "allowed_actions": ["approve", "reject"],
  "expires_at": "2026-09-14T09:00:00Z",
  "default": { "action": "reject" },
  "source": { "function": "issue_refund" },
  "reply_with": { "thread_id": "order-4471", "question_id": "a1b2c3d4e5f60718", "answer": null }
}

reply_with is a filled-in stub: copy it, set answer, send it back. Whatever goes in answer is what the @hitl function receives — verbatim.

Full schema, the recommended answer shape, and how to deduplicate: Message formats.

Announcers

Subclass BaseAnnounce, write one method:

from agent_wait import BaseAnnounce


class RedisAnnounce(BaseAnnounce):
    name = "redis"

    def __init__(self, client, **kw):
        super().__init__(**kw)
        self.client = client

    def deliver(self, envelope, transition):
        self.client.set(envelope.dedupe_key, envelope.to_json())

A raise inside deliver() becomes a log line; the run completes. Shipped:

Writing an announcer.

What the library does not do

• Receive answers. No endpoint, no validation, no ledger reads. The if above is yours.
• Enforce anything. expires_at, default, answer_ttl, allowed_actions are published so the consumer has the asker's intent. Acting on them is the consumer's.
• Store anything. LangGraph's checkpoint is the only state in interrupt mode; in async mode there is none unless you keep one.

Two things to know

Not compatible with LangChain's HumanInTheLoopMiddleware. It interrupts before a tool is called, in its own shape; @hitl interrupts inside the call. Both on one tool means two interrupts for one approval. Use one or the other.

One interrupt() per node on langgraph 1.2.x. Two interrupting tools dispatched by one ToolNode get the same interrupt id (#6626), and only one surfaces per run (#6624). Give each interrupting tool its own node. Pinned by a test that fails if LangGraph changes it.

Layout

packages/agent-wait        core: Question, WaitPolicy, the envelope, publish(), announcers. No deps.
packages/langgraph-wait    @hitl and publish_interrupts(). The only LangGraph import.
packages/agent-wait-aws    four announce adapters, and a CDK stack for the example.
examples/refund_agent      a graph and a host, deployed to Lambda behind SQS.
docs/                      message contract, architecture, announcer guide, consumer guide.

MIT. Issues and PRs at github.com/skamalj/agent-wait.