Building Human-in-the-Loop (HITL) AI Agents with Strands Agents, AG-UI, and CopilotKit on Amazon Bedrock AgentCore

import { CopilotRuntime, ExperimentalEmptyAdapter, copilotRuntimeNodeHttpEndpoint } from "@copilotkit/runtime";
import { HttpAgent } from "@ag-ui/client";
const AGENT_URL = process.env.AGENT_URL; // Points to your FastAPI proxy
const runtime = new CopilotRuntime({
agents: {
expense_agent: new HttpAgent({ url: AGENT_URL }),
},
});

from botocore.session import Session as BotoSession
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

# Build the invocations URL
parsed = urlparse(AGENTCORE_ENDPOINT)
invocations_path = parsed.path.rstrip('/') + '/invocations'
endpoint_url = urlunparse(parsed._replace(path=invocations_path))

# Sign with SigV4
boto_session = BotoSession()
credentials = boto_session.get_credentials().get_frozen_credentials()
aws_request = AWSRequest(
    method="POST", url=endpoint_url, data=body_bytes,
    headers={"Content-Type": "application/json", "Accept": "text/event-stream"},
)
SigV4Auth(credentials, "bedrock-agentcore", "us-east-1").add_auth(aws_request)
The AGENTCORE_ENDPOINT includes the accountId query parameter:
https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/{runtime-id}?accountId={account-id}


import { useHumanInTheLoop, ToolCallStatus } from "@copilotkit/react-core/v2";
import { z } from "zod";

function ExpenseApproval() {
  useHumanInTheLoop({
    name: "approve_expense",
    description: "Request manager approval for expenses $500 or more",
    parameters: z.object({
      description: z.string(),
      amount: z.number(),
      category: z.string(),
    }),
    render: ({ args, status, respond, result }) => {
      if (status === ToolCallStatus.Executing && respond) {
        return (
          <div className="approval-card">
            <h3>Manager Approval Required</h3>
            <p>Description: {args.description}</p>
            <p>Amount: ${args.amount?.toFixed(2)}</p>
            <p>Category: {args.category}</p>
            <button onClick={() => respond(
              "APPROVED: Manager approved this expense. Proceed with process_expense."
            )}>Approve</button>
            <button onClick={() => respond(
              "REJECTED: Manager rejected this expense. Do NOT call process_expense."
            )}>Reject</button>
          </div>
        );
      }
      if (status === ToolCallStatus.Complete && result) {
        const isApproved = typeof result === "string" && result.startsWith("APPROVED");
        return <p>{isApproved ? "✅ Approved" : "❌ Rejected"}: {args.description}</p>;
      }
      return null;
    },
  }, []);
  return null;
}


SYSTEM_PROMPT = """You are an expense processing agent. When a user submits an expense:
1. Determine the category from: travel, meals, supplies, equipment, or other
2. Extract the description and dollar amount
3. If the amount is $500 or more, call approve_expense FIRST
4. If under $500, skip approval and go directly to step 6
5. CRITICAL: After calling approve_expense, read the tool result:
   - If it starts with "APPROVED", proceed to step 6
   - If it starts with "REJECTED", STOP. Do NOT call process_expense.
6. Only if approved: Use process_expense with the description, amount, and category
7. Report the result

IMPORTANT: You must NEVER call process_expense after a rejection."""

from ag_ui_strands import StrandsAgent
from ag_ui_strands.config import StrandsAgentConfig

def _context_builder(run_input, default_msg):
    if not run_input.messages:
        return default_msg
    last = run_input.messages[-1]
    if last.role == "tool" and last.content:
        content = last.content if isinstance(last.content, str) else str(last.content)
        expense_summary = ""
        for msg in run_input.messages:
            if msg.role == "user" and msg.content and isinstance(msg.content, str):
                expense_summary = msg.content
        if content.startswith("APPROVED"):
            return (
                f"The manager approved the expense and it has already been processed. "
                f"The original request was: '{expense_summary}'. Confirm to the user briefly."
            )
        if content.startswith("REJECTED"):
            return (
                f"REJECTED: Manager rejected this expense. Do NOT call process_expense. "
                f"The original request was: '{expense_summary}'. "
                f"Inform the user their expense was rejected."
            )
    return default_msg

config = StrandsAgentConfig(state_context_builder=_context_builder)
agui_agent = StrandsAgent(agent=agent, name="expense_agent", config=config)


from ag_ui_strands import StrandsAgent, create_strands_app
from ag_ui_strands.config import StrandsAgentConfig
from agent.agent import create_agent

agent = create_agent()
config = StrandsAgentConfig(state_context_builder=_context_builder)
agui_agent = StrandsAgent(
    agent=agent,
    name="expense_agent",
    description="Expense processing agent with HITL approval",
    config=config,
)
app = create_strands_app(agui_agent, path="/invocations", ping_path="/ping")


agentcore configure -e backend/entrypoint.py --protocol AGUI
agentcore deploy

<CopilotKit runtimeUrl="/copilotkit" agent="expense_agent">
  <ExpenseApproval />
  <CopilotChat labels={{ title: "Expense Agent" }} className="chat-panel" />
</CopilotKit>