{"id":825,"date":"2026-05-01T13:06:51","date_gmt":"2026-05-01T05:06:51","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=825"},"modified":"2026-05-01T13:06:51","modified_gmt":"2026-05-01T05:06:51","slug":"a-coding-deep-dive-into-agentic-ui-generative-ui-state-synchronization-and-interrupt-driven-approval-flows","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=825","title":{"rendered":"A Coding Deep Dive into Agentic UI, Generative UI, State Synchronization, and Interrupt-Driven Approval Flows"},"content":{"rendered":"

In this tutorial, we build the entire Agentic UI stack from the ground up using plain Python, without relying on external frameworks to abstract away the core ideas. We implement the AG-UI event stream to make agent behavior observable in real time, and we bring in A2UI as a declarative layer that allows interfaces to be defined as structured JSON rather than executable code. As we progress, we enable an LLM to generate full user interfaces from natural language, synchronize agent and UI state through JSON Patch updates, and enforce human-in-the-loop safety for critical actions. Also, we gain a clear, end-to-end understanding of how agent reasoning transforms into interactive, protocol-compliant user interfaces.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import subprocess, sys\nfor pkg in [\"openai\", \"rich\", \"pydantic\"]:\n   subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", \"-q\", pkg])\n\n\nimport os, getpass\n\n\nif os.environ.get(\"OPENAI_API_KEY\"):\n   API_KEY = os.environ[\"OPENAI_API_KEY\"]\n   print(\"\"\u2705\" Using OPENAI_API_KEY from environment.\")\nelse:\n   try:\n       from google.colab import userdata\n       API_KEY = userdata.get(\"OPENAI_API_KEY\")\n       print(\"\"\u2705\" Using OPENAI_API_KEY from Colab Secrets.\")\n   except Exception:\n       API_KEY = getpass.getpass(\"\"\ud83d\udd11\" Enter your OpenAI API key (hidden): \")\n       print(\"\"\u2705\" API key received.\")\n\n\nBASE_URL = os.environ.get(\"OPENAI_BASE_URL\", \"https:\/\/api.openai.com\/v1\")\nMODEL    = os.environ.get(\"OPENAI_MODEL\", \"gpt-4o-mini\")\n\n\nimport json, re, time, uuid, copy, textwrap\nfrom enum import Enum\nfrom dataclasses import dataclass, field, asdict\nfrom typing import Any, Optional, Generator\nfrom pydantic import BaseModel, Field\n\n\nfrom openai import OpenAI\nfrom rich.console import Console\nfrom rich.panel import Panel\nfrom rich.table import Table\nfrom rich.tree import Tree\nfrom rich.text import Text\nfrom rich.markdown import Markdown\nfrom rich import box\n\n\nconsole = Console(width=105)\nclient  = OpenAI(api_key=API_KEY, base_url=BASE_URL)\n\n\ndef llm(messages, **kw):\n   try:\n       return client.chat.completions.create(model=MODEL, messages=messages, temperature=0.2, **kw)\n   except Exception as e:\n       console.print(f\"[red]LLM error: {e}[\/]\")\n       return None\n\n\ndef hdr(n, title, sub=\"\"):\n   console.print()\n   console.rule(f\"[bold cyan]SECTION {n}\", style=\"cyan\")\n   body = f\"[bold white]{title}[\/]n[dim]{sub}[\/]\" if sub else f\"[bold white]{title}[\/]\"\n   console.print(Panel(body, border_style=\"cyan\", padding=(1, 2)))\n\n\n\n\nhdr(1, \"AG-UI Protocol \u2014 Event System\",\n   \"The real AG-UI protocol uses ~16 event types streamed via SSE.n\"\n   \"We implement all core event types and a streaming emitter in pure Python.\")\n\n\nclass AGUIEventType(str, Enum):\n   RUN_STARTED          = \"RUN_STARTED\"\n   RUN_FINISHED         = \"RUN_FINISHED\"\n   RUN_ERROR            = \"RUN_ERROR\"\n   TEXT_MESSAGE_START   = \"TEXT_MESSAGE_START\"\n   TEXT_MESSAGE_CONTENT = \"TEXT_MESSAGE_CONTENT\"\n   TEXT_MESSAGE_END     = \"TEXT_MESSAGE_END\"\n   TOOL_CALL_START      = \"TOOL_CALL_START\"\n   TOOL_CALL_ARGS       = \"TOOL_CALL_ARGS\"\n   TOOL_CALL_RESULT     = \"TOOL_CALL_RESULT\"\n   TOOL_CALL_END        = \"TOOL_CALL_END\"\n   STATE_SNAPSHOT       = \"STATE_SNAPSHOT\"\n   STATE_DELTA          = \"STATE_DELTA\"\n   INTERRUPT            = \"INTERRUPT\"\n   CUSTOM               = \"CUSTOM\"\n   STEP_STARTED         = \"STEP_STARTED\"\n   STEP_FINISHED        = \"STEP_FINISHED\"\n\n\n\n\n@dataclass\nclass AGUIEvent:\n   type: AGUIEventType\n   data: dict = field(default_factory=dict)\n   event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])\n   timestamp: float = field(default_factory=time.time)\n\n\n   def to_sse(self) -> str:\n       payload = {\"type\": self.type.value, \"id\": self.event_id, **self.data}\n       return f\"event: ag-uindata: {json.dumps(payload)}nn\"\n\n\n   def to_json(self) -> dict:\n       return {\"type\": self.type.value, \"id\": self.event_id, \"ts\": self.timestamp, **self.data}\n\n\n\n\nclass AGUIEventStream:\n   def __init__(self):\n       self.events: list[AGUIEvent] = []\n       self.listeners: list = []\n\n\n   def emit(self, event: AGUIEvent):\n       self.events.append(event)\n       for listener in self.listeners:\n           listener(event)\n\n\n   def on(self, callback):\n       self.listeners.append(callback)\n\n\n   def replay(self) -> list[dict]:\n       return [e.to_json() for e in self.events]\n\n\n\n\ndef demo_agui_lifecycle():\n   stream = AGUIEventStream()\n\n\n   event_colors = {\n       \"RUN_\": \"bold green\", \"TEXT_\": \"cyan\", \"TOOL_\": \"magenta\",\n       \"STATE_\": \"yellow\", \"INTERRUPT\": \"bold red\", \"STEP_\": \"dim\",\n   }\n\n\n   def frontend_listener(event: AGUIEvent):\n       color = \"white\"\n       for prefix, c in event_colors.items():\n           if event.type.value.startswith(prefix):\n               color = c\n               break\n       detail = json.dumps(event.data)[:80] if event.data else \"\"\n       console.print(f\"  [{color}]\"\u26a1\" {event.type.value:.<28}[\/] {detail}\")\n\n\n   stream.on(frontend_listener)\n\n\n   run_id = str(uuid.uuid4())[:8]\n   console.print(\"[bold]Simulating full AG-UI agent run...[\/]n\")\n\n\n   stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {\"run_id\": run_id}))\n\n\n   stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {\"step\": \"analyzing_query\", \"label\": \"Understanding request\"}))\n   stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {\"step\": \"analyzing_query\"}))\n\n\n   msg_id = str(uuid.uuid4())[:8]\n   stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {\"message_id\": msg_id, \"role\": \"assistant\"}))\n   for chunk in [\"I'll \", \"look up \", \"the data \", \"and build \", \"a dashboard \", \"for you.\"]:\n       stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {\"message_id\": msg_id, \"delta\": chunk}))\n   stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {\"message_id\": msg_id}))\n\n\n   tool_id = str(uuid.uuid4())[:8]\n   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_START, {\"tool_call_id\": tool_id, \"name\": \"query_database\"}))\n   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_ARGS, {\"tool_call_id\": tool_id, \"args_delta\": '{\"query\": \"SELECT revenue FROM sales\"}'}))\n   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_RESULT, {\"tool_call_id\": tool_id, \"result\": [{\"month\": \"Jan\", \"revenue\": 42000}, {\"month\": \"Feb\", \"revenue\": 58000}]}))\n   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_END, {\"tool_call_id\": tool_id}))\n\n\n   stream.emit(AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {\n       \"state\": {\"active_agent\": \"DataAnalyst\", \"stage\": \"rendering\", \"progress\": 0.75}\n   }))\n   stream.emit(AGUIEvent(AGUIEventType.STATE_DELTA, {\n       \"delta\": [{\"op\": \"replace\", \"path\": \"\/progress\", \"value\": 1.0}]\n   }))\n\n\n   stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {\n       \"reason\": \"high_risk_action\",\n       \"description\": \"Agent wants to send an email to all 5,000 customers.\",\n       \"options\": [\"approve\", \"reject\", \"modify\"],\n   }))\n\n\n   stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {\"run_id\": run_id, \"status\": \"completed\"}))\n\n\n   console.print(Panel(\n       stream.events[3].to_sse(),\n       title=\"[bold]Example SSE wire format (how it looks on the network)\",\n       border_style=\"dim\",\n   ))\n\n\n   table = Table(title=\"AG-UI Event Stream Summary\", box=box.ROUNDED)\n   table.add_column(\"Category\", style=\"cyan\", width=15)\n   table.add_column(\"Events\", justify=\"center\", style=\"green\")\n   counts = {}\n   for e in stream.events:\n       cat = e.type.value.rsplit(\"_\", 1)[0] if \"_\" in e.type.value else e.type.value\n       counts[cat] = counts.get(cat, 0) + 1\n   for cat, n in counts.items():\n       table.add_row(cat, str(n))\n   table.add_row(\"[bold]TOTAL\", f\"[bold]{len(stream.events)}\")\n   console.print(table)\n\n\ndemo_agui_lifecycle()<\/code><\/pre>\n<\/div>\n<\/div>\n
We start by building the backbone of every agentic frontend: the AG-UI event stream. We implement all 16 event types from the real AG-UI specification, lifecycle events, token-by-token text streaming, streamed tool calls, state snapshots, deltas, and interrupt signals, and serialize them into the SSE wire format that production systems use over HTTP. We then wire up a frontend listener that reacts to each event as it arrives, simulating the exact experience a React or Flutter app would have consuming this stream.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
hdr(2, \"A2UI \u2014 Declarative Component Trees\",\n   \"Google's A2UI spec: agents emit flat JSON component lists with ID refs.n\"\n   \"The client's widget registry maps types \u2192 native widgets.n\"\n   \"Safe like data, expressive like code. No executable code sent.\")\n\n\nclass A2UIMessageType(str, Enum):\n   CREATE_SURFACE      = \"createSurface\"\n   UPDATE_COMPONENTS   = \"updateComponents\"\n   UPDATE_DATA_MODEL   = \"updateDataModel\"\n   DELETE_SURFACE       = \"deleteSurface\"\n\n\n\n\n@dataclass\nclass A2UIComponent:\n   id: str\n   type: str\n   properties: dict = field(default_factory=dict)\n   children: list[str] = field(default_factory=list)\n\n\n   def to_dict(self) -> dict:\n       d = {\"id\": self.id, \"type\": self.type, **self.properties}\n       if self.children:\n           d[\"children\"] = self.children\n       return d\n\n\n\n\n@dataclass\nclass A2UIDataModel:\n   data: dict = field(default_factory=dict)\n\n\n   def get_binding(self, path: str) -> Any:\n       parts = [p for p in path.split(\"\/\") if p]\n       val = self.data\n       for p in parts:\n           if isinstance(val, dict):\n               val = val.get(p)\n           elif isinstance(val, list) and p.isdigit():\n               val = val[int(p)]\n           else:\n               return None\n       return val\n\n\n\n\n@dataclass\nclass A2UISurface:\n   surface_id: str\n   components: list[A2UIComponent] = field(default_factory=list)\n   data_model: A2UIDataModel = field(default_factory=A2UIDataModel)\n\n\n   def to_messages(self) -> list[dict]:\n       msgs = []\n       msgs.append({\n           \"type\": A2UIMessageType.CREATE_SURFACE.value,\n           \"surfaceId\": self.surface_id,\n       })\n       msgs.append({\n           \"type\": A2UIMessageType.UPDATE_COMPONENTS.value,\n           \"surfaceId\": self.surface_id,\n           \"components\": [c.to_dict() for c in self.components],\n       })\n       if self.data_model.data:\n           msgs.append({\n               \"type\": A2UIMessageType.UPDATE_DATA_MODEL.value,\n               \"surfaceId\": self.surface_id,\n               \"dataModel\": self.data_model.data,\n           })\n       return msgs\n\n\n\n\nclass WidgetRegistry:\n   def __init__(self):\n       self._renderers = {}\n\n\n   def register(self, component_type: str, render_fn):\n       self._renderers[component_type] = render_fn\n\n\n   def render(self, component: A2UIComponent, surface: A2UISurface, indent: int = 0):\n       fn = self._renderers.get(component.type)\n       if fn:\n           fn(component, surface, indent)\n       else:\n           pad = \"  \" * indent\n           console.print(f\"{pad}[dim]\u27e8{component.type} id={component.id}\u27e9 (no renderer)[\/]\")\n\n\n   def render_tree(self, surface: A2UISurface):\n       comp_map = {c.id: c for c in surface.components}\n       all_children = set()\n       for c in surface.components:\n           all_children.update(c.children)\n       roots = [c for c in surface.components if c.id not in all_children]\n\n\n       def _render(comp_id: str, indent: int):\n           comp = comp_map.get(comp_id)\n           if not comp:\n               return\n           self.render(comp, surface, indent)\n           for child_id in comp.children:\n               _render(child_id, indent + 1)\n\n\n       for root in roots:\n           _render(root.id, 0)\n\n\n\n\nregistry = WidgetRegistry()\n\n\ndef _resolve(comp, surface, key, default=None):\n   val = comp.properties.get(key, default)\n   binding = comp.properties.get(\"dataBinding\")\n   if binding and isinstance(binding, str) and binding.startswith(\"\/\"):\n       resolved = surface.data_model.get_binding(binding)\n       if resolved is not None:\n           return resolved\n   if isinstance(val, str) and val.startswith(\"\/\") and \"\/\" in val[1:]:\n       resolved = surface.data_model.get_binding(val)\n       if resolved is not None:\n           return resolved\n   return val\n\n\ndef _to_float(val, default=0.0):\n   if isinstance(val, (int, float)):\n       return float(val)\n   if isinstance(val, str):\n       cleaned = val.strip().rstrip(\"%\")\n       try:\n           f = float(cleaned)\n           if \"%\" in val or f > 1:\n               return f \/ 100.0\n           return f\n       except ValueError:\n           return default\n   return default\n\n\ndef render_card(comp, surface, indent):\n   pad = \"  \" * indent\n   title = str(_resolve(comp, surface, \"title\", \"Card\"))\n   console.print(f\"{pad}\u250c\u2500{'\u2500' * 50}\u2500\u2510\")\n   console.print(f\"{pad}\u2502 [bold]{title:^50}[\/] \u2502\")\n   console.print(f\"{pad}\u251c\u2500{'\u2500' * 50}\u2500\u2524\")\n   if not comp.children:\n       subtitle = str(_resolve(comp, surface, \"subtitle\", \"\"))\n       if subtitle:\n           console.print(f\"{pad}\u2502  {subtitle:<49}\u2502\")\n       console.print(f\"{pad}\u2514\u2500{'\u2500' * 50}\u2500\u2518\")\n\n\ndef render_text(comp, surface, indent):\n   pad = \"  \" * indent\n   text = _resolve(comp, surface, \"text\", \"\")\n   style = comp.properties.get(\"style\", \"body\")\n   styles = {\"headline\": \"bold white\", \"body\": \"white\", \"caption\": \"dim\", \"label\": \"bold cyan\"}\n   console.print(f\"{pad}[{styles.get(style, 'white')}]{text}[\/]\")\n\n\ndef render_button(comp, surface, indent):\n   pad = \"  \" * indent\n   label = str(_resolve(comp, surface, \"label\", \"Button\"))\n   variant = comp.properties.get(\"variant\", \"primary\")\n   colors = {\"primary\": \"bold white on blue\", \"secondary\": \"white on grey30\", \"danger\": \"bold white on red\"}\n   console.print(f\"{pad}  [{colors.get(variant, 'white')}]  {label}  [\/]\")\n\n\ndef render_text_field(comp, surface, indent):\n   pad = \"  \" * indent\n   label = comp.properties.get(\"label\", \"Input\")\n   placeholder = comp.properties.get(\"placeholder\", \"\")\n   console.print(f\"{pad}  {label}: [dim]\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510[\/]\")\n   console.print(f\"{pad}          [dim]\u2502 {placeholder:<25}\u2502[\/]\")\n   console.print(f\"{pad}          [dim]\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518[\/]\")\n\n\ndef render_row(comp, surface, indent):\n   pass\n\n\ndef render_column(comp, surface, indent):\n   pass\n\n\ndef render_image(comp, surface, indent):\n   pad = \"  \" * indent\n   alt = comp.properties.get(\"alt\", \"image\")\n   console.print(f\"{pad}  [dim]\"\ud83d\uddbc\"  [{alt}][\/]\")\n\n\ndef render_divider(comp, surface, indent):\n   pad = \"  \" * indent\n   console.print(f\"{pad}  {'\u2500' * 50}\")\n\n\ndef render_chip(comp, surface, indent):\n   pad = \"  \" * indent\n   label = str(_resolve(comp, surface, \"label\", \"\"))\n   console.print(f\"{pad}  [on grey23] {label} [\/]\")\n\n\ndef render_progress(comp, surface, indent):\n   pad = \"  \" * indent\n   raw_value = _resolve(comp, surface, \"value\", 0)\n   value = max(0.0, min(1.0, _to_float(raw_value, 0.0)))\n   label = str(_resolve(comp, surface, \"label\", \"\"))\n   bar_len = int(value * 40)\n   bar = f\"[green]{'\u2588' * bar_len}[\/][dim]{'\u2591' * (40 - bar_len)}[\/]\"\n   console.print(f\"{pad}  {label}: {bar} {value*100:.0f}%\")\n\n\nfor name, fn in [\n   (\"card\", render_card), (\"text\", render_text), (\"button\", render_button),\n   (\"text-field\", render_text_field), (\"row\", render_row), (\"column\", render_column),\n   (\"image\", render_image), (\"divider\", render_divider), (\"chip\", render_chip),\n   (\"progress-bar\", render_progress),\n]:\n   registry.register(name, fn)\n\n\n\n\nconsole.print(\"n[bold]Demo: A2UI booking form \u2014 agent generates a restaurant reservation UI[\/]n\")\n\n\nbooking_surface = A2UISurface(\n   surface_id=\"booking-form-1\",\n   components=[\n       A2UIComponent(\"root\", \"card\", {\"title\": \"\"\ud83c\udf7d\"  Reserve a Table\"}, children=[\"c1\", \"c2\", \"c3\", \"c4\", \"c5\", \"c6\"]),\n       A2UIComponent(\"c1\", \"text\", {\"text\": \"\", \"dataBinding\": \"\/restaurant\/name\", \"style\": \"headline\"}),\n       A2UIComponent(\"c2\", \"text\", {\"text\": \"\", \"dataBinding\": \"\/restaurant\/cuisine\", \"style\": \"caption\"}),\n       A2UIComponent(\"c3\", \"divider\", {}),\n       A2UIComponent(\"c4\", \"text-field\", {\"label\": \"Date\", \"placeholder\": \"YYYY-MM-DD\"}),\n       A2UIComponent(\"c5\", \"text-field\", {\"label\": \"Guests\", \"placeholder\": \"1-12\"}),\n       A2UIComponent(\"c6\", \"button\", {\"label\": \"Reserve Now\", \"variant\": \"primary\", \"action\": \"submit_booking\"}),\n   ],\n   data_model=A2UIDataModel({\"restaurant\": {\"name\": \"Chez Laurent\", \"cuisine\": \"French Contemporary \u2022 $$$$\"}})\n)\n\n\nconsole.print(Panel(\n   \"n\".join(json.dumps(m, indent=2)[:200] for m in booking_surface.to_messages()),\n   title=\"[bold]A2UI JSONL stream (what goes over the wire)\",\n   border_style=\"yellow\",\n))\n\n\nconsole.print(\"[bold]Rendered by client widget registry:[\/]n\")\nregistry.render_tree(booking_surface)\n\n\nconsole.print()\nt = Table(title=\"A2UI Flat Component List (Adjacency Model)\", box=box.ROUNDED)\nt.add_column(\"ID\", style=\"cyan\", width=8)\nt.add_column(\"Type\", style=\"green\", width=14)\nt.add_column(\"Children\", style=\"yellow\", width=20)\nt.add_column(\"Bindings\", style=\"magenta\", width=25)\nfor c in booking_surface.components:\n   binding = c.properties.get(\"dataBinding\", \"\")\n   t.add_row(c.id, c.type, \", \".join(c.children) if c.children else \"\u2014\", binding or \"\u2014\")\nconsole.print(t)<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement Google\u2019s A2UI specification: a flat adjacency-list model where components reference children by ID rather than nesting, making the format trivially streamable and easy for LLMs to generate incrementally. We build a client-side Widget Registry that maps abstract type strings like \u201ccard\u201d, \u201ctext-field\u201d, and \u201cprogress-bar\u201d to concrete terminal renderers, mirroring how a production app maps them to React components or Flutter widgets. We demonstrate the full cycle with a restaurant booking form, complete with data model bindings that decouple dynamic values from UI structure, exactly as the A2UI spec prescribes.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
hdr(3, \"Generative UI \u2014 LLM Produces Live Interfaces\",\n   \"The agent generates A2UI component trees dynamically based on the query.n\"\n   \"This is the core of 'Generative UI' \u2014 context-adaptive interfacesn\"\n   \"that go far beyond text-only chat responses.\")\n\n\nA2UI_GENERATION_PROMPT = \"\"\"\nYou are an A2UI Generative UI agent. Given a user query, you generate a rich \ninteractive interface \u2014 NOT text. You output an A2UI component tree as JSON.\n\n\nRULES:\n1. Output a flat list of components using the adjacency model (children = list of IDs).\n2. Available component types: card, text, button, text-field, row, column, divider, chip, image, progress-bar, select, date-picker, data-table\n3. Include a separate \"dataModel\" object for dynamic values. Use \"\/path\/to\/value\" bindings.\n4. The ROOT component should be a \"card\" with all others as descendants.\n5. Think about what UI BEST serves the user \u2014 forms for input, tables for data, \n  progress bars for status, chips for tags, buttons for actions.\n\n\nOUTPUT FORMAT (strict JSON, nothing else):\n{\n \"surfaceId\": \"unique-id\",\n \"components\": [\n   {\"id\": \"root\", \"type\": \"card\", \"title\": \"...\", \"children\": [\"c1\", \"c2\"]},\n   {\"id\": \"c1\", \"type\": \"text\", \"text\": \"...\", \"style\": \"headline\"},\n   ...\n ],\n \"dataModel\": { ... }\n}\n\"\"\"\n\n\n\n\ndef generate_ui(user_query: str) -> Optional[A2UISurface]:\n   console.print(f\"  [dim]Generating UI for:[\/] [bold]{user_query}[\/]\")\n   response = llm([\n       {\"role\": \"system\", \"content\": A2UI_GENERATION_PROMPT},\n       {\"role\": \"user\", \"content\": user_query},\n   ], max_tokens=1200)\n\n\n   if not response:\n       return None\n\n\n   raw = response.choices[0].message.content\n   try:\n       cleaned = re.sub(r'```jsons*|s*```', '', raw).strip()\n       spec = json.loads(cleaned)\n   except json.JSONDecodeError:\n       console.print(f\"[red]Failed to parse generated UI: {raw[:200]}[\/]\")\n       return None\n\n\n   components = []\n   for c in spec.get(\"components\", []):\n       components.append(A2UIComponent(\n           id=c.get(\"id\", str(uuid.uuid4())[:6]),\n           type=c.get(\"type\", \"text\"),\n           properties={k: v for k, v in c.items() if k not in (\"id\", \"type\", \"children\")},\n           children=c.get(\"children\", []),\n       ))\n\n\n   surface = A2UISurface(\n       surface_id=spec.get(\"surfaceId\", f\"gen-{uuid.uuid4().hex[:6]}\"),\n       components=components,\n       data_model=A2UIDataModel(spec.get(\"dataModel\", {})),\n   )\n\n\n   return surface\n\n\n\n\ndef demo_generative_ui(query: str):\n   surface = generate_ui(query)\n   if surface:\n       console.print(f\"n[bold green]Generated {len(surface.components)} components:[\/]\")\n       registry.render_tree(surface)\n\n\n       console.print()\n       types = {}\n       for c in surface.components:\n           types[c.type] = types.get(c.type, 0) + 1\n       console.print(\"  [dim]Component types used:[\/] \" + \", \".join(f\"[cyan]{t}[\/]\u00d7{n}\" for t, n in types.items()))\n\n\n       if surface.data_model.data:\n           console.print(f\"  [dim]Data model keys:[\/] {list(surface.data_model.data.keys())}\")\n       console.print()\n\n\n\n\nconsole.print(\"n[bold]Demo 1: Agent generates an onboarding form[\/]\")\ndemo_generative_ui(\n   \"Create a user onboarding flow: collect name, email, role (dropdown), \"\n   \"preferred notification method (chips), and a 'Get Started' button.\"\n)\n\n\nconsole.print(\"n[bold]Demo 2: Agent generates a data dashboard[\/]\")\ndemo_generative_ui(\n   \"Show a project status dashboard with: project name 'Atlas v2', \"\n   \"4 team members, sprint progress at 68%, 3 blockers flagged as critical, \"\n   \"and action buttons for 'View Backlog' and 'Schedule Standup'.\"\n)\n\n\nconsole.print(\"n[bold]Demo 3: Agent generates a confirmation dialog[\/]\")\ndemo_generative_ui(\n   \"Show a payment confirmation: $2,450 charge to Visa ending 4242, \"\n   \"order #ORD-8891, with Approve and Decline buttons.\"\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We hand the keys to the LLM and let it generate complete A2UI component trees at runtime from plain English descriptions, this is Generative UI in its purest form. We prompt the model with the A2UI schema and component catalog, and it produces fully structured surfaces with cards, forms, chips, progress bars, and data bindings, choosing the best UI pattern for each query. We run three demos, an onboarding flow, a project dashboard, and a payment confirmation, showing how the same agent adapts its interface to wildly different contexts without a single hardcoded layout.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
hdr(4, \"State Synchronization \u2014 Shared State Between Agent & UI\",\n   \"AG-UI syncs state bidirectionally using STATE_SNAPSHOT and STATE_DELTA.n\"\n   \"The agent IS the state machine; the UI IS the renderer.n\"\n   \"JSON Patch diffs keep updates minimal and efficient.\")\n\n\n\n\nclass SharedState:\n   def __init__(self, initial: dict = None):\n       self.state: dict = initial or {}\n       self.history: list[dict] = []\n       self.version: int = 0\n\n\n   def snapshot(self) -> AGUIEvent:\n       return AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {\"state\": copy.deepcopy(self.state), \"version\": self.version})\n\n\n   def apply_delta(self, operations: list[dict]) -> AGUIEvent:\n       for op in operations:\n           path_parts = [p for p in op[\"path\"].split(\"\/\") if p]\n           target = self.state\n           for part in path_parts[:-1]:\n               if isinstance(target, dict):\n                   target = target.setdefault(part, {})\n               elif isinstance(target, list) and part.isdigit():\n                   target = target[int(part)]\n\n\n           key = path_parts[-1] if path_parts else None\n           if key is None:\n               continue\n\n\n           if op[\"op\"] == \"replace\":\n               target[key] = op[\"value\"]\n           elif op[\"op\"] == \"add\":\n               if isinstance(target, list) and key.isdigit():\n                   target.insert(int(key), op[\"value\"])\n               else:\n                   target[key] = op[\"value\"]\n           elif op[\"op\"] == \"remove\":\n               if isinstance(target, dict):\n                   target.pop(key, None)\n\n\n       self.version += 1\n       self.history.append({\"version\": self.version, \"ops\": operations})\n       return AGUIEvent(AGUIEventType.STATE_DELTA, {\"delta\": operations, \"version\": self.version})\n\n\n\n\nconsole.print(\"n[bold]Demo: Document review pipeline \u2014 3 agents, shared state[\/]n\")\n\n\nstream = AGUIEventStream()\nstate = SharedState({\n   \"document\": {\"title\": \"Q4 Strategy Report\", \"status\": \"draft\", \"word_count\": 2840},\n   \"pipeline\": {\"stage\": \"research\", \"progress\": 0.0},\n   \"agents\": {\"active\": \"Researcher\", \"queue\": [\"Editor\", \"Reviewer\"]},\n   \"feedback\": [],\n})\n\n\ndef log_event(event: AGUIEvent):\n   if event.type in (AGUIEventType.STATE_SNAPSHOT, AGUIEventType.STATE_DELTA):\n       if event.type == AGUIEventType.STATE_DELTA:\n           ops = event.data.get(\"delta\", [])\n           for op in ops:\n               console.print(f\"  [yellow]STATE_DELTA[\/] v{event.data.get('version')}: \"\n                             f\"[cyan]{op['op']}[\/] {op['path']} \u2192 {op.get('value', '\u2205')}\")\n       else:\n           console.print(f\"  [yellow]STATE_SNAPSHOT[\/] v{event.data.get('version')}: {list(event.data['state'].keys())}\")\n\n\nstream.on(log_event)\n\n\nstream.emit(state.snapshot())\n\n\nconsole.print(\"n[bold green]\u25b8 Researcher agent working...[\/]\")\nstream.emit(state.apply_delta([\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"research_complete\"},\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/progress\", \"value\": 0.33},\n   {\"op\": \"add\",     \"path\": \"\/feedback\/0\", \"value\": {\"agent\": \"Researcher\", \"note\": \"Added 4 new data sources\"}},\n]))\n\n\nconsole.print(\"n[bold green]\u25b8 Editor agent working...[\/]\")\nstream.emit(state.apply_delta([\n   {\"op\": \"replace\", \"path\": \"\/agents\/active\", \"value\": \"Editor\"},\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"editing\"},\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/progress\", \"value\": 0.66},\n   {\"op\": \"replace\", \"path\": \"\/document\/word_count\", \"value\": 3150},\n]))\n\n\nconsole.print(\"n[bold green]\u25b8 Reviewer agent working...[\/]\")\nstream.emit(state.apply_delta([\n   {\"op\": \"replace\", \"path\": \"\/agents\/active\", \"value\": \"Reviewer\"},\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"review_complete\"},\n   {\"op\": \"replace\", \"path\": \"\/pipeline\/progress\", \"value\": 1.0},\n   {\"op\": \"replace\", \"path\": \"\/document\/status\", \"value\": \"approved\"},\n]))\n\n\nconsole.print(Panel(\n   json.dumps(state.state, indent=2),\n   title=\"[bold]Final shared state after pipeline\",\n   border_style=\"green\",\n))\n\n\nt = Table(title=\"State History (versions)\", box=box.ROUNDED)\nt.add_column(\"Version\", style=\"cyan\", justify=\"center\")\nt.add_column(\"Operations\", style=\"yellow\")\nfor h in state.history:\n   ops_summary = \"; \".join(f\"{o['op']} {o['path']}\" for o in h[\"ops\"])\n   t.add_row(str(h[\"version\"]), ops_summary[:70])\nconsole.print(t)\n\n\n\n\nhdr(5, \"Human-in-the-Loop \u2014 AG-UI INTERRUPT Events\",\n   \"When an agent hits a high-stakes action, it emits an INTERRUPT event.n\"\n   \"The frontend renders an approval UI. Execution pauses until the humann\"\n   \"approves, rejects, or modifies. State is preserved throughout.\")\n\n\n\n\n@dataclass\nclass InterruptRequest:\n   interrupt_id: str\n   action_description: str\n   risk_level: str\n   affected_resources: list[str]\n   proposed_changes: dict\n   options: list[str] = field(default_factory=lambda: [\"approve\", \"reject\", \"modify\"])\n\n\n\n\n@dataclass\nclass InterruptResponse:\n   interrupt_id: str\n   decision: str\n   modifications: Optional[dict] = None\n\n\n\n\nclass InterruptableAgent:\n   RISK_RULES = {\n       \"delete\":     \"critical\",\n       \"payment\":    \"critical\",\n       \"email_all\":  \"high\",\n       \"publish\":    \"high\",\n       \"update\":     \"medium\",\n       \"read\":       \"low\",\n   }\n\n\n   def __init__(self):\n       self.stream = AGUIEventStream()\n       self.pending_interrupts: dict[str, InterruptRequest] = {}\n\n\n   def assess_and_maybe_interrupt(self, action: str, details: dict) -> Optional[InterruptRequest]:\n       risk = \"low\"\n       for keyword, level in self.RISK_RULES.items():\n           if keyword in action.lower():\n               risk = level\n               break\n\n\n       if risk in (\"critical\", \"high\"):\n           interrupt = InterruptRequest(\n               interrupt_id=str(uuid.uuid4())[:8],\n               action_description=action,\n               risk_level=risk,\n               affected_resources=details.get(\"resources\", []),\n               proposed_changes=details.get(\"changes\", {}),\n           )\n           self.pending_interrupts[interrupt.interrupt_id] = interrupt\n\n\n           self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {\n               \"interrupt_id\": interrupt.interrupt_id,\n               \"reason\": risk,\n               \"description\": interrupt.action_description,\n               \"affected_resources\": interrupt.affected_resources,\n               \"proposed_changes\": interrupt.proposed_changes,\n               \"options\": interrupt.options,\n           }))\n           return interrupt\n       return None\n\n\n   def resolve_interrupt(self, response: InterruptResponse) -> str:\n       interrupt = self.pending_interrupts.pop(response.interrupt_id, None)\n       if not interrupt:\n           return \"No pending interrupt found.\"\n\n\n       if response.decision == \"approve\":\n           return f\"\"\u2705\" APPROVED: '{interrupt.action_description}' executing now.\"\n       elif response.decision == \"reject\":\n           return f\"\"\ud83d\udeab\" REJECTED: '{interrupt.action_description}' cancelled.\"\n       elif response.decision == \"modify\":\n           return f\"\"\u270f\"  MODIFIED: '{interrupt.action_description}' updated with: {response.modifications}\"\n       return f\"Unknown decision: {response.decision}\"\n\n\n\n\nconsole.print(\"n[bold]Demo: Agent encounters actions of varying risk levels[\/]n\")\nagent = InterruptableAgent()\n\n\nactions = [\n   (\"Read user profile\",      {\"resources\": [\"user:123\"]}),\n   (\"Update user preferences\", {\"resources\": [\"user:123\"], \"changes\": {\"theme\": \"dark\"}}),\n   (\"Delete user account\",     {\"resources\": [\"user:123\", \"data:all\"], \"changes\": {\"action\": \"permanent_delete\"}}),\n   (\"Email all 12,000 users\",  {\"resources\": [\"email:newsletter\"], \"changes\": {\"subject\": \"Big Announcement\"}}),\n   (\"Publish blog post\",       {\"resources\": [\"post:draft-42\"], \"changes\": {\"status\": \"public\"}}),\n]\n\n\ndef event_logger(event):\n   if event.type == AGUIEventType.INTERRUPT:\n       d = event.data\n       risk_style = {\"critical\": \"bold red\", \"high\": \"bold yellow\"}.get(d[\"reason\"], \"white\")\n       console.print(f\"n  [bold]\"\ud83d\udea8\" INTERRUPT EVENT[\/]\")\n       console.print(f\"    Risk: [{risk_style}]{d['reason'].upper()}[\/]\")\n       console.print(f\"    Action: {d['description']}\")\n       console.print(f\"    Affected: {d['affected_resources']}\")\n       console.print(f\"    Options: {d['options']}\")\n\n\nagent.stream.on(event_logger)\n\n\nresults = []\nfor action_desc, details in actions:\n   interrupt = agent.assess_and_maybe_interrupt(action_desc, details)\n   if interrupt:\n       decision = \"reject\" if interrupt.risk_level == \"critical\" else \"approve\"\n       result = agent.resolve_interrupt(InterruptResponse(interrupt.interrupt_id, decision))\n   else:\n       result = f\"\"\u26a1\" AUTO-EXECUTED: '{action_desc}' (low risk, no approval needed)\"\n\n\n   results.append((action_desc, result))\n\n\nconsole.print()\nt = Table(title=\"Execution Results\", box=box.ROUNDED, show_lines=True)\nt.add_column(\"Action\", style=\"white\", width=28)\nt.add_column(\"Outcome\", style=\"dim\", width=55)\nfor action_desc, result in results:\n   t.add_row(action_desc, result)\nconsole.print(t)<\/code><\/pre>\n<\/div>\n<\/div>\n
We build a SharedState engine that emits AG-UI STATE_SNAPSHOT and STATE_DELTA events using JSON Patch operations, keeping the agent backend and the frontend UI perfectly synchronized through every mutation. We demonstrate this with a three-agent document review pipeline in which a Researcher, Editor, and Reviewer each modify the shared state in sequence, and the frontend sees every change the instant it occurs. We then implement the AG-UI INTERRUPT pattern, in which the agent assesses risk levels for proposed actions, emits interrupt events for any dangerous actions, and pauses execution until a human approves, rejects, or modifies the plan.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
hdr(6, \"Full Pipeline \u2014 LLM-Driven Adaptive UI\",\n   \"The complete Agentic UI architecture in one pipeline:n\"\n   \"  User query \u2192 Intent analysis \u2192 UI pattern selection \u2192n\"\n   \"  A2UI generation \u2192 AG-UI event streaming \u2192 State sync \u2192 Render\")\n\n\n\n\nUI_ROUTER_PROMPT = \"\"\"\nYou are a UI routing agent. Given a user query, decide what type of UI to generate.\n\n\nRESPOND IN JSON ONLY:\n{\n \"intent\": \"form | dashboard | confirmation | list | detail | wizard | error\",\n \"reasoning\": \"why this UI pattern fits\",\n \"ui_complexity\": \"simple | moderate | complex\",\n \"needs_approval\": true\/false,\n \"data_requirements\": [\"what data the UI needs\"]\n}\n\"\"\"\n\n\n\n\nclass AgenticUIPipeline:\n   def __init__(self):\n       self.stream = AGUIEventStream()\n       self.state = SharedState({\"pipeline\": {\"stage\": \"idle\"}, \"renders\": 0})\n       self.interrupt_agent = InterruptableAgent()\n\n\n   def route(self, query: str) -> dict:\n       resp = llm([\n           {\"role\": \"system\", \"content\": UI_ROUTER_PROMPT},\n           {\"role\": \"user\", \"content\": query},\n       ], max_tokens=300)\n\n\n       if not resp:\n           return {\"intent\": \"dashboard\", \"reasoning\": \"fallback\", \"ui_complexity\": \"simple\",\n                   \"needs_approval\": False, \"data_requirements\": []}\n       raw = resp.choices[0].message.content\n       try:\n           return json.loads(re.sub(r'```jsons*|s*```', '', raw).strip())\n       except json.JSONDecodeError:\n           return {\"intent\": \"dashboard\", \"reasoning\": \"parse_fallback\", \"ui_complexity\": \"simple\",\n                   \"needs_approval\": False, \"data_requirements\": []}\n\n\n   def run(self, user_query: str):\n       run_id = str(uuid.uuid4())[:8]\n       console.print(Panel(f\"[bold]{user_query}[\/]\", title=\"\"\ud83e\uddd1\" User Query\", border_style=\"white\"))\n\n\n       self.stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {\"run_id\": run_id}))\n       self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {\"step\": \"routing\"}))\n\n\n       routing = self.route(user_query)\n       console.print(f\"n  [bold cyan]\"\ud83d\udce1\" Router Decision:[\/]\")\n       console.print(f\"    Intent: [green]{routing.get('intent')}[\/] | \"\n                     f\"Complexity: [yellow]{routing.get('ui_complexity')}[\/] | \"\n                     f\"Approval: {'\"\ud83d\udd12\"' if routing.get('needs_approval') else '\"\u2705\"'}\")\n       console.print(f\"    Reasoning: [dim]{routing.get('reasoning', '')}[\/]\")\n\n\n       self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {\"step\": \"routing\", \"result\": routing}))\n\n\n       self.state.apply_delta([\n           {\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"generating\"},\n       ])\n\n\n       self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {\"step\": \"generating_ui\"}))\n\n\n       msg_id = str(uuid.uuid4())[:8]\n       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {\"message_id\": msg_id}))\n       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {\n           \"message_id\": msg_id,\n           \"delta\": f\"Building a {routing.get('intent')} interface for you...\"\n       }))\n       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {\"message_id\": msg_id}))\n\n\n       surface = generate_ui(user_query)\n       self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {\"step\": \"generating_ui\"}))\n\n\n       if routing.get(\"needs_approval\") and surface:\n           self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {\n               \"reason\": \"ui_confirmation\",\n               \"description\": f\"Generated {len(surface.components)} component UI. Render it?\",\n               \"options\": [\"render\", \"regenerate\", \"cancel\"],\n           }))\n           console.print(\"n  [bold yellow]\"\u23f8\"  INTERRUPT:[\/] UI generated, awaiting human approval...\")\n           console.print(\"  [green]\u2192 Auto-approving for demo...[\/]\")\n\n\n       if surface:\n           self.state.apply_delta([\n               {\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"rendering\"},\n               {\"op\": \"replace\", \"path\": \"\/renders\", \"value\": self.state.state.get(\"renders\", 0) + 1},\n           ])\n\n\n           console.print(f\"n[bold green]\"\ud83d\udda5\"  Rendered Interface ({len(surface.components)} components):[\/]n\")\n           registry.render_tree(surface)\n\n\n           self.stream.emit(AGUIEvent(AGUIEventType.CUSTOM, {\n               \"subtype\": \"a2ui_surface\",\n               \"surface\": surface.to_messages(),\n           }))\n\n\n       self.state.apply_delta([{\"op\": \"replace\", \"path\": \"\/pipeline\/stage\", \"value\": \"complete\"}])\n       self.stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {\"run_id\": run_id, \"status\": \"success\"}))\n\n\n       console.print()\n       event_counts = {}\n       for e in self.stream.events:\n           event_counts[e.type.value] = event_counts.get(e.type.value, 0) + 1\n       t = Table(title=\"Pipeline Event Summary\", box=box.ROUNDED)\n       t.add_column(\"Event Type\", style=\"cyan\")\n       t.add_column(\"Count\", justify=\"center\", style=\"green\")\n       for etype, count in sorted(event_counts.items()):\n           t.add_row(etype, str(count))\n       console.print(t)\n\n\n\n\npipeline = AgenticUIPipeline()\n\n\nconsole.print(\"n[bold]Demo 1: Agent builds a settings form[\/]\")\npipeline.run(\n   \"Create a notification settings panel where I can toggle email\/SMS\/push, \"\n   \"set quiet hours, and pick a notification sound.\"\n)\n\n\npipeline.stream = AGUIEventStream()\npipeline.state = SharedState({\"pipeline\": {\"stage\": \"idle\"}, \"renders\": 0})\n\n\nconsole.print(\"n[bold]Demo 2: Agent builds an order tracking dashboard[\/]\")\npipeline.run(\n   \"Show order #ORD-7742 status: shipped via FedEx, tracking 789456123, \"\n   \"estimated delivery March 24, 2 of 3 items delivered. Show a progress bar \"\n   \"and action buttons for 'Contact Support' and 'Request Refund'.\"\n)\n\n\n\n\nhdr(7, \"Incremental UI Updates \u2014 Live Surface Modification\",\n   \"A2UI surfaces are incrementally updateable. The agent can add, remove,n\"\n   \"or modify components and data bindings on a live surface withoutn\"\n   \"regenerating the whole tree. Essential for real-time collaboration.\")\n\n\n\n\nclass LiveSurface:\n   def __init__(self, surface: A2UISurface):\n       self.surface = surface\n       self.update_log: list[str] = []\n\n\n   def add_component(self, component: A2UIComponent, parent_id: Optional[str] = None):\n       self.surface.components.append(component)\n       if parent_id:\n           for c in self.surface.components:\n               if c.id == parent_id:\n                   c.children.append(component.id)\n                   break\n       self.update_log.append(f\"ADD  {component.type}#{component.id} \u2192 parent:{parent_id or 'root'}\")\n\n\n   def update_component(self, component_id: str, new_props: dict):\n       for c in self.surface.components:\n           if c.id == component_id:\n               c.properties.update(new_props)\n               self.update_log.append(f\"UPD  #{component_id} props: {list(new_props.keys())}\")\n               return\n       self.update_log.append(f\"ERR  #{component_id} not found\")\n\n\n   def remove_component(self, component_id: str):\n       self.surface.components = [c for c in self.surface.components if c.id != component_id]\n       for c in self.surface.components:\n           if component_id in c.children:\n               c.children.remove(component_id)\n       self.update_log.append(f\"DEL  #{component_id}\")\n\n\n   def update_data(self, path: str, value: Any):\n       self.surface.data_model.data = _set_nested(self.surface.data_model.data, path, value)\n       self.update_log.append(f\"DATA {path} = {value}\")\n\n\n\n\ndef _set_nested(d: dict, path: str, value: Any) -> dict:\n   parts = [p for p in path.split(\"\/\") if p]\n   d = copy.deepcopy(d)\n   current = d\n   for p in parts[:-1]:\n       current = current.setdefault(p, {})\n   if parts:\n       current[parts[-1]] = value\n   return d\n\n\n\n\nconsole.print(\"n[bold]Demo: Live collaborative editing \u2014 agent modifies UI in real-time[\/]n\")\n\n\ninitial = A2UISurface(\n   surface_id=\"task-board\",\n   components=[\n       A2UIComponent(\"board\", \"card\", {\"title\": \"\"\ud83d\uddc2\"  Sprint Board\"}, children=[\"t1\", \"t2\", \"t3\"]),\n       A2UIComponent(\"t1\", \"chip\", {\"label\": \"AUTH-101: Login flow\", \"variant\": \"in_progress\"}),\n       A2UIComponent(\"t2\", \"chip\", {\"label\": \"AUTH-102: OAuth setup\", \"variant\": \"todo\"}),\n       A2UIComponent(\"t3\", \"chip\", {\"label\": \"AUTH-103: 2FA\", \"variant\": \"todo\"}),\n   ],\n   data_model=A2UIDataModel({\"sprint\": {\"name\": \"Sprint 14\", \"velocity\": 21}}),\n)\n\n\nlive = LiveSurface(initial)\n\n\nconsole.print(\"[bold]Initial board:[\/]\")\nregistry.render_tree(live.surface)\n\n\nconsole.print(\"n[bold yellow]Agent updating board in real-time...[\/]n\")\n\n\nlive.update_component(\"t1\", {\"variant\": \"done\", \"label\": \"\"\u2705\" AUTH-101: Login flow\"})\nlive.update_component(\"t2\", {\"variant\": \"in_progress\", \"label\": \"\"\ud83d\udd04\" AUTH-102: OAuth setup\"})\nlive.add_component(\n   A2UIComponent(\"t4\", \"chip\", {\"label\": \"AUTH-104: Password reset\", \"variant\": \"todo\"}),\n   parent_id=\"board\"\n)\nlive.update_data(\"\/sprint\/velocity\", 25)\nlive.remove_component(\"t3\")\n\n\nconsole.print(\"[bold]Updated board:[\/]\")\nregistry.render_tree(live.surface)\n\n\nconsole.print()\nt = Table(title=\"Incremental Update Log\", box=box.ROUNDED)\nt.add_column(\"#\", style=\"cyan\", width=4, justify=\"center\")\nt.add_column(\"Operation\", style=\"yellow\")\nfor i, entry in enumerate(live.update_log, 1):\n   t.add_row(str(i), entry)\nconsole.print(t)<\/code><\/pre>\n<\/div>\n<\/div>\n
We wire every piece together into a single AgenticUIPipeline class that takes a user query, classifies its intent with an LLM router, selects the right UI pattern, generates an A2UI surface, streams the entire process over AG-UI events, manages shared state, and renders the result, the complete architecture in one run. We then build a LiveSurface class that supports incremental A2UI updates: adding, modifying, and removing components on an already-rendered surface without regenerating the whole tree, which is essential for real-time collaborative experiences. We demo this with a sprint board that an agent updates live, marking tasks complete, adding new ones, and adjusting data model values, all tracked in a detailed operation log.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
hdr(8, \"Reference \u2014 The Agentic UI Protocol Stack\",\n   \"How AG-UI, A2UI, MCP, and A2A fit together in the modern agent architecture.\")\n\n\nconsole.print(Panel(\"\"\"\n[bold white]THE AGENTIC UI STACK (2026)[\/]\n\n\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502  [bold cyan]USER INTERFACE[\/]  (React, Flutter, SwiftUI, Terminal)  \u2502\n \u2502  Renders native widgets from component specs          \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n                          \u2502  A2UI component trees (JSON)\n                          \u2502  AG-UI events (SSE \/ WebSocket)\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502  [bold yellow]AG-UI PROTOCOL[\/]  (Agent \"\u2194\" User Interaction)          \u2502\n \u2502  \u2022 Event streaming (TEXT, TOOL_CALL, STATE, INTERRUPT)\u2502\n \u2502  \u2022 Bidirectional state sync (SNAPSHOT + DELTA)        \u2502\n \u2502  \u2022 Human-in-the-loop (INTERRUPT \u2192 approval flow)      \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n                          \u2502\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502  [bold magenta]AGENT RUNTIME[\/]  (LangGraph, CrewAI, custom, etc.)    \u2502\n \u2502  \u2022 Generates A2UI surfaces (Generative UI)            \u2502\n \u2502  \u2022 Manages shared state                               \u2502\n \u2502  \u2022 Orchestrates sub-agents via A2A protocol           \u2502\n \u2502  \u2022 Accesses tools via MCP protocol                    \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n                          \u2502\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502  [bold green]LLM BACKBONE[\/]  (GPT, Claude, Gemini, etc.)           \u2502\n \u2502  \u2022 Generates component trees as structured output     \u2502\n \u2502  \u2022 Reasons about UI patterns per context              \u2502\n \u2502  \u2022 Streams tokens for real-time rendering              \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n\n\n[dim]Protocol roles:\n AG-UI  = Agent \"\u2194\" User (streaming events, state, HITL)\n A2UI   = Agent \u2192 UI (declarative component specs)\n A2A    = Agent \u2192 Agent (delegation, sub-agents)\n MCP    = Agent \u2192 Tools (function calling, context)[\/]\n\"\"\", title=\"[bold]Architecture Reference\", border_style=\"cyan\"))\n\n\n\n\nref = Table(title=\"Agentic UI Concepts \u2014 Quick Reference\", box=box.DOUBLE_EDGE, show_lines=True)\nref.add_column(\"Concept\", style=\"bold cyan\", width=22)\nref.add_column(\"What It Does\", style=\"white\", width=35)\nref.add_column(\"Key Mechanism\", style=\"yellow\", width=28)\n\n\nref.add_row(\"AG-UI Events\",     \"Stream agent actions to frontend in real-time\",  \"SSE\/WebSocket + ~16 event types\")\nref.add_row(\"A2UI Components\",  \"Declarative UI trees \u2014 safe, portable, native\",  \"Flat JSON + widget registry\")\nref.add_row(\"State Sync\",       \"Keep agent & UI state in lockstep\",              \"STATE_SNAPSHOT + STATE_DELTA\")\nref.add_row(\"Generative UI\",    \"LLM generates UI at runtime, not just text\",     \"A2UI JSON as structured output\")\nref.add_row(\"INTERRUPT (HITL)\", \"Pause execution for human approval\",             \"INTERRUPT event \u2192 approval flow\")\nref.add_row(\"Incremental Update\",\"Modify live surfaces without full regeneration\", \"A2UI updateComponents message\")\nref.add_row(\"Data Binding\",     \"UI reads from a shared data model\",              \"JSON Pointer paths (\/path\/to\/val)\")\nref.add_row(\"Widget Registry\",  \"Client maps abstract types to native widgets\",   \"Catalog of trusted components\")\n\n\nconsole.print(ref)\n\n\n\n\nconsole.print(Panel(\n   \"[bold green]Tutorial complete![\/]nn\"\n   \"[dim]What you built:[\/]n\"\n   \"  \u2022 A full AG-UI event system with all 16 event typesn\"\n   \"  \u2022 An A2UI renderer with flat adjacency-list components + data bindingn\"\n   \"  \u2022 LLM-powered Generative UI that creates interfaces from natural languagen\"\n   \"  \u2022 Bidirectional state sync with JSON Patch deltasn\"\n   \"  \u2022 Human-in-the-loop interrupt and approval flowsn\"\n   \"  \u2022 Incremental live surface updatesnn\"\n   \"[dim]To go further:[\/]n\"\n   \"  \u2022 Serve AG-UI events over real SSE with FastAPIn\"\n   \"  \u2022 Connect to CopilotKit React components for a real frontendn\"\n   \"  \u2022 Use Pydantic AI's AGUIAdapter for production agent hostingn\"\n   \"  \u2022 Add A2A protocol for multi-agent delegationn\"\n   \"  \u2022 Deploy on AWS Bedrock AgentCore with native AG-UI support\",\n   title=\"[bold]\"\ud83c\udf93\" What's Next?\",\n   border_style=\"green\",\n   padding=(1, 2),\n))\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We close with a visual protocol stack diagram showing exactly how AG-UI, A2UI, A2A, and MCP fit together in the modern agentic architecture, from the LLM backbone at the bottom to the native UI at the top. We provide a quick-reference table mapping every concept we built, event streaming, component trees, state sync, generative UI, interrupts, incremental updates, data binding, and widget registries, to their core mechanisms. We point the way forward to production: serving AG-UI events over real SSE with FastAPI, connecting to CopilotKit React components, using Pydantic AI\u2019s AGUIAdapter, and deploying on AWS Bedrock AgentCore.<\/p>\n
In conclusion, we have a fully functional Agentic UI pipeline that takes a simple natural-language query and transforms it into a structured, interactive interface powered by an intelligent agent. We do not just assemble components; we understand how each layer operates and connects, from real-time AG-UI event streaming and declarative A2UI interface definitions to state synchronization through JSON Patch and enforced human-in-the-loop safety mechanisms. This clarity allows us to reason about system behavior, debug effectively, and extend functionality without relying on black-box abstractions. Also, we leave with the ability to design our own agent-driven UI systems, adapt them to different use cases, and confidently build production-ready experiences where agents and interfaces evolve together in a controlled, transparent, and scalable manner.<\/p>\n
\n
Check out\u00a0the\u00a0Full Codes with Notebook here<\/a><\/strong>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0130k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post A Coding Deep Dive into Agentic UI, Generative UI, State Synchronization, and Interrupt-Driven Approval Flows<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build the…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-825","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/825","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=825"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/825\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=825"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=825"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=825"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}