HITL Overview

"use client";import React from "react";import {  CopilotKit,  CopilotChat,  useHumanInTheLoop,  useConfigureSuggestions,} from "@copilotkit/react-core/v2";import { z } from "zod";import type { TimeSlot } from "./time-picker-card";import { TimePickerCard } from "./time-picker-card";const DEFAULT_SLOTS: TimeSlot[] = [  { label: "Tomorrow 10:00 AM", iso: "2026-04-19T10:00:00-07:00" },  { label: "Tomorrow 2:00 PM", iso: "2026-04-19T14:00:00-07:00" },  { label: "Monday 9:00 AM", iso: "2026-04-21T09:00:00-07:00" },  { label: "Monday 3:30 PM", iso: "2026-04-21T15:30:00-07:00" },];export default function HitlInChatDemo() {  return (    <CopilotKit runtimeUrl="/api/copilotkit" agent="hitl-in-chat">      <div className="flex justify-center items-center h-screen w-full">        <div className="h-full w-full max-w-4xl">          <Chat />        </div>      </div>    </CopilotKit>  );}function Chat() {  useConfigureSuggestions({    suggestions: [      {        title: "Book a call with sales",        message:          "Please book an intro call with the sales team to discuss pricing.",      },      {        title: "Schedule a 1:1 with Alice",        message: "Schedule a 1:1 with Alice next week to review Q2 goals.",      },    ],    available: "always",  });  useHumanInTheLoop({    agentId: "hitl-in-chat",    name: "book_call",    description:      "Ask the user to pick a time slot for a call. The picker UI presents fixed candidate slots; the user's choice is returned to the agent.",    parameters: z.object({      topic: z        .string()        .describe("What the call is about (e.g. 'Intro with sales')"),      attendee: z        .string()        .describe("Who the call is with (e.g. 'Alice from Sales')"),    }),    render: ({ args, status, respond }: any) => (      <TimePickerCard        topic={args?.topic ?? "a call"}        attendee={args?.attendee}        slots={DEFAULT_SLOTS}        status={status}        onSubmit={(result) => respond?.(result)}      />    ),  });  return <CopilotChat agentId="hitl-in-chat" className="h-full rounded-2xl" />;}

"""The agent behind gen-ui-interrupt and interrupt-headless.``schedule_meeting`` is marked ``needs_approval``, so the SDK pauses the runbefore the tool executes and the adapter ends the AG-UI run with an interruptfor the call, its arguments on ``metadata.tool_input``. The page renders thetime picker from them and resolves the interrupt with ``{"approved": true,"chosen_time": ..., "chosen_label": ...}``, or ``{"approved": false}`` whenthe user cancels. The next request carries that answer as a ``resume`` entry,and the adapter runs an approved call before the model sees the conversationagain.Nothing is kept on the server between the two runs: the paused call lives inthe conversation the page holds."""from collections.abc import AsyncIteratorfrom contextlib import aclosingfrom contextvars import ContextVarfrom typing import Anyfrom ag_ui.core import BaseEvent, RunAgentInputfrom ag_ui_openai_agents import OpenAIAgentsAgentfrom agents import Agent, function_toolfrom agents.tool_context import ToolContextfrom demo_agents._common import MODEL, chat_settingsSYSTEM_PROMPT = """You are a scheduling assistant.Whenever the user asks you to book a call or schedule a meeting, you MUST callthe `schedule_meeting` tool. Pass a short `topic` describing the purpose and, ifknown, an `attendee` describing who the meeting is with.The tool shows the user a time picker and returns their choice. Then brieflyconfirm whether the meeting was scheduled and at what time, or note that theuser cancelled. Do not ask for approval yourself: always call the tool and letthe picker handle the decision. Keep responses short and friendly.Never claim a meeting is scheduled unless the tool result says so."""# This request's interrupt answers, by tool call id (see SchedulerAgent)._answers: ContextVar[dict[str, dict[str, Any]]] = ContextVar("meeting_answers")@function_tool(needs_approval=True)def schedule_meeting(ctx: ToolContext[Any], topic: str, attendee: str = "") -> str:    """Ask the user to pick a meeting time, then confirm what was scheduled.    Args:        topic: Short description of the meeting purpose.        attendee: Who the meeting is with, if known.    """    # Runs once the user approved, with the time they picked beside it.    answer = _answers.get({}).get(ctx.tool_call_id, {})    label = answer.get("chosen_label") or answer.get("chosen_time")    if not label:        return f"User did not pick a time. Meeting NOT scheduled: {topic}"    return f"Meeting scheduled for {label}: {topic}"interrupt_agent = Agent(    name="scheduler",    instructions=SYSTEM_PROMPT,    model=MODEL,    model_settings=chat_settings(parallel_tool_calls=False),    tools=[schedule_meeting],)class SchedulerAgent(OpenAIAgentsAgent):    """Lets ``schedule_meeting`` read the time the user picked.    The adapter runs an approved call with its arguments alone: an Agents SDK    approval is a yes or a no, with nothing to hand the tool. The picker    sends the time beside the approval, so this keeps each answer where the    tool can read it.    """    async def run_streamed(self, input: RunAgentInput) -> AsyncIterator[BaseEvent]:        _answers.set(            {                entry.interrupt_id: entry.payload                for entry in input.resume or []                if isinstance(entry.payload, dict)            }        )        async with aclosing(super().run_streamed(input)) as events:            async for event in events:                yield event