On this tutorial, we construct the whole Agentic UI stack from the bottom up utilizing plain Python, with out counting on exterior frameworks to summary away the core concepts. We implement the AG-UI occasion stream to make agent conduct observable in actual time, and we herald A2UI as a declarative layer that permits interfaces to be outlined as structured JSON slightly than executable code. As we progress, we allow an LLM to generate full consumer interfaces from pure language, synchronize agent and UI state by way of JSON Patch updates, and implement human-in-the-loop security for vital actions. Additionally, we acquire a transparent, end-to-end understanding of how agent reasoning transforms into interactive, protocol-compliant consumer interfaces.
import subprocess, sys
for pkg in ["openai", "rich", "pydantic"]:
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", pkg])
import os, getpass
if os.environ.get("OPENAI_API_KEY"):
API_KEY = os.environ["OPENAI_API_KEY"]
print(" Utilizing OPENAI_API_KEY from setting.")
else:
attempt:
from google.colab import userdata
API_KEY = userdata.get("OPENAI_API_KEY")
print(" Utilizing OPENAI_API_KEY from Colab Secrets and techniques.")
besides Exception:
API_KEY = getpass.getpass(" Enter your OpenAI API key (hidden): ")
print(" API key obtained.")
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
import json, re, time, uuid, copy, textwrap
from enum import Enum
from dataclasses import dataclass, discipline, asdict
from typing import Any, Non-obligatory, Generator
from pydantic import BaseModel, Discipline
from openai import OpenAI
from wealthy.console import Console
from wealthy.panel import Panel
from wealthy.desk import Desk
from wealthy.tree import Tree
from wealthy.textual content import Textual content
from wealthy.markdown import Markdown
from wealthy import field
console = Console(width=105)
shopper = OpenAI(api_key=API_KEY, base_url=BASE_URL)
def llm(messages, **kw):
attempt:
return shopper.chat.completions.create(mannequin=MODEL, messages=messages, temperature=0.2, **kw)
besides Exception as e:
console.print(f"[red]LLM error: {e}[/]")
return None
def hdr(n, title, sub=""):
console.print()
console.rule(f"[bold cyan]SECTION {n}", fashion="cyan")
physique = f"[bold white]{title}[/]n[dim]{sub}[/]" if sub else f"[bold white]{title}[/]"
console.print(Panel(physique, border_style="cyan", padding=(1, 2)))
hdr(1, "AG-UI Protocol — Occasion System",
"The true AG-UI protocol makes use of ~16 occasion varieties streamed through SSE.n"
"We implement all core occasion varieties and a streaming emitter in pure Python.")
class AGUIEventType(str, Enum):
RUN_STARTED = "RUN_STARTED"
RUN_FINISHED = "RUN_FINISHED"
RUN_ERROR = "RUN_ERROR"
TEXT_MESSAGE_START = "TEXT_MESSAGE_START"
TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT"
TEXT_MESSAGE_END = "TEXT_MESSAGE_END"
TOOL_CALL_START = "TOOL_CALL_START"
TOOL_CALL_ARGS = "TOOL_CALL_ARGS"
TOOL_CALL_RESULT = "TOOL_CALL_RESULT"
TOOL_CALL_END = "TOOL_CALL_END"
STATE_SNAPSHOT = "STATE_SNAPSHOT"
STATE_DELTA = "STATE_DELTA"
INTERRUPT = "INTERRUPT"
CUSTOM = "CUSTOM"
STEP_STARTED = "STEP_STARTED"
STEP_FINISHED = "STEP_FINISHED"
@dataclass
class AGUIEvent:
sort: AGUIEventType
knowledge: dict = discipline(default_factory=dict)
event_id: str = discipline(default_factory=lambda: str(uuid.uuid4())[:8])
timestamp: float = discipline(default_factory=time.time)
def to_sse(self) -> str:
payload = {"sort": self.sort.worth, "id": self.event_id, **self.knowledge}
return f"occasion: ag-uindata: {json.dumps(payload)}nn"
def to_json(self) -> dict:
return {"sort": self.sort.worth, "id": self.event_id, "ts": self.timestamp, **self.knowledge}
class AGUIEventStream:
def __init__(self):
self.occasions: checklist[AGUIEvent] = []
self.listeners: checklist = []
def emit(self, occasion: AGUIEvent):
self.occasions.append(occasion)
for listener in self.listeners:
listener(occasion)
def on(self, callback):
self.listeners.append(callback)
def replay(self) -> checklist[dict]:
return [e.to_json() for e in self.events]
def demo_agui_lifecycle():
stream = AGUIEventStream()
event_colors = {
"RUN_": "daring inexperienced", "TEXT_": "cyan", "TOOL_": "magenta",
"STATE_": "yellow", "INTERRUPT": "daring crimson", "STEP_": "dim",
}
def frontend_listener(occasion: AGUIEvent):
coloration = "white"
for prefix, c in event_colors.objects():
if occasion.sort.worth.startswith(prefix):
coloration = c
break
element = json.dumps(occasion.knowledge)[:80] if occasion.knowledge else ""
console.print(f" [{color}] {occasion.sort.worth:.<28}[/] {element}")
stream.on(frontend_listener)
run_id = str(uuid.uuid4())[:8]
console.print("[bold]Simulating full AG-UI agent run...[/]n")
stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {"run_id": run_id}))
stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "analyzing_query", "label": "Understanding request"}))
stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "analyzing_query"}))
msg_id = str(uuid.uuid4())[:8]
stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {"message_id": msg_id, "function": "assistant"}))
for chunk in ["I'll ", "look up ", "the data ", "and build ", "a dashboard ", "for you."]:
stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {"message_id": msg_id, "delta": chunk}))
stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {"message_id": msg_id}))
tool_id = str(uuid.uuid4())[:8]
stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_START, {"tool_call_id": tool_id, "title": "query_database"}))
stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_ARGS, {"tool_call_id": tool_id, "args_delta": '{"question": "SELECT income FROM gross sales"}'}))
stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_RESULT, {"tool_call_id": tool_id, "outcome": [{"month": "Jan", "revenue": 42000}, {"month": "Feb", "revenue": 58000}]}))
stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_END, {"tool_call_id": tool_id}))
stream.emit(AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {
"state": {"active_agent": "DataAnalyst", "stage": "rendering", "progress": 0.75}
}))
stream.emit(AGUIEvent(AGUIEventType.STATE_DELTA, {
"delta": [{"op": "replace", "path": "/progress", "value": 1.0}]
}))
stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
"motive": "high_risk_action",
"description": "Agent needs to ship an electronic mail to all 5,000 prospects.",
"choices": ["approve", "reject", "modify"],
}))
stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {"run_id": run_id, "standing": "accomplished"}))
console.print(Panel(
stream.occasions[3].to_sse(),
title="[bold]Instance SSE wire format (the way it appears to be like on the community)",
border_style="dim",
))
desk = Desk(title="AG-UI Occasion Stream Abstract", field=field.ROUNDED)
desk.add_column("Class", fashion="cyan", width=15)
desk.add_column("Occasions", justify="middle", fashion="inexperienced")
counts = {}
for e in stream.occasions:
cat = e.sort.worth.rsplit("_", 1)[0] if "_" in e.sort.worth else e.sort.worth
counts[cat] = counts.get(cat, 0) + 1
for cat, n in counts.objects():
desk.add_row(cat, str(n))
desk.add_row("[bold]TOTAL", f"[bold]{len(stream.occasions)}")
console.print(desk)
demo_agui_lifecycle()We begin by constructing the spine of each agentic frontend: the AG-UI occasion stream. We implement all 16 occasion varieties from the true AG-UI specification, lifecycle occasions, token-by-token textual content streaming, streamed instrument calls, state snapshots, deltas, and interrupt alerts, and serialize them into the SSE wire format that manufacturing methods use over HTTP. We then wire up a frontend listener that reacts to every occasion because it arrives, simulating the precise expertise a React or Flutter app would have consuming this stream.
hdr(2, "A2UI — Declarative Part Bushes",
"Google's A2UI spec: brokers emit flat JSON element lists with ID refs.n"
"The shopper's widget registry maps varieties → native widgets.n"
"Secure like knowledge, expressive like code. No executable code despatched.")
class A2UIMessageType(str, Enum):
CREATE_SURFACE = "createSurface"
UPDATE_COMPONENTS = "updateComponents"
UPDATE_DATA_MODEL = "updateDataModel"
DELETE_SURFACE = "deleteSurface"
@dataclass
class A2UIComponent:
id: str
sort: str
properties: dict = discipline(default_factory=dict)
kids: checklist[str] = discipline(default_factory=checklist)
def to_dict(self) -> dict:
d = {"id": self.id, "sort": self.sort, **self.properties}
if self.kids:
d["children"] = self.kids
return d
@dataclass
class A2UIDataModel:
knowledge: dict = discipline(default_factory=dict)
def get_binding(self, path: str) -> Any:
components = [p for p in path.split("/") if p]
val = self.knowledge
for p in components:
if isinstance(val, dict):
val = val.get(p)
elif isinstance(val, checklist) and p.isdigit():
val = val[int(p)]
else:
return None
return val
@dataclass
class A2UISurface:
surface_id: str
parts: checklist[A2UIComponent] = discipline(default_factory=checklist)
data_model: A2UIDataModel = discipline(default_factory=A2UIDataModel)
def to_messages(self) -> checklist[dict]:
msgs = []
msgs.append({
"sort": A2UIMessageType.CREATE_SURFACE.worth,
"surfaceId": self.surface_id,
})
msgs.append({
"sort": A2UIMessageType.UPDATE_COMPONENTS.worth,
"surfaceId": self.surface_id,
"parts": [c.to_dict() for c in self.components],
})
if self.data_model.knowledge:
msgs.append({
"sort": A2UIMessageType.UPDATE_DATA_MODEL.worth,
"surfaceId": self.surface_id,
"dataModel": self.data_model.knowledge,
})
return msgs
class WidgetRegistry:
def __init__(self):
self._renderers = {}
def register(self, component_type: str, render_fn):
self._renderers[component_type] = render_fn
def render(self, element: A2UIComponent, floor: A2UISurface, indent: int = 0):
fn = self._renderers.get(element.sort)
if fn:
fn(element, floor, indent)
else:
pad = " " * indent
console.print(f"{pad}[dim]⟨{element.sort} id={element.id}⟩ (no renderer)[/]")
def render_tree(self, floor: A2UISurface):
comp_map = {c.id: c for c in floor.parts}
all_children = set()
for c in floor.parts:
all_children.replace(c.kids)
roots = [c for c in surface.components if c.id not in all_children]
def _render(comp_id: str, indent: int):
comp = comp_map.get(comp_id)
if not comp:
return
self.render(comp, floor, indent)
for child_id in comp.kids:
_render(child_id, indent + 1)
for root in roots:
_render(root.id, 0)
registry = WidgetRegistry()
def _resolve(comp, floor, key, default=None):
val = comp.properties.get(key, default)
binding = comp.properties.get("dataBinding")
if binding and isinstance(binding, str) and binding.startswith("/"):
resolved = floor.data_model.get_binding(binding)
if resolved will not be None:
return resolved
if isinstance(val, str) and val.startswith("/") and "/" in val[1:]:
resolved = floor.data_model.get_binding(val)
if resolved will not be None:
return resolved
return val
def _to_float(val, default=0.0):
if isinstance(val, (int, float)):
return float(val)
if isinstance(val, str):
cleaned = val.strip().rstrip("%")
attempt:
f = float(cleaned)
if "%" in val or f > 1:
return f / 100.0
return f
besides ValueError:
return default
return default
def render_card(comp, floor, indent):
pad = " " * indent
title = str(_resolve(comp, floor, "title", "Card"))
console.print(f"{pad}┌─{'─' * 50}─┐")
console.print(f"{pad}│ [bold]{title:^50}[/] │")
console.print(f"{pad}├─{'─' * 50}─┤")
if not comp.kids:
subtitle = str(_resolve(comp, floor, "subtitle", ""))
if subtitle:
console.print(f"{pad}│ {subtitle:<49}│")
console.print(f"{pad}└─{'─' * 50}─┘")
def render_text(comp, floor, indent):
pad = " " * indent
textual content = _resolve(comp, floor, "textual content", "")
fashion = comp.properties.get("fashion", "physique")
types = {"headline": "daring white", "physique": "white", "caption": "dim", "label": "daring cyan"}
console.print(f"{pad}[{styles.get(style, 'white')}]{textual content}[/]")
def render_button(comp, floor, indent):
pad = " " * indent
label = str(_resolve(comp, floor, "label", "Button"))
variant = comp.properties.get("variant", "main")
colours = {"main": "daring white on blue", "secondary": "white on grey30", "hazard": "daring white on crimson"}
console.print(f"{pad} [{colors.get(variant, 'white')}] {label} [/]")
def render_text_field(comp, floor, indent):
pad = " " * indent
label = comp.properties.get("label", "Enter")
placeholder = comp.properties.get("placeholder", "")
console.print(f"{pad} {label}: [dim]┌──────────────────────────┐[/]")
console.print(f"{pad} [dim]│ {placeholder:<25}│[/]")
console.print(f"{pad} [dim]└──────────────────────────┘[/]")
def render_row(comp, floor, indent):
cross
def render_column(comp, floor, indent):
cross
def render_image(comp, floor, indent):
pad = " " * indent
alt = comp.properties.get("alt", "picture")
console.print(f"{pad} [dim] [{alt}][/]")
def render_divider(comp, floor, indent):
pad = " " * indent
console.print(f"{pad} {'─' * 50}")
def render_chip(comp, floor, indent):
pad = " " * indent
label = str(_resolve(comp, floor, "label", ""))
console.print(f"{pad} [on grey23] {label} [/]")
def render_progress(comp, floor, indent):
pad = " " * indent
raw_value = _resolve(comp, floor, "worth", 0)
worth = max(0.0, min(1.0, _to_float(raw_value, 0.0)))
label = str(_resolve(comp, floor, "label", ""))
bar_len = int(worth * 40)
bar = f"[green]{'█' * bar_len}[/][dim]{'░' * (40 - bar_len)}[/]"
console.print(f"{pad} {label}: {bar} {worth*100:.0f}%")
for title, fn in [
("card", render_card), ("text", render_text), ("button", render_button),
("text-field", render_text_field), ("row", render_row), ("column", render_column),
("image", render_image), ("divider", render_divider), ("chip", render_chip),
("progress-bar", render_progress),
]:
registry.register(title, fn)
console.print("n[bold]Demo: A2UI reserving kind — agent generates a restaurant reservation UI[/]n")
booking_surface = A2UISurface(
surface_id="booking-form-1",
parts=[
A2UIComponent("root", "card", {"title": " Reserve a Table"}, children=["c1", "c2", "c3", "c4", "c5", "c6"]),
A2UIComponent("c1", "textual content", {"textual content": "", "dataBinding": "/restaurant/title", "fashion": "headline"}),
A2UIComponent("c2", "textual content", {"textual content": "", "dataBinding": "/restaurant/delicacies", "fashion": "caption"}),
A2UIComponent("c3", "divider", {}),
A2UIComponent("c4", "text-field", {"label": "Date", "placeholder": "YYYY-MM-DD"}),
A2UIComponent("c5", "text-field", {"label": "Friends", "placeholder": "1-12"}),
A2UIComponent("c6", "button", {"label": "Reserve Now", "variant": "main", "motion": "submit_booking"}),
],
data_model=A2UIDataModel({"restaurant": {"title": "Chez Laurent", "delicacies": "French Up to date • $$$$"}})
)
console.print(Panel(
"n".be part of(json.dumps(m, indent=2)[:200] for m in booking_surface.to_messages()),
title="[bold]A2UI JSONL stream (what goes over the wire)",
border_style="yellow",
))
console.print("[bold]Rendered by shopper widget registry:[/]n")
registry.render_tree(booking_surface)
console.print()
t = Desk(title="A2UI Flat Part Checklist (Adjacency Mannequin)", field=field.ROUNDED)
t.add_column("ID", fashion="cyan", width=8)
t.add_column("Sort", fashion="inexperienced", width=14)
t.add_column("Youngsters", fashion="yellow", width=20)
t.add_column("Bindings", fashion="magenta", width=25)
for c in booking_surface.parts:
binding = c.properties.get("dataBinding", "")
t.add_row(c.id, c.sort, ", ".be part of(c.kids) if c.kids else "—", binding or "—")
console.print(t)We implement Google’s A2UI specification: a flat adjacency-list mannequin the place parts reference kids by ID slightly than nesting, making the format trivially streamable and simple for LLMs to generate incrementally. We construct a client-side Widget Registry that maps summary sort strings like “card”, “text-field”, and “progress-bar” to concrete terminal renderers, mirroring how a manufacturing app maps them to React parts or Flutter widgets. We show the complete cycle with a restaurant reserving kind, full with knowledge mannequin bindings that decouple dynamic values from UI construction, precisely because the A2UI spec prescribes.
hdr(3, "Generative UI — LLM Produces Reside Interfaces",
"The agent generates A2UI element timber dynamically based mostly on the question.n"
"That is the core of 'Generative UI' — context-adaptive interfacesn"
"that go far past text-only chat responses.")
A2UI_GENERATION_PROMPT = """
You might be an A2UI Generative UI agent. Given a consumer question, you generate a wealthy
interactive interface — NOT textual content. You output an A2UI element tree as JSON.
RULES:
1. Output a flat checklist of parts utilizing the adjacency mannequin (kids = checklist of IDs).
2. Out there element varieties: card, textual content, button, text-field, row, column, divider, chip, picture, progress-bar, choose, date-picker, data-table
3. Embody a separate "dataModel" object for dynamic values. Use "/path/to/worth" bindings.
4. The ROOT element ought to be a "card" with all others as descendants.
5. Take into consideration what UI BEST serves the consumer — varieties for enter, tables for knowledge,
progress bars for standing, chips for tags, buttons for actions.
OUTPUT FORMAT (strict JSON, nothing else):
{
"surfaceId": "unique-id",
"parts": [
{"id": "root", "type": "card", "title": "...", "children": ["c1", "c2"]},
{"id": "c1", "sort": "textual content", "textual content": "...", "fashion": "headline"},
...
],
"dataModel": { ... }
}
"""
def generate_ui(user_query: str) -> Non-obligatory[A2UISurface]:
console.print(f" [dim]Producing UI for:[/] [bold]{user_query}[/]")
response = llm([
{"role": "system", "content": A2UI_GENERATION_PROMPT},
{"role": "user", "content": user_query},
], max_tokens=1200)
if not response:
return None
uncooked = response.decisions[0].message.content material
attempt:
cleaned = re.sub(r'```jsons*|s*```', '', uncooked).strip()
spec = json.masses(cleaned)
besides json.JSONDecodeError:
console.print(f"[red]Did not parse generated UI: {uncooked[:200]}[/]")
return None
parts = []
for c in spec.get("parts", []):
parts.append(A2UIComponent(
id=c.get("id", str(uuid.uuid4())[:6]),
sort=c.get("sort", "textual content"),
properties={ok: v for ok, v in c.objects() if ok not in ("id", "sort", "kids")},
kids=c.get("kids", []),
))
floor = A2UISurface(
surface_id=spec.get("surfaceId", f"gen-{uuid.uuid4().hex[:6]}"),
parts=parts,
data_model=A2UIDataModel(spec.get("dataModel", {})),
)
return floor
def demo_generative_ui(question: str):
floor = generate_ui(question)
if floor:
console.print(f"n[bold green]Generated {len(floor.parts)} parts:[/]")
registry.render_tree(floor)
console.print()
varieties = {}
for c in floor.parts:
varieties[c.type] = varieties.get(c.sort, 0) + 1
console.print(" [dim]Part varieties used:[/] " + ", ".be part of(f"[cyan]{t}[/]×{n}" for t, n in varieties.objects()))
if floor.data_model.knowledge:
console.print(f" [dim]Knowledge mannequin keys:[/] {checklist(floor.data_model.knowledge.keys())}")
console.print()
console.print("n[bold]Demo 1: Agent generates an onboarding kind[/]")
demo_generative_ui(
"Create a consumer onboarding circulate: gather title, electronic mail, function (dropdown), "
"most popular notification methodology (chips), and a 'Get Began' button."
)
console.print("n[bold]Demo 2: Agent generates an information dashboard[/]")
demo_generative_ui(
"Present a challenge standing dashboard with: challenge title 'Atlas v2', "
"4 crew members, dash progress at 68%, 3 blockers flagged as vital, "
"and motion buttons for 'View Backlog' and 'Schedule Standup'."
)
console.print("n[bold]Demo 3: Agent generates a affirmation dialog[/]")
demo_generative_ui(
"Present a cost affirmation: $2,450 cost to Visa ending 4242, "
"order #ORD-8891, with Approve and Decline buttons."
)
We hand the keys to the LLM and let it generate full A2UI element timber at runtime from plain English descriptions, that is Generative UI in its purest kind. We immediate the mannequin with the A2UI schema and element catalog, and it produces absolutely structured surfaces with playing cards, varieties, chips, progress bars, and knowledge bindings, selecting the perfect UI sample for every question. We run three demos, an onboarding circulate, a challenge dashboard, and a cost affirmation, displaying how the identical agent adapts its interface to wildly completely different contexts with no single hardcoded format.
hdr(4, "State Synchronization — Shared State Between Agent & UI",
"AG-UI syncs state bidirectionally utilizing STATE_SNAPSHOT and STATE_DELTA.n"
"The agent IS the state machine; the UI IS the renderer.n"
"JSON Patch diffs preserve updates minimal and environment friendly.")
class SharedState:
def __init__(self, preliminary: dict = None):
self.state: dict = preliminary or {}
self.historical past: checklist[dict] = []
self.model: int = 0
def snapshot(self) -> AGUIEvent:
return AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {"state": copy.deepcopy(self.state), "model": self.model})
def apply_delta(self, operations: checklist[dict]) -> AGUIEvent:
for op in operations:
path_parts = [p for p in op["path"].cut up("/") if p]
goal = self.state
for half in path_parts[:-1]:
if isinstance(goal, dict):
goal = goal.setdefault(half, {})
elif isinstance(goal, checklist) and half.isdigit():
goal = goal[int(part)]
key = path_parts[-1] if path_parts else None
if secret's None:
proceed
if op["op"] == "exchange":
goal[key] = op["value"]
elif op["op"] == "add":
if isinstance(goal, checklist) and key.isdigit():
goal.insert(int(key), op["value"])
else:
goal[key] = op["value"]
elif op["op"] == "take away":
if isinstance(goal, dict):
goal.pop(key, None)
self.model += 1
self.historical past.append({"model": self.model, "ops": operations})
return AGUIEvent(AGUIEventType.STATE_DELTA, {"delta": operations, "model": self.model})
console.print("n[bold]Demo: Doc overview pipeline — 3 brokers, shared state[/]n")
stream = AGUIEventStream()
state = SharedState({
"doc": {"title": "This autumn Technique Report", "standing": "draft", "word_count": 2840},
"pipeline": {"stage": "analysis", "progress": 0.0},
"brokers": {"lively": "Researcher", "queue": ["Editor", "Reviewer"]},
"suggestions": [],
})
def log_event(occasion: AGUIEvent):
if occasion.sort in (AGUIEventType.STATE_SNAPSHOT, AGUIEventType.STATE_DELTA):
if occasion.sort == AGUIEventType.STATE_DELTA:
ops = occasion.knowledge.get("delta", [])
for op in ops:
console.print(f" [yellow]STATE_DELTA[/] v{occasion.knowledge.get('model')}: "
f"[cyan]{op['op']}[/] {op['path']} → {op.get('worth', '∅')}")
else:
console.print(f" [yellow]STATE_SNAPSHOT[/] v{occasion.knowledge.get('model')}: {checklist(occasion.knowledge['state'].keys())}")
stream.on(log_event)
stream.emit(state.snapshot())
console.print("n[bold green]▸ Researcher agent working...[/]")
stream.emit(state.apply_delta([
{"op": "replace", "path": "/pipeline/stage", "value": "research_complete"},
{"op": "replace", "path": "/pipeline/progress", "value": 0.33},
{"op": "add", "path": "/feedback/0", "value": {"agent": "Researcher", "note": "Added 4 new data sources"}},
]))
console.print("n[bold green]▸ Editor agent working...[/]")
stream.emit(state.apply_delta([
{"op": "replace", "path": "/agents/active", "value": "Editor"},
{"op": "replace", "path": "/pipeline/stage", "value": "editing"},
{"op": "replace", "path": "/pipeline/progress", "value": 0.66},
{"op": "replace", "path": "/document/word_count", "value": 3150},
]))
console.print("n[bold green]▸ Reviewer agent working...[/]")
stream.emit(state.apply_delta([
{"op": "replace", "path": "/agents/active", "value": "Reviewer"},
{"op": "replace", "path": "/pipeline/stage", "value": "review_complete"},
{"op": "replace", "path": "/pipeline/progress", "value": 1.0},
{"op": "replace", "path": "/document/status", "value": "approved"},
]))
console.print(Panel(
json.dumps(state.state, indent=2),
title="[bold]Ultimate shared state after pipeline",
border_style="inexperienced",
))
t = Desk(title="State Historical past (variations)", field=field.ROUNDED)
t.add_column("Model", fashion="cyan", justify="middle")
t.add_column("Operations", fashion="yellow")
for h in state.historical past:
ops_summary = "; ".be part of(f"{o['op']} {o['path']}" for o in h["ops"])
t.add_row(str(h["version"]), ops_summary[:70])
console.print(t)
hdr(5, "Human-in-the-Loop — AG-UI INTERRUPT Occasions",
"When an agent hits a high-stakes motion, it emits an INTERRUPT occasion.n"
"The frontend renders an approval UI. Execution pauses till the humann"
"approves, rejects, or modifies. State is preserved all through.")
@dataclass
class InterruptRequest:
interrupt_id: str
action_description: str
risk_level: str
affected_resources: checklist[str]
proposed_changes: dict
choices: checklist[str] = discipline(default_factory=lambda: ["approve", "reject", "modify"])
@dataclass
class InterruptResponse:
interrupt_id: str
determination: str
modifications: Non-obligatory[dict] = None
class InterruptableAgent:
RISK_RULES = {
"delete": "vital",
"cost": "vital",
"email_all": "excessive",
"publish": "excessive",
"replace": "medium",
"learn": "low",
}
def __init__(self):
self.stream = AGUIEventStream()
self.pending_interrupts: dict[str, InterruptRequest] = {}
def assess_and_maybe_interrupt(self, motion: str, particulars: dict) -> Non-obligatory[InterruptRequest]:
threat = "low"
for key phrase, degree in self.RISK_RULES.objects():
if key phrase in motion.decrease():
threat = degree
break
if threat in ("vital", "excessive"):
interrupt = InterruptRequest(
interrupt_id=str(uuid.uuid4())[:8],
action_description=motion,
risk_level=threat,
affected_resources=particulars.get("assets", []),
proposed_changes=particulars.get("modifications", {}),
)
self.pending_interrupts[interrupt.interrupt_id] = interrupt
self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
"interrupt_id": interrupt.interrupt_id,
"motive": threat,
"description": interrupt.action_description,
"affected_resources": interrupt.affected_resources,
"proposed_changes": interrupt.proposed_changes,
"choices": interrupt.choices,
}))
return interrupt
return None
def resolve_interrupt(self, response: InterruptResponse) -> str:
interrupt = self.pending_interrupts.pop(response.interrupt_id, None)
if not interrupt:
return "No pending interrupt discovered."
if response.determination == "approve":
return f" APPROVED: '{interrupt.action_description}' executing now."
elif response.determination == "reject":
return f" REJECTED: '{interrupt.action_description}' cancelled."
elif response.determination == "modify":
return f" MODIFIED: '{interrupt.action_description}' up to date with: {response.modifications}"
return f"Unknown determination: {response.determination}"
console.print("n[bold]Demo: Agent encounters actions of various threat ranges[/]n")
agent = InterruptableAgent()
actions = [
("Read user profile", {"resources": ["user:123"]}),
("Replace consumer preferences", {"assets": ["user:123"], "modifications": {"theme": "darkish"}}),
("Delete consumer account", {"assets": ["user:123", "data:all"], "modifications": {"motion": "permanent_delete"}}),
("E-mail all 12,000 customers", {"assets": ["email:newsletter"], "modifications": {"topic": "Massive Announcement"}}),
("Publish weblog put up", {"assets": ["post:draft-42"], "modifications": {"standing": "public"}}),
]
def event_logger(occasion):
if occasion.sort == AGUIEventType.INTERRUPT:
d = occasion.knowledge
risk_style = {"vital": "daring crimson", "excessive": "daring yellow"}.get(d["reason"], "white")
console.print(f"n [bold] INTERRUPT EVENT[/]")
console.print(f" Danger: [{risk_style}]{d['reason'].higher()}[/]")
console.print(f" Motion: {d['description']}")
console.print(f" Affected: {d['affected_resources']}")
console.print(f" Choices: {d['options']}")
agent.stream.on(event_logger)
outcomes = []
for action_desc, particulars in actions:
interrupt = agent.assess_and_maybe_interrupt(action_desc, particulars)
if interrupt:
determination = "reject" if interrupt.risk_level == "vital" else "approve"
outcome = agent.resolve_interrupt(InterruptResponse(interrupt.interrupt_id, determination))
else:
outcome = f" AUTO-EXECUTED: '{action_desc}' (low threat, no approval wanted)"
outcomes.append((action_desc, outcome))
console.print()
t = Desk(title="Execution Outcomes", field=field.ROUNDED, show_lines=True)
t.add_column("Motion", fashion="white", width=28)
t.add_column("Consequence", fashion="dim", width=55)
for action_desc, end in outcomes:
t.add_row(action_desc, outcome)
console.print(t)We construct a SharedState engine that emits AG-UI STATE_SNAPSHOT and STATE_DELTA occasions utilizing JSON Patch operations, preserving the agent backend and the frontend UI completely synchronized by way of each mutation. We show this with a three-agent doc overview pipeline through which a Researcher, Editor, and Reviewer every modify the shared state in sequence, and the frontend sees each change the moment it happens. We then implement the AG-UI INTERRUPT sample, through which the agent assesses threat ranges for proposed actions, emits interrupt occasions for any harmful actions, and pauses execution till a human approves, rejects, or modifies the plan.
hdr(6, "Full Pipeline — LLM-Pushed Adaptive UI",
"The entire Agentic UI structure in a single pipeline:n"
" Consumer question → Intent evaluation → UI sample choice →n"
" A2UI technology → AG-UI occasion streaming → State sync → Render")
UI_ROUTER_PROMPT = """
You're a UI routing agent. Given a consumer question, determine what sort of UI to generate.
RESPOND IN JSON ONLY:
wizard
"""
class AgenticUIPipeline:
def __init__(self):
self.stream = AGUIEventStream()
self.state = SharedState({"pipeline": {"stage": "idle"}, "renders": 0})
self.interrupt_agent = InterruptableAgent()
def route(self, question: str) -> dict:
resp = llm([
{"role": "system", "content": UI_ROUTER_PROMPT},
{"role": "user", "content": query},
], max_tokens=300)
if not resp:
return {"intent": "dashboard", "reasoning": "fallback", "ui_complexity": "easy",
"needs_approval": False, "data_requirements": []}
uncooked = resp.decisions[0].message.content material
attempt:
return json.masses(re.sub(r'```jsons*|s*```', '', uncooked).strip())
besides json.JSONDecodeError:
return {"intent": "dashboard", "reasoning": "parse_fallback", "ui_complexity": "easy",
"needs_approval": False, "data_requirements": []}
def run(self, user_query: str):
run_id = str(uuid.uuid4())[:8]
console.print(Panel(f"[bold]{user_query}[/]", title=" Consumer Question", border_style="white"))
self.stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {"run_id": run_id}))
self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "routing"}))
routing = self.route(user_query)
console.print(f"n [bold cyan] Router Choice:[/]")
console.print(f" Intent: [green]{routing.get('intent')}[/] | "
f"Complexity: [yellow]{routing.get('ui_complexity')}[/] | "
f"Approval: {'' if routing.get('needs_approval') else ''}")
console.print(f" Reasoning: [dim]{routing.get('reasoning', '')}[/]")
self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "routing", "outcome": routing}))
self.state.apply_delta([
{"op": "replace", "path": "/pipeline/stage", "value": "generating"},
])
self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "generating_ui"}))
msg_id = str(uuid.uuid4())[:8]
self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {"message_id": msg_id}))
self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {
"message_id": msg_id,
"delta": f"Constructing a {routing.get('intent')} interface for you..."
}))
self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {"message_id": msg_id}))
floor = generate_ui(user_query)
self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "generating_ui"}))
if routing.get("needs_approval") and floor:
self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
"motive": "ui_confirmation",
"description": f"Generated {len(floor.parts)} element UI. Render it?",
"choices": ["render", "regenerate", "cancel"],
}))
console.print("n [bold yellow] INTERRUPT:[/] UI generated, awaiting human approval...")
console.print(" [green]→ Auto-approving for demo...[/]")
if floor:
self.state.apply_delta([
{"op": "replace", "path": "/pipeline/stage", "value": "rendering"},
{"op": "replace", "path": "/renders", "value": self.state.state.get("renders", 0) + 1},
])
console.print(f"n[bold green] Rendered Interface ({len(floor.parts)} parts):[/]n")
registry.render_tree(floor)
self.stream.emit(AGUIEvent(AGUIEventType.CUSTOM, {
"subtype": "a2ui_surface",
"floor": floor.to_messages(),
}))
self.state.apply_delta([{"op": "replace", "path": "/pipeline/stage", "value": "complete"}])
self.stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {"run_id": run_id, "standing": "success"}))
console.print()
event_counts = {}
for e in self.stream.occasions:
event_counts[e.type.value] = event_counts.get(e.sort.worth, 0) + 1
t = Desk(title="Pipeline Occasion Abstract", field=field.ROUNDED)
t.add_column("Occasion Sort", fashion="cyan")
t.add_column("Rely", justify="middle", fashion="inexperienced")
for etype, depend in sorted(event_counts.objects()):
t.add_row(etype, str(depend))
console.print(t)
pipeline = AgenticUIPipeline()
console.print("n[bold]Demo 1: Agent builds a settings kind[/]")
pipeline.run(
"Create a notification settings panel the place I can toggle electronic mail/SMS/push, "
"set quiet hours, and decide a notification sound."
)
pipeline.stream = AGUIEventStream()
pipeline.state = SharedState({"pipeline": {"stage": "idle"}, "renders": 0})
console.print("n[bold]Demo 2: Agent builds an order monitoring dashboard[/]")
pipeline.run(
"Present order #ORD-7742 standing: shipped through FedEx, monitoring 789456123, "
"estimated supply March 24, 2 of three objects delivered. Present a progress bar "
"and motion buttons for 'Contact Help' and 'Request Refund'."
)
hdr(7, "Incremental UI Updates — Reside Floor Modification",
"A2UI surfaces are incrementally updateable. The agent can add, take away,n"
"or modify parts and knowledge bindings on a reside floor withoutn"
"regenerating the entire tree. Important for real-time collaboration.")
class LiveSurface:
def __init__(self, floor: A2UISurface):
self.floor = floor
self.update_log: checklist[str] = []
def add_component(self, element: A2UIComponent, parent_id: Non-obligatory[str] = None):
self.floor.parts.append(element)
if parent_id:
for c in self.floor.parts:
if c.id == parent_id:
c.kids.append(element.id)
break
self.update_log.append(f"ADD {element.sort}#{element.id} → dad or mum:{parent_id or 'root'}")
def update_component(self, component_id: str, new_props: dict):
for c in self.floor.parts:
if c.id == component_id:
c.properties.replace(new_props)
self.update_log.append(f"UPD #{component_id} props: {checklist(new_props.keys())}")
return
self.update_log.append(f"ERR #{component_id} not discovered")
def remove_component(self, component_id: str):
self.floor.parts = [c for c in self.surface.components if c.id != component_id]
for c in self.floor.parts:
if component_id in c.kids:
c.kids.take away(component_id)
self.update_log.append(f"DEL #{component_id}")
def update_data(self, path: str, worth: Any):
self.floor.data_model.knowledge = _set_nested(self.floor.data_model.knowledge, path, worth)
self.update_log.append(f"DATA {path} = {worth}")
def _set_nested(d: dict, path: str, worth: Any) -> dict:
components = [p for p in path.split("/") if p]
d = copy.deepcopy(d)
present = d
for p in components[:-1]:
present = present.setdefault(p, {})
if components:
present[parts[-1]] = worth
return d
console.print("n[bold]Demo: Reside collaborative enhancing — agent modifies UI in real-time[/]n")
preliminary = A2UISurface(
surface_id="task-board",
parts=[
A2UIComponent("board", "card", {"title": " Sprint Board"}, children=["t1", "t2", "t3"]),
A2UIComponent("t1", "chip", {"label": "AUTH-101: Login circulate", "variant": "in_progress"}),
A2UIComponent("t2", "chip", {"label": "AUTH-102: OAuth setup", "variant": "todo"}),
A2UIComponent("t3", "chip", {"label": "AUTH-103: 2FA", "variant": "todo"}),
],
data_model=A2UIDataModel({"dash": {"title": "Dash 14", "velocity": 21}}),
)
reside = LiveSurface(preliminary)
console.print("[bold]Preliminary board:[/]")
registry.render_tree(reside.floor)
console.print("n[bold yellow]Agent updating board in real-time...[/]n")
reside.update_component("t1", {"variant": "finished", "label": " AUTH-101: Login circulate"})
reside.update_component("t2", {"variant": "in_progress", "label": " AUTH-102: OAuth setup"})
reside.add_component(
A2UIComponent("t4", "chip", {"label": "AUTH-104: Password reset", "variant": "todo"}),
parent_id="board"
)
reside.update_data("/dash/velocity", 25)
reside.remove_component("t3")
console.print("[bold]Up to date board:[/]")
registry.render_tree(reside.floor)
console.print()
t = Desk(title="Incremental Replace Log", field=field.ROUNDED)
t.add_column("#", fashion="cyan", width=4, justify="middle")
t.add_column("Operation", fashion="yellow")
for i, entry in enumerate(reside.update_log, 1):
t.add_row(str(i), entry)
console.print(t)We wire each piece collectively right into a single AgenticUIPipeline class that takes a consumer question, classifies its intent with an LLM router, selects the appropriate UI sample, generates an A2UI floor, streams the whole course of over AG-UI occasions, manages shared state, and renders the outcome, the entire structure in a single run. We then construct a LiveSurface class that helps incremental A2UI updates: including, modifying, and eradicating parts on an already-rendered floor with out regenerating the entire tree, which is important for real-time collaborative experiences. We demo this with a dash board that an agent updates reside, marking duties full, including new ones, and adjusting knowledge mannequin values, all tracked in an in depth operation log.
hdr(8, "Reference — The Agentic UI Protocol Stack",
"How AG-UI, A2UI, MCP, and A2A match collectively within the fashionable agent structure.")
console.print(Panel("""
[bold white]THE AGENTIC UI STACK (2026)[/]
┌──────────────────────────────────────────────────────┐
│ [bold cyan]USER INTERFACE[/] (React, Flutter, SwiftUI, Terminal) │
│ Renders native widgets from element specs │
└────────────────────────┬─────────────────────────────┘
│ A2UI element timber (JSON)
│ AG-UI occasions (SSE / WebSocket)
┌────────────────────────┴─────────────────────────────┐
│ [bold yellow]AG-UI PROTOCOL[/] (Agent Consumer Interplay) │
│ • Occasion streaming (TEXT, TOOL_CALL, STATE, INTERRUPT)│
│ • Bidirectional state sync (SNAPSHOT + DELTA) │
│ • Human-in-the-loop (INTERRUPT → approval circulate) │
└────────────────────────┬─────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────┐
│ [bold magenta]AGENT RUNTIME[/] (LangGraph, CrewAI, customized, and so on.) │
│ • Generates A2UI surfaces (Generative UI) │
│ • Manages shared state │
│ • Orchestrates sub-agents through A2A protocol │
│ • Accesses instruments through MCP protocol │
└────────────────────────┬─────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────┐
│ [bold green]LLM BACKBONE[/] (GPT, Claude, Gemini, and so on.) │
│ • Generates element timber as structured output │
│ • Causes about UI patterns per context │
│ • Streams tokens for real-time rendering │
└──────────────────────────────────────────────────────┘
[dim]Protocol roles:
AG-UI = Agent Consumer (streaming occasions, state, HITL)
A2UI = Agent → UI (declarative element specs)
A2A = Agent → Agent (delegation, sub-agents)
MCP = Agent → Instruments (operate calling, context)[/]
""", title="[bold]Structure Reference", border_style="cyan"))
ref = Desk(title="Agentic UI Ideas — Fast Reference", field=field.DOUBLE_EDGE, show_lines=True)
ref.add_column("Idea", fashion="daring cyan", width=22)
ref.add_column("What It Does", fashion="white", width=35)
ref.add_column("Key Mechanism", fashion="yellow", width=28)
ref.add_row("AG-UI Occasions", "Stream agent actions to frontend in real-time", "SSE/WebSocket + ~16 occasion varieties")
ref.add_row("A2UI Elements", "Declarative UI timber — secure, moveable, native", "Flat JSON + widget registry")
ref.add_row("State Sync", "Hold agent & UI state in lockstep", "STATE_SNAPSHOT + STATE_DELTA")
ref.add_row("Generative UI", "LLM generates UI at runtime, not simply textual content", "A2UI JSON as structured output")
ref.add_row("INTERRUPT (HITL)", "Pause execution for human approval", "INTERRUPT occasion → approval circulate")
ref.add_row("Incremental Replace","Modify reside surfaces with out full regeneration", "A2UI updateComponents message")
ref.add_row("Knowledge Binding", "UI reads from a shared knowledge mannequin", "JSON Pointer paths (/path/to/val)")
ref.add_row("Widget Registry", "Consumer maps summary varieties to native widgets", "Catalog of trusted parts")
console.print(ref)
console.print(Panel(
"[bold green]Tutorial full![/]nn"
"[dim]What you constructed:[/]n"
" • A full AG-UI occasion system with all 16 occasion typesn"
" • An A2UI renderer with flat adjacency-list parts + knowledge bindingn"
" • LLM-powered Generative UI that creates interfaces from pure languagen"
" • Bidirectional state sync with JSON Patch deltasn"
" • Human-in-the-loop interrupt and approval flowsn"
" • Incremental reside floor updatesnn"
"[dim]To go additional:[/]n"
" • Serve AG-UI occasions over actual SSE with FastAPIn"
" • Connect with CopilotKit React parts for an actual frontendn"
" • Use Pydantic AI's AGUIAdapter for manufacturing agent hostingn"
" • Add A2A protocol for multi-agent delegationn"
" • Deploy on AWS Bedrock AgentCore with native AG-UI assist",
title="[bold] What's Subsequent?",
border_style="inexperienced",
padding=(1, 2),
))
We shut with a visible protocol stack diagram displaying precisely how AG-UI, A2UI, A2A, and MCP match collectively within the fashionable agentic structure, from the LLM spine on the backside to the native UI on the high. We offer a quick-reference desk mapping each idea we constructed, occasion streaming, element timber, state sync, generative UI, interrupts, incremental updates, knowledge binding, and widget registries, to their core mechanisms. We level the way in which ahead to manufacturing: serving AG-UI occasions over actual SSE with FastAPI, connecting to CopilotKit React parts, utilizing Pydantic AI’s AGUIAdapter, and deploying on AWS Bedrock AgentCore.
In conclusion, we’ve got a completely useful Agentic UI pipeline that takes a easy natural-language question and transforms it right into a structured, interactive interface powered by an clever agent. We don’t simply assemble parts; we perceive how every layer operates and connects, from real-time AG-UI occasion streaming and declarative A2UI interface definitions to state synchronization by way of JSON Patch and enforced human-in-the-loop security mechanisms. This readability permits us to motive about system conduct, debug successfully, and lengthen performance with out counting on black-box abstractions. Additionally, we depart with the flexibility to design our personal agent-driven UI methods, adapt them to completely different use circumstances, and confidently construct production-ready experiences the place brokers and interfaces evolve collectively in a managed, clear, and scalable method.
Try the Full Codes with Notebook here. Additionally, be happy to comply with us on Twitter and don’t neglect to hitch our 130k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Have to associate with us for selling your GitHub Repo OR Hugging Face Web page OR Product Launch OR Webinar and so on.? Connect with us
The put up A Coding Deep Dive into Agentic UI, Generative UI, State Synchronization, and Interrupt-Pushed Approval Flows appeared first on MarkTechPost.
