a2ui-adk

# A2UI + ADK Skill Build Google ADK agents that generate rich, interactive UIs using the A2UI declarative JSON format, styled with Tailwind CSS and shadcn/ui design principles. ## What is A2UI? **A2UI (Agent-to-User Interface)** is an open standard (Apache 2.0, github.com/google/A2UI) that lets agents "speak UI." Agents send declarative JSON describing UI intent; client apps render it using native components (Flutter, Angular, Lit, React, etc.). Core principles: - **Security first**: Declarative data format, not executable code. Clients maintain a catalog of trusted, pre-approved UI components. - **LLM-friendly**: Flat list of components with ID references, easy for LLMs to generate incrementally. - **Framework-agnostic**: Same JSON payload renders on any supported client framework. - **Currently v0.8 (Public Preview)**, v0.9 also available. ## Architecture Overview ``` Agent (ADK + A2UI SDK) → A2UI JSON Response → Transport (A2A/AG UI) → Client Renderer (Lit/Flutter/etc.) ``` 1. **Generation**: Agent uses LLM to generate A2UI JSON payload 2. **Transport**: Sent via A2A protocol or AG UI 3. **Resolution**: Client's A2UI Renderer parses the JSON 4. **Rendering**: Maps abstract components to native widgets ## Key Dependencies ```toml [project] dependencies = [ "a2a-sdk>=0.3.0", "google-adk>=1.8.0", "google-genai>=1.27.0", "litellm", "jsonschema>=4.0.0", "a2ui-agent", # The A2UI agent SDK (from agent_sdks/python in the A2UI repo) "python-dotenv>=1.1.0", "click>=8.1.8", ] ``` The `a2ui-agent` package is sourced from the A2UI repo at `agent_sdks/python` (use `uv` workspace source): ```toml [tool.uv.sources] a2ui-agent = { path = "../../../agent_sdks/python", editable = true } ``` ## Core A2UI SDK Imports ```python # Schema management from a2ui.core.schema.constants import VERSION_0_8, VERSION_0_9, A2UI_OPEN_TAG, A2UI_CLOSE_TAG from a2ui.core.schema.manager import A2uiSchemaManager from a2ui.core.schema.common_modifiers import remove_strict_validation # Catalog (built-in component catalog) from a2ui.basic_catalog.provider import BasicCatalog # Response parsing from a2ui.core.parser.parser import parse_response, ResponsePart # A2A integration helpers from a2ui.a2a import ( create_a2ui_part, # Wrap A2UI JSON as A2A DataPart is_a2ui_part, # Check if an A2A Part contains A2UI data get_a2ui_agent_extension,# Create AgentExtension for agent card parse_response_to_parts, # Parse LLM response → list of A2A Parts try_activate_a2ui_extension, # Activate A2UI extension from request context A2UI_EXTENSION_URI, ) ``` ## Pattern 1: Single Agent with A2UI (e.g., Restaurant Finder) ### Step 1: Schema Manager Setup ```python from a2ui.core.schema.manager import A2uiSchemaManager from a2ui.core.schema.constants import VERSION_0_8 from a2ui.basic_catalog.provider import BasicCatalog from a2ui.core.schema.common_modifiers import remove_strict_validation schema_manager = A2uiSchemaManager( VERSION_0_8, catalogs=[ BasicCatalog.get_config(version=VERSION_0_8, examples_path="examples") ], schema_modifiers=[remove_strict_validation], ) ``` - `examples_path` points to a directory of JSON example files used for few-shot prompting - `remove_strict_validation` strips `additionalProperties: false` to make LLM generation easier ### Step 2: Build the ADK LlmAgent ```python from google.adk.agents.llm_agent import LlmAgent from google.adk.models.lite_llm import LiteLlm ROLE_DESCRIPTION = "You are a helpful assistant. Your final output MUST be an A2UI UI JSON response." UI_DESCRIPTION = "Describe when to use which template/layout..." instruction = schema_manager.generate_system_prompt( role_description=ROLE_DESCRIPTION, ui_description=UI_DESCRIPTION, include_schema=True, # Injects the JSON schema into the prompt include_examples=True, # Injects example A2UI payloads validate_examples=True, # Validates examples against schema at startup ) agent = LlmAgent( model=LiteLlm(model="gemini/gemini-2.5-flash"), name="my_agent", description="Agent description", instruction=instruction, tools=[my_tool_function], ) ``` ### Step 3: Agent Class with Streaming + Validation ```python from google.adk.runners import Runner from google.adk.artifacts import InMemoryArtifactService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.sessions import InMemorySessionService from google.genai import types from a2ui.core.parser.parser import parse_response from a2ui.a2a import parse_response_to_parts class MyAgent: def __init__(self, base_url: str, use_ui: bool = False): self.use_ui = use_ui self._schema_manager = A2uiSchemaManager(...) if use_ui else None self._agent = self._build_agent(use_ui) self._runner = Runner( app_name=self._agent.name, agent=self._agent, artifact_service=InMemoryArtifactService(), session_service=InMemorySessionService(), memory_service=InMemoryMemoryService(), ) async def stream(self, query, session_id): # Create/get session session = await self._runner.session_service.get_session(...) if session is None: session = await self._runner.session_service.create_session(...) # Run agent with retry on validation failure max_retries = 1 for attempt in range(max_retries + 1): message = types.Content(role="user", parts=[types.Part.from_text(text=query)]) final_response = None async for event in self._runner.run_async( user_id=self._user_id, session_id=session.id, new_message=message ): if event.is_final_response(): final_response = "\n".join( [p.text for p in event.content.parts if p.text] ) break else: yield {"is_task_complete": False, "updates": "Processing..."} # Validate A2UI JSON if using UI if self.use_ui and final_response: try: response_parts = parse_response(final_response) for part in response_parts: if part.a2ui_json: selected_catalog = self._schema_manager.get_selected_catalog() selected_catalog.validator.validate(part.a2ui_json) # Valid! Send final response yield { "is_task_complete": True, "parts": parse_response_to_parts(final_response, fallback_text="OK."), } return except Exception as e: if attempt < max_retries: query = f"Previous response invalid: {e}. Retry with valid A2UI JSON." continue # Exhausted retries, send error yield {"is_task_complete": True, "parts": [Part(root=TextPart(text="Error generating UI."))]} return else: yield {"is_task_complete": True, "parts": parse_response_to_parts(final_response)} return ``` ### Step 4: Agent Executor (A2A Server) The executor bridges A2A protocol to your agent. Key pattern: maintain both a UI agent and a text-only agent, selecting based on whether the A2UI extension is active. ```python from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.server.tasks import TaskUpdater from a2a.types import DataPart, Part, TaskState, TextPart from a2a.utils import new_agent_parts_message, new_agent_text_message, new_task from a2ui.a2a import try_activate_a2ui_extension class MyAgentExecutor(AgentExecutor): def __init__(self, ui_agent, text_agent): self.ui_agent = ui_agent self.text_agent = text_agent async def execute(self, context: RequestContext, event_queue: EventQueue): use_ui = try_activate_a2ui_extension(context) agent = self.ui_agent if use_ui else self.text_agent # Extract query from message parts query = context.get_user_input() # Handle A2UI user actions (button clicks, form submissions) for part in (context.message.parts or []): if isinstance(part.root, DataPart) and "userAction" in part.root.data: ui_event = part.root.data["userAction"] action = ui_event.get("actionName") ctx = ui_event.get("context", {}) query = f"User action: {action} with data: {ctx}" break task = context.current_task or new_task(context.message) if not context.current_task: await event_queue.enqueue_event(task) updater = TaskUpdater(event_queue, task.id, task.context_id) async for item in agent.stream(query, task.context_id): if not item["is_task_complete"]: await updater.update_status( TaskState.working, new_agent_text_message(item["updates"], task.context_id, task.id), ) else: await updater.update_status( TaskState.input_required, # or TaskState.completed new_agent_parts_message(item["parts"], task.context_id, task.id), final=True, ) break ``` ### Step 5: Server Entry Point (__main__.py) ```python import click, os, logging, uvicorn from a2a.server.apps import A2AStarletteApplication from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore from dotenv import load_dotenv from starlette.middleware.cors import CORSMiddleware load_dotenv() @click.command() @click.option("--host", default="localhost") @click.option("--port", default=10002) def main(host, port): base_url = f"http://{host}:{port}" ui_agent = MyAgent(base_url=base_url, use_ui=True) text_agent = MyAgent(base_url=base_url, use_ui=False) executor = MyAgentExecutor(ui_agent, text_agent) handler = DefaultRequestHandler(agent_executor=executor, task_store=InMemoryTaskStore()) server = A2AStarletteApplication(agent_card=ui_agent.get_agent_card(), http_handler=handler) app = server.build() app.add_middleware(CORSMiddleware, allow_origin_regex=r"http://localhost:\d+", allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) uvicorn.run(app, host=host, port=port) if __name__ == "__main__": main() ``` ### Step 6: Agent Card with A2UI Extension ```python from a2a.types import AgentCapabilities, AgentCard, AgentSkill from a2ui.a2a import get_a2ui_agent_extension def get_agent_card(self) -> AgentCard: return AgentCard( name="My Agent", description="Agent description", url=self.base_url, version="1.0.0", default_input_modes=["text", "text/plain"], default_output_modes=["text", "text/plain"], capabilities=AgentCapabilities( streaming=True, extensions=[ get_a2ui_agent_extension( self._schema_manager.accepts_inline_catalogs, self._schema_manager.supported_catalog_ids, ) ], ), skills=[AgentSkill(id="my_skill", name="My Skill", description="...", tags=["tag"], examples=["example query"])], ) ``` ## Pattern 2: Orchestrator with Multi-Agent UI Routing For orchestrating multiple sub-agents that each produce A2UI, use: ### Remote Sub-Agent Discovery ```python from a2a.client import A2ACardResolver from google.adk.agents.remote_a2a_agent import RemoteA2aAgent from google.adk.agents.llm_agent import LlmAgent from google.adk.planners.built_in_planner import BuiltInPlanner # Discover sub-agents via A2A card resolution async with httpx.AsyncClient() as client: resolver = A2ACardResolver(httpx_client=client, base_url=subagent_url) subagent_card = await resolver.get_agent_card() # Create remote agent wrapper remote_agent = RemoteA2aAgent( name, subagent_card, description=json.dumps({...}), a2a_part_converter=convert_a2a_part_to_genai_part, genai_part_converter=convert_genai_part_to_a2a_part, a2a_client_factory=A2AClientFactoryWithA2UIMetadata(...) ) # Build orchestrator orchestrator = LlmAgent( model=LiteLlm(model="gemini/gemini-2.5-flash"), name="orchestrator_agent", instruction="Route tasks to the appropriate subagent.", sub_agents=[remote_agent_1, remote_agent_2], planner=BuiltInPlanner(thinking_config=genai_types.ThinkingConfig(include_thoughts=True)), before_model_callback=programmatic_route_user_action, # Route UI actions to correct sub-agent ) ``` ### Part Converters (A2A <-> GenAI) When A2UI parts pass through the orchestrator, they need conversion between A2A and GenAI formats: ```python from a2ui.a2a import is_a2ui_part def convert_a2a_part_to_genai_part(a2a_part): """Serialize A2UI DataParts to text for LLM context.""" if is_a2ui_part(a2a_part): return genai_types.Part(text=a2a_part.model_dump_json()) return part_converter.convert_a2a_part_to_genai_part(a2a_part) def convert_genai_part_to_a2a_part(part): """Deserialize A2UI text back to DataParts.""" if part.text: try: a2a_part = a2a_types.Part.model_validate_json(part.text) if is_a2ui_part(a2a_part): return a2a_part except pydantic.ValidationError: pass return part_converter.convert_genai_part_to_a2a_part(part) ``` ### Surface-to-Subagent Routing Route user actions to the correct sub-agent based on `surfaceId`: