Human-in-the-Loop and Streaming

An agent that can call tools can call the wrong one. In this module you add an approval step to the tutorial agent so every tool call waits for a human, resume the paused run through the Runs API, and then watch a run unfold live over Server-Sent Events (SSE). AgentVisor™ makes the pause durable: the run waits in the thread's checkpoint, not in a process that has to stay alive.

What You'll Learn

• Pausing a run with LangGraph's interrupt() and reading what it is waiting for
• Resuming the run by posting a decision to the same thread, and rejecting a tool call
• Streaming a run with curl -N and reading its events
• How the pause shows up in the stream, and what each event means

Prerequisites

This module continues in the project you finished in the Basic tutorial. If you skipped it, scaffold the same agent:

agentvisor template create langgraph/tutorial-agent
cd tutorial-agent

You also need Docker running, jq, and the services from the Basic tutorial:

docker compose up -d

No provider key is needed. The agent uses the local Ollama model.

How It Works

POST /threads/{id}/runs          chat ──▶ approve ──▶ (pause)         run returns, thread waits
                                          ▲
POST /threads/{id}/runs (decision) ───────┘ resume ──▶ tools ──▶ chat ──▶ done

interrupt() stops the graph at that node and saves its state through the AgentVisor checkpointer. The run you started ends there. When you start another run on the same thread, AgentVisor sees the pending interrupt and hands your request's input to the waiting interrupt() call as its return value. The graph continues from where it stopped, even if the host restarted in between.

Step 1: Add an Approval Step

Today the agent's graph goes chat → tools → chat. Insert an approve node so a tool call must pass a human first: chat → approve → tools → chat.

In agent.py, import ToolMessage and interrupt:

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage, ToolMessage
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt

Change should_continue to route tool calls to the new node, and add the node and its router just below it:

def should_continue(state: ChatState) -> str:
    """Send tool calls to the approval step."""
    messages = state.get("messages", [])
    if messages and isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "approve"
    return "end"


def approve(state: ChatState) -> dict:
    """Pause until a human approves the tool calls the model asked for."""
    last = state["messages"][-1]
    decision = interrupt(
        {
            "message": "Approve these tool calls?",
            "tool_calls": [{"name": c["name"], "args": c["args"]} for c in last.tool_calls],
        }
    )
    if isinstance(decision, dict) and decision.get("approved") is True:
        return {}
    # Rejected: answer every pending tool call so the history stays valid.
    return {
        "messages": [
            ToolMessage(content="The user rejected this tool call.", tool_call_id=c["id"])
            for c in last.tool_calls
        ]
    }


def after_approval(state: ChatState) -> str:
    """Run the tools if approved; go back to the model if rejected."""
    return "chat" if isinstance(state["messages"][-1], ToolMessage) else "tools"

interrupt() takes any JSON-serializable payload, which is what the human sees. Put in everything needed to decide, here the tool names and arguments. Its return value is whatever the resuming request sends as input.

Rewire the graph at the bottom of the file. Add the approve node and replace the edges around tools:

builder.add_node("chat", chat)
builder.add_node("approve", approve)
builder.add_node("tools", ToolNode(tools))

builder.set_entry_point("chat")
builder.add_conditional_edges("chat", should_continue, {"approve": "approve", "end": END})
builder.add_conditional_edges("approve", after_approval, {"tools": "tools", "chat": "chat"})
builder.add_edge("tools", "chat")

Rejecting needs an answer for every tool call

A model message that contains tool calls must be followed by a result for each one. The rejection branch returns a ToolMessage per call, so the model can see what happened and reply, instead of the conversation being left in a state the model API refuses.

Step 2: Run It and See It Pause

Start the agent:

agentvisor serve .

In a second terminal, create a thread and ask for a fetch. The fetch <url> form calls the tool directly, so the result doesn't depend on the local model:

THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "fetch https://httpbin.org/get"}]}}' \
    | jq '.output.__interrupt__'

The run returns right away instead of fetching:

[
  "Interrupt(value={'message': 'Approve these tool calls?', 'tool_calls': [{'name': 'fetch_url', 'args': {'url': 'https://httpbin.org/get'}}]}, id='88caa949...', response_schema=None)"
]

The tool did not run. The agent stopped at approve, and the payload you passed to interrupt() is in the response under __interrupt__.

A paused run reports completed

The run itself finished: it ran until the interrupt and returned. What marks it as paused is __interrupt__ in its output, not the run's status. A client that needs to know whether to ask a human should look for that key.

The thread is still holding the pending decision. Confirm with the thread state, which shows the model's tool call and nothing after it:

curl -s "http://localhost:8090/threads/$THREAD/state" | jq '.values.messages | map(.type)'

[
  "human",
  "ai"
]

There is no tool message, because the tool hasn't run.

Step 3: Resume with a Decision

Approve by running the thread again. The request's input becomes the value interrupt() returns, so send the shape approve checks for:

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"approved": true}}' \
    | jq '[.output.messages[] | {type, content}]'

This time the graph continues past approve: the tool runs, and the model summarizes its result.

[
  { "type": "human", "content": "fetch https://httpbin.org/get" },
  { "type": "ai",    "content": "" },
  { "type": "tool",  "content": "HTTP 200: { \"args\": {}, ..." },
  { "type": "ai",    "content": "The request succeeded. ..." }
]

You didn't resend the question or the thread ID: the history came from the checkpoint. The tool message is the proof the approval gate opened.

Step 4: Reject a Tool Call

Start a new request on a fresh thread and reject it:

THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "fetch https://example.com"}]}}' \
    | jq '.output.__interrupt__ != null'

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"approved": false}}' \
    | jq '[.output.messages[] | {type, content}]'

The first call prints true: the run paused. The second shows the rejection reaching the model as the tool's result, and the tool never ran:

[
  { "type": "human", "content": "fetch https://example.com" },
  { "type": "ai",    "content": "" },
  { "type": "tool",  "content": "The user rejected this tool call." },
  { "type": "ai",    "content": "I couldn't retrieve the content from the specified URL. ..." }
]

Small models call tools eagerly

The default llama3.2:1b model sometimes asks for a tool even for a plain question. With the approval step in place that is harmless, and it is a good way to see the gate working: ask something ordinary and reject whatever it proposes.

Step 5: Stream a Run

So far each request waited for the run to finish. To watch it happen, start the run with "stream": true and follow its SSE stream. A run created without wait returns immediately with its run_id:

THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

RUN=$(curl -sX POST "http://localhost:8090/threads/$THREAD/runs" \
    -H "Content-Type: application/json" \
    -d '{"stream": true, "input": {"messages": [{"role": "user", "content": "fetch https://httpbin.org/get"}]}}' \
    | jq -r '.run_id')

curl -N "http://localhost:8090/threads/$THREAD/runs/$RUN/stream"

-N turns off curl's output buffering so each event prints as it arrives. You see one event per step of the graph, then the pause:

event: updates
data: {"chat":{"messages":[{"content":"","tool_calls":[{"args":{"url":"https://httpbin.org/get"},"name":"fetch_url", ...}],"type":"ai"}]}}

event: updates
data: {"__interrupt__":"(Interrupt(value={'message': 'Approve these tool calls?', ...}, ...),)"}

event: done
data: {}

Each updates event is named for the node that produced it and carries that node's output. The __interrupt__ event is the pause. done closes the stream: it means the run ended, not that the work is finished, so the pause ends the stream just as a completed run would.

Now resume with streaming on, to watch the rest of the work:

RUN=$(curl -sX POST "http://localhost:8090/threads/$THREAD/runs" \
    -H "Content-Type: application/json" \
    -d '{"stream": true, "input": {"approved": true}}' \
    | jq -r '.run_id')

curl -N "http://localhost:8090/threads/$THREAD/runs/$RUN/stream"

event: updates
data: {"approve":null}

event: updates
data: {"tools":{"messages":[{"content":"HTTP 200: ...","name":"fetch_url", ...}]}}

event: updates
data: {"chat":{"messages":[{"content":"The request succeeded. ...", ...}]}}

event: done
data: {}

You see the approval pass (approve returns nothing to record), the tool's result, and the model's answer, each as it happens. A UI would render these as progress and show an approval prompt when it sees __interrupt__.

Other Stream Shapes

The default stream emits one event per graph step. Other LangGraph stream modes, and chunks your own code emits mid-node (such as individual tokens), are covered in the Streaming guide. The done event and the curl -N pattern are the same for all of them.

Next Steps

You now have an agent that asks before it acts, and a client can watch it work.

• Human-in-the-Loop: more patterns, including multi-step approvals and timeouts
• Streaming: all three SSE endpoints, event types, and client examples in Python and JavaScript
• Runs API: run creation, status and the stream endpoints
• Tracing Your Agent: see the same runs, including the pause and resume, as traces