Strands AgentsのHuman in the Loopハンドラーを試してみた

uv add 'strands-agents==1.46.0'
uv add 'bedrock-agentcore==1.18.0'

from strands import Agent, tool
from strands.models import BedrockModel
from strands.vended_interventions.hitl import HumanInTheLoop

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"\n[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[delete_file],
    interventions=[HumanInTheLoop(ask="stdio")],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent("old_report.pdf を削除してください")

uv run hitl1_stdio.py

了解いたしました。old_report.pdf を削除します。
Tool #1: delete_file
Tool "delete_file" requires human approval. Input: {"path": "old_report.pdf"} (y/n): y

[delete_file] old_report.pdf を削除しました
old_report.pdf の削除が完了いたしました。

ファイル old_report.pdf を削除します。
Tool #1: delete_file
Tool "delete_file" requires human approval. Input: {"path": "old_report.pdf"} (y/n): n
申し訳ございませんが、セキュリティ上の理由からファイル削除操作には人間による承認が必要です。old_report.pdf を削除してもよろしいでしょうか？確認をお願いします。

from strands import Agent, tool
from strands.models import BedrockModel
from strands.vended_interventions.hitl import HumanInTheLoop

@tool
def read_file(path: str) -> str:
    """指定パスのファイルを読み取る"""
    print(f"\n[read_file] {path} を読み取りました")
    return "log_retention: 7days\nlog_dir: /tmp/old-logs"

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"\n[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[read_file, delete_file],
    interventions=[
        HumanInTheLoop(
            ask="stdio",
            allowed_tools=["read_file"],
        )
    ],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent("config.yaml を読んでから /tmp/old-logs を削除してください")

uv run python hitl2_allowed.py

Tool #1: read_file

[read_file] config.yaml を読み取りました
config.yaml を確認しました。次に /tmp/old-logs を削除します。
Tool #2: delete_file
Tool "delete_file" requires human approval. Input: {"path": "/tmp/old-logs"} (y/n): y

[delete_file] /tmp/old-logs を削除しました
完了しました。

from strands import Agent, tool
from strands.models import BedrockModel
from strands.vended_interventions.hitl import HumanInTheLoop

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"\n[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[delete_file],
    interventions=[
        HumanInTheLoop(
            ask="stdio",
            enable_trust=True,
        )
    ],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

result = agent("temp1.log を削除して、成功したら続けて temp2.log と temp3.log も1つずつ削除してください")

uv run python hitl3_trust.py

temp1.log を削除します。
Tool #1: delete_file
Tool "delete_file" requires human approval. Input: {"path": "temp1.log"} (y/n/t): t

[delete_file] temp1.log を削除しました
temp1.log の削除が成功しました。次に temp2.log を削除します。
Tool #2: delete_file

[delete_file] temp2.log を削除しました
temp2.log の削除が成功しました。最後に temp3.log を削除します。
Tool #3: delete_file

[delete_file] temp3.log を削除しました
完了しました！

from bedrock_agentcore.memory import MemoryClient

client = MemoryClient(region_name="us-east-1")
memory = client.create_memory(
    name="hitlTrustDemo",
    description="HumanInTheLoop Trust persistence demo",
)
print(f"Memory ID: {memory['id']}")

from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import (
    AgentCoreMemorySessionManager,
)
from strands import Agent, tool
from strands.models import BedrockModel
from strands.vended_interventions.hitl import HumanInTheLoop

MEMORY_ID = "your-memory-id"
SESSION_ID = "user-session-001"

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"\n[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

config = AgentCoreMemoryConfig(
    memory_id=MEMORY_ID,
    session_id=SESSION_ID,
    actor_id="demo-user",
)

with AgentCoreMemorySessionManager(
    agentcore_memory_config=config, region_name="us-east-1"
) as session_manager:
    agent = Agent(
        model=model,
        tools=[delete_file],
        interventions=[HumanInTheLoop(ask="stdio", enable_trust=True)],
        session_manager=session_manager,
    )

    agent("temp.log を削除してください")

temp.log を削除します。
Tool #1: delete_file
Tool "delete_file" requires human approval. Input: {"path": "temp.log"} (y/n/t): t

[delete_file] temp.log を削除しました
temp.log ファイルの削除が完了しました。

{
  "hitl:trusted_tools": [
    "delete_file"
  ]
}

# プロセス再起動後、同じSESSION_IDで新しいAgentを作成
config = AgentCoreMemoryConfig(
    memory_id=MEMORY_ID,
    session_id=SESSION_ID,  # 同じセッションID
    actor_id="demo-user",
)

with AgentCoreMemorySessionManager(
    agentcore_memory_config=config, region_name="us-east-1"
) as session_manager:
    agent = Agent(
        model=model,
        tools=[delete_file],
        interventions=[HumanInTheLoop(ask="stdio", enable_trust=True)],
        session_manager=session_manager,
    )

    # stateが復元されているか確認
    print(agent.state.get())
    # => {'hitl:trusted_tools': ['delete_file']}

    agent("another.log を削除してください")

{'hitl:trusted_tools': ['delete_file']}
another.log を削除します。
Tool #1: delete_file
another.log ファイルの削除が完了しました。

from bedrock_agentcore.runtime import BedrockAgentCoreApp

from strands import Agent, tool
from strands.models import BedrockModel
from strands.vended_interventions.hitl import HumanInTheLoop

@tool
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する"""
    print(f"[delete_file] {path} を削除しました")
    return f"{path} を削除しました"

model = BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0")

agent = Agent(
    model=model,
    tools=[delete_file],
    interventions=[HumanInTheLoop()],
    system_prompt="ユーザーに確認を求めず、自律的に判断してタスクを完了してください。",
)

app = BedrockAgentCoreApp()

@app.entrypoint
def invoke(payload):
    if payload.get("responses"):
        result = agent(
            [
                {
                    "interruptResponse": {
                        "interruptId": r["interrupt_id"],
                        "response": r.get("response"),
                    }
                }
                for r in payload["responses"]
            ]
        )
    else:
        result = agent(payload.get("message"))

    if result.stop_reason == "interrupt":
        return {
            "status": "pending_approval",
            "interrupts": [
                {
                    "interrupt_id": interrupt.id,
                    "prompt": interrupt.reason,
                }
                for interrupt in result.interrupts
            ],
        }
    return {"status": "done", "message": str(result)}

if __name__ == "__main__":
    app.run()

uv run python hitl6_agentcore.py

curl -s -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"message": "old_report.pdf を削除してください"}'

{
    "status": "pending_approval",
    "interrupts": [
        {
            "interrupt_id": "v1:before_tool_call:tooluse_bM0fXqg8P1ZC1AkXqQEF1v:b9e6f0f7-048e-5f43-a7b0-2b41b64a57ec",
            "prompt": "Tool \"delete_file\" requires human approval. Input: {\"path\": \"old_report.pdf\"}"
        }
    ]
}

curl -s -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"responses": [{"interrupt_id": "v1:before_tool_call:tooluse_bM0fXqg8P1ZC1AkXqQEF1v:b9e6f0f7-048e-5f43-a7b0-2b41b64a57ec", "response": "yes"}]}'

{
    "status": "done",
    "message": "完了しました。ファイル「old_report.pdf」を削除しました。\n"
}

{
    "status": "done",
    "message": "申し訳ありませんが、ファイルの削除には人的確認が必要です。backup.zip を本当に削除してよろしいですか？削除すると復元できない可能性があります。確認をお願いします。\n"
}

// invokeAgent はサーバー経由でInvokeAgentRuntimeを呼ぶ関数の想定
let res = await invokeAgent({ message: userInput });

while (res.status === "pending_approval") {
  const responses = res.interrupts.map((intr) => ({
    interrupt_id: intr.interrupt_id,
    response: window.confirm(intr.prompt) ? "yes" : "no",
  }));
  res = await invokeAgent({ responses });
}

showMessage(res.message);

async def ask(prompt: str) -> str:
    # Slack DMやWebモーダルなど、好きなUIで質問して回答を返す
    return await ask_user_via_slack(prompt)

agent = Agent(
    tools=[delete_file],
    interventions=[HumanInTheLoop(ask=ask)],
)

Step 1: interruptで停止
interrupt_id: v1:before_tool_call:tooluse_T1BF...
messages count: 2

Step 2: agentを破棄（microVM停止を模擬）

Step 3: 同じsession_idで復元してresume
復元後のmessages count: 2

[delete_file] important.doc を削除しました
✅ AgentCore Memoryでinterrupt状態を復元してresumeできた

HumanInTheLoop(
    ask="stdio",
    evaluate=lambda response: isinstance(response, str) and response.lower() == "confirm",
)