Skip to main content

Strands Graph Approval Agent

Two-node Strands Graph demonstrating MultiAgent persistence and interrupt/resume approval.

Difficulty: Advanced

What You'll Learn​

  • Building a two-node Graph with GraphBuilder
  • MultiAgent (Graph/Swarm) checkpointing via serialize_state()/deserialize_state() — restored automatically across a fresh guest process
  • Interrupt/resume for a tool call nested inside a graph node, using the same event.interrupt() contract as a plain Agent
  • LiteLLM-backed model configuration — AgentVisor's proven production route to Bedrock
  • Credential brokering for ANTHROPIC_API_KEY (never exposed in the sandbox)

Setup​

agentvisor template create strands/graph-approval-agent
cd graph-approval-agent

export ANTHROPIC_API_KEY=your-api-key-here
temporal server start-dev &
agentvisor serve . --sandbox=none

Test It​

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

# Submit a topic — the graph drafts, then pauses for approval before publishing
curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"message": "our new dark mode feature"}}' \
    | tee /tmp/run1.json | jq '.output'

# Approve, resuming on the same thread
INTERRUPT_ID=$(jq -r '.output.interrupts[0].interruptId' /tmp/run1.json)
curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=60s" \
    -H "Content-Type: application/json" \
    -d "{\"input\": {\"resume\": [{\"interruptResponse\": {\"interruptId\": \"$INTERRUPT_ID\", \"response\": \"approved\"}}]}}" \
    | jq '.output'

How It Works​

  1. Graph: drafter (an Agent) writes a short announcement from the user's topic; GraphBuilder wires its output as input to publisher (a second Agent)
  2. Selection: mav-agent-config.yaml explicitly sets framework.provider: "strands" — Strands has no discovery file for AgentVisor to auto-detect
  3. Execution: python3 -m agentvisor.strands.runner resolves root_agent — the same Agent/Graph/Swarm-agnostic resolution used by every Strands example
  4. Interrupt: publisher's ApprovalHook calls event.interrupt() from a BeforeToolCallEvent callback before publish_announcement runs, pausing the whole graph
  5. State persistence: the graph's serialize_state() — including drafter's completed output and publisher's pending interrupt — is saved via checkpoint.save() after the run and restored via checkpoint.load("")/deserialize_state() on the resuming run, in a fresh guest process
  6. Resume: the caller submits a resume payload carrying Strands' own InterruptResponseContent shape; publisher continues from exactly where it paused

MultiAgent Checkpointing​

AspectDetail
MechanismGraph.serialize_state() / Graph.deserialize_state() (also implemented by Swarm)
Storageagentvisor.checkpoint.save()/load(), Temporal-backed, per thread — the same simple API a plain Agent uses
ScopeNot cross-turn conversation memory the way a plain Agent's snapshot is — a Graph/Swarm that has already completed resets every node's message history on the next restore. What persists is in-flight task and interrupt/resume durability, exactly what this example exercises
DispatchThe runner detects whether root_agent exposes take_snapshot/load_snapshot (plain Agent) or serialize_state/deserialize_state (Graph/Swarm) and picks the matching strategy automatically
CompatibilityNot cross-compatible with the LangGraph AgentVisorCheckpointer — see Checkpointing

See the Strands Agents Guide for the full picture, including known limitations (cumulative execution limits across runs, no cross-turn memory after a completed graph run) and what to use instead if your agent needs cross-turn memory.

Project Files​

strands/graph-approval-agent/
├── agent.py               # Graph: drafter -> publisher, with an approval-gated tool
├── mav-agent-config.yaml  # REQUIRED: selects the strands framework provider
├── requirements.txt       # Python dependencies
├── agentvisor.yaml        # AgentVisor config (proxy credential substitution)
└── policies/
    └── domain.yml         # Authorization policies

agent.py (key section)​

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry
from strands.multiagent.graph import GraphBuilder
from agentvisor.strands import register_agent

@tool
def publish_announcement(text: str) -> str:
    return f"Published: {text}"

class ApprovalHook(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
        registry.add_callback(BeforeToolCallEvent, self.approve)

    def approve(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] != "publish_announcement":
            return
        approval = event.interrupt("publish_approval", reason="...")
        if approval != "approved":
            event.cancel_tool = "approval was not granted"

drafter = Agent(name="drafter", model=..., system_prompt="...")
publisher = Agent(name="publisher", model=..., tools=[publish_announcement], hooks=[ApprovalHook()])

builder = GraphBuilder()
builder.add_node(drafter, "drafter")
builder.add_node(publisher, "publisher")
builder.add_edge("drafter", "publisher")
root_agent = builder.build()
register_agent(root_agent)

Next Steps​