Headless Interrupts

"""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