Add agent development guide to leverage the agent SDK

It also updates the method name from `get_effective_catalog` to
`get_selected_catalog` to remove any confusion.

The `fix` method is renamed to `validate_and_fix` to be more clear.

Author: nan-yu

a2a_agents/python/a2ui_agent/agent_development.md

@@ -0,0 +1,232 @@
+# A2A Agent Development Guide
+
+This guide explains how to build AI agents that generate A2UI interfaces using the `a2ui_agent` SDK. The SDK simplifies schema management, prompt engineering, and message validation for A2A (Agent-to-Agent/Agent-to-Client) communication.
+
+## Core Concepts
+
+The `a2ui_agent` SDK revolves around three main classes:
+
+*   **`CustomCatalogConfig`**: Defines the metadata for a component catalog (name, schema path, examples path).
+*   **`A2uiCatalog`**: Represents a processed catalog, providing methods for validation and LLM instruction rendering.
+*   **`A2uiSchemaManager`**: The central coordinator that loads catalogs, manages versioning, and generates system prompts.
+
+## Generating A2UI Messages
+
+### Step 1: Set up the Schema Manager
+
+The first step in any A2UI-enabled agent is initializing the `A2uiSchemaManager`.
+
+```python
+from a2ui.inference.schema.manager import A2uiSchemaManager, CustomCatalogConfig
+
+schema_manager = A2uiSchemaManager(
+    version="0.8",
+    basic_examples_path="path/to/basic/examples",
+    custom_catalogs=[
+        CustomCatalogConfig(
+            name="my_custom_catalog",
+            catalog_path="path/to/catalog.json",
+            examples_path="path/to/examples"
+        )
+    ]
+)
+```
+
+Notes:
+- The `custom_catalogs` parameter is optional. If not provided, the schema manager will use the basic catalog maintained by the A2UI team.
+- The provided custom catalog must be freestanding, i.e. it should not reference any external schemas or components, except for the common types.
+- If you have a modular catalog that references other catalogs, refer to [Freestanding Catalogs](../../../docs/catalogs.md#freestanding-catalogs) for more information.
+
+### Step 2: Generate System Prompt
+
+Use the `generate_system_prompt` method to assemble the LLM's system instructions. This method takes your high-level descriptions (role, workflow, UI goals) and automatically injects the relevant A2UI JSON Schema and few-shot examples from your catalog configuration.
+
+```python
+instruction = schema_manager.generate_system_prompt(
+    role_description="You are a helpful assistant...",
+    workflow_description="Analyze the request and return UI...",
+    ui_description="Use the following components...",
+    include_schema=True,  # Injects the raw JSON schema
+    include_examples=True, # Injects few-shot examples
+    allowed_components=["Heading", "Text", "Button"] # Optional: prune schema to save tokens
+)
+```
+
+### Step 3: Build an LLM Agent with the System Prompt
+
+Configure your `LlmAgent` using the generated system instructions. This agent serves as the core logic for interpreting user queries and deciding when to generate rich UI responses.
+
+```python
+from google.adk.agents.llm_agent import LlmAgent
+from google.adk.models.lite_llm import LiteLlm
+
+agent = LlmAgent(
+    model=LiteLlm(model=LITELLM_MODEL),
+    name="Your agent name",
+    description="Your agent description.",
+    instruction=instruction,
+    tools=[Your tools],
+)
+```
+
+### Step 4: Process Request and Stream UI
+
+The final step is to build an executor (or a custom streaming handler) that manages the runtime lifecycle of a request: running the LLM, validating the generated JSON, and streaming parts to the client.
+
+#### 4a. Build an Agent Executor
+Build an agent executor that uses the agent to process requests.
+
+```python
+from a2a.server.agent_execution import AgentExecutor
+
+class MyAgentExecutor(AgentExecutor):
+    def __init__(self, agent: LlmAgent, ...):
+        self.agent = agent
+        ...
+
+agent_executor = MyAgentExecutor(
+    agent=agent,
+    ...
+)
+```
+
+#### 4b. Parse, Validate, and Fix LLM Output
+To ensure reliability, always validate the LLM's JSON output before returning it. The SDK's `A2uiCatalog` provides a validator that checks the payload against the A2UI schema. If the payload is invalid, the validator will attempt to fix it.
+
+```python
+# Get the catalog for the current request
+selected_catalog = schema_manager.get_selected_catalog()
+
+try:
+    # Parse the LLM's JSON part
+    parsed_json = json.loads(json_string)
+    
+    # Validate and fix against the schema
+    selected_catalog.payload_fixer.validate_and_fix(parsed_json)
+except Exception as e:
+    # Handle validation errors (e.g., log error or retry with correction prompt)
+    print(f"Validation failed: {e}")
+```
+
+#### 4c. Stream the A2UI Payload
+After parsing and validating the A2UI JSON payloads, wrap them in an A2A DataPart and stream them to the client.
+
+To ensure the A2UI Renderers on the frontend recognize the data, add `{"mimeType": "application/json+a2ui"}` to the DataPart's metadata.
+
+**Recommendation:** Use the [create_a2ui_datapart](src/a2ui/extension/a2ui_extension.py#L37-L54) helper method to convert A2UI JSON payloads into an A2A DataPart.
+
+## Use Cases
+
+### 1. Simple Agents with Static Schemas
+
+For agents with a fixed set of UI capabilities, simply use the `schema_manager` to generate the system instruction.
+
+**Example Samples:** [contact_lookup](../../../samples/agent/adk/contact_lookup), [restaurant_finder](../../../samples/agent/adk/restaurant_finder)
+
+```python
+# Generate system prompt
+instruction = schema_manager.generate_system_prompt(
+    role_description="You are a helpful assistant...",
+    workflow_description="Analyze the request and return UI...",
+    ui_description="Use the following components...",
+    include_schema=True,
+    include_examples=True,
+)
+
+# Use with your LLM framework (e.g., ADK)
+agent = LlmAgent(instruction=instruction, ...)
+```
+
+### 2. Dynamic Schemas (Context-Aware)
+
+Some agents may need to attach different catalogs or examples depending on the user's request, client capabilities, or conversational context. This is common for dashboard-style agents that support multiple distinct visualization types (e.g., Charts vs. Maps).
+
+**Example Sample:** [rizzcharts](../../../samples/agent/adk/rizzcharts)
+
+#### 2a. Injecting Catalogs into Session State
+In a dynamic scenario, you don't provide a static catalog to the agent. Instead, you resolve the selected catalog at runtime (e.g., during session preparation) and store it in the session state.
+
+```python
+# In your AgentExecutor subclass
+async def _prepare_session(self, context, run_request, runner):
+    session = await super()._prepare_session(context, run_request, runner)
+    
+    # 1. Determine client capabilities from metadata
+    capabilities = context.message.metadata.get("a2ui_client_capabilities")
+    
+    # 2. Get selected catalog and load examples
+    a2ui_catalog = self.schema_manager.get_selected_catalog(
+        client_ui_capabilities=capabilities
+    )
+    examples = self.schema_manager.load_examples(a2ui_catalog, validate=True)
+
+    # 3. Store in session state for tool access
+    await runner.session_service.append_event(
+        session,
+        Event(
+            actions=EventActions(
+                state_delta={
+                    "system:a2ui_enabled": True,
+                    "system:a2ui_catalog": a2ui_catalog,
+                    "system:a2ui_examples": examples,
+                }
+            ),
+        ),
+    )
+    return session
+```
+
+#### 2b. Accessing Catalogs via Providers
+The `SendA2uiToClientToolset` can use **Providers**—callables that retrieve the catalog and examples from the current context state at runtime.
+
+```python
+# Providers that read from context state
+def get_a2ui_catalog(ctx: ReadonlyContext):
+    return ctx.state.get("system:a2ui_catalog")
+
+def get_a2ui_examples(ctx: ReadonlyContext):
+    return ctx.state.get("system:a2ui_examples")
+
+# Initialize the toolset with providers
+ui_toolset = SendA2uiToClientToolset(
+    a2ui_enabled=True,
+    a2ui_catalog=get_a2ui_catalog,
+    a2ui_examples=get_a2ui_examples,
+)
+```
+
+#### 2c. Runtime Validation
+
+When the LLM calls the UI tool, the toolset uses the dynamic catalog to:
+1.  **Generate Instructions**: Automatically inject the specific schema and examples into the LLM's system prompt for that turn.
+2.  **Validate and Fix Payloads**: Automatically validate and fix the LLM's generated JSON against the specific `A2uiCatalog` object's validator and auto-fixer.
+
+### 3. Orchestration and Delegation
+
+Orchestrator agents delegate work to sub-agents. They often need to propagate UI capabilities and handle cross-agent UI state.
+
+**Example Sample:** [orchestrator](../../../samples/agent/adk/orchestrator)
+
+The orchestrator inspects sub-agent capabilities and aggregates their supported catalog IDs into its own `AgentCard`.
+
+```python
+# Aggregating capabilities from sub-agents
+supported_catalog_ids = set()
+for subagent in subagents:
+    # ... fetch subagent_card ...
+    for extension in subagent_card.capabilities.extensions:
+        if extension.uri == A2UI_EXTENSION_URI:
+            supported_catalog_ids.update(extension.params.get("supportedCatalogIds") or [])
+
+# Creating the orchestrator's AgentCard
+agent_card = AgentCard(
+    capabilities=AgentCapabilities(
+        extensions=[
+            get_a2ui_agent_extension(supported_catalog_ids=list(supported_catalog_ids))
+        ]
+    )
+)
+```
+
+
+

a2a_agents/python/a2ui_agent/src/a2ui/extension/send_a2ui_to_client_toolset.py

@@ -267,7 +267,7 @@ async def run_async(
           )
 
         a2ui_catalog = await self._resolve_a2ui_catalog(tool_context)
-        a2ui_json_payload = a2ui_catalog.payload_fixer.fix(a2ui_json)
+        a2ui_json_payload = a2ui_catalog.payload_fixer.validate_and_fix(a2ui_json)
 
         logger.info(
             f"Validated call to tool {self.TOOL_NAME} with {self.A2UI_JSON_ARG_NAME}"

a2a_agents/python/a2ui_agent/src/a2ui/inference/schema/manager.py

@@ -288,12 +288,12 @@ def _determine_catalog(
         f" {list(self._supported_catalogs.keys())}"
     )
 
-  def get_effective_catalog(
+  def get_selected_catalog(
       self,
       client_ui_capabilities: Optional[dict[str, Any]] = None,
       allowed_components: List[str] = [],
   ) -> A2uiCatalog:
-    """Gets the effective catalog after selection and component pruning."""
+    """Gets the selected catalog after selection and component pruning."""
     catalog = self._determine_catalog(client_ui_capabilities)
     pruned_catalog = catalog.with_pruned_components(allowed_components)
     return pruned_catalog
@@ -324,7 +324,7 @@ def generate_system_prompt(
     if ui_description:
       parts.append(f"## UI Description:\n{ui_description}")
 
-    final_catalog = self.get_effective_catalog(
+    final_catalog = self.get_selected_catalog(
         client_ui_capabilities, allowed_components
     )
 

a2a_agents/python/a2ui_agent/src/a2ui/inference/schema/payload_fixer.py

@@ -59,8 +59,8 @@ def _parse(self, payload: str) -> List[Dict[str, Any]]:
       logger.error(f"Failed to parse JSON: {e}")
       raise ValueError(f"Failed to parse JSON: {e}")
 
-  def fix(self, payload: str) -> List[Dict[str, Any]]:
-    """Applies autofixes to a raw JSON string and returns the parsed payload.
+  def validate_and_fix(self, payload: str) -> List[Dict[str, Any]]:
+    """Validates and applies autofixes to a raw JSON string and returns the parsed payload.
 
     Args:
       payload: The raw JSON string from the LLM.

a2a_agents/python/a2ui_agent/tests/extension/test_send_a2ui_to_client_toolset.py

@@ -193,7 +193,7 @@ async def test_send_tool_run_async_valid():
   tool_context_mock.actions = MagicMock(skip_summarization=False)
 
   valid_a2ui = [{"type": "Text", "text": "Hello"}]
-  catalog_mock.payload_fixer.fix.return_value = valid_a2ui
+  catalog_mock.payload_fixer.validate_and_fix.return_value = valid_a2ui
   args = {
       SendA2uiToClientToolset._SendA2uiJsonToClientTool.A2UI_JSON_ARG_NAME: json.dumps(
           valid_a2ui
@@ -218,7 +218,7 @@ async def test_send_tool_run_async_valid_list():
   tool_context_mock.actions = MagicMock(skip_summarization=False)
 
   valid_a2ui = [{"type": "Text", "text": "Hello"}]
-  catalog_mock.payload_fixer.fix.return_value = valid_a2ui
+  catalog_mock.payload_fixer.validate_and_fix.return_value = valid_a2ui
   args = {
       SendA2uiToClientToolset._SendA2uiJsonToClientTool.A2UI_JSON_ARG_NAME: json.dumps(
           valid_a2ui
@@ -250,7 +250,9 @@ async def test_send_tool_run_async_missing_arg():
 @pytest.mark.asyncio
 async def test_send_tool_run_async_invalid_json():
   catalog_mock = MagicMock(spec=A2uiCatalog)
-  catalog_mock.payload_fixer.fix.side_effect = Exception("Failed to parse JSON")
+  catalog_mock.payload_fixer.validate_and_fix.side_effect = Exception(
+      "Failed to parse JSON"
+  )
   tool = SendA2uiToClientToolset._SendA2uiJsonToClientTool(catalog_mock, "examples")
   args = {
       SendA2uiToClientToolset._SendA2uiJsonToClientTool.A2UI_JSON_ARG_NAME: "{invalid"
@@ -263,7 +265,7 @@ async def test_send_tool_run_async_invalid_json():
 @pytest.mark.asyncio
 async def test_send_tool_run_async_schema_validation_fail():
   catalog_mock = MagicMock(spec=A2uiCatalog)
-  catalog_mock.payload_fixer.fix.side_effect = Exception(
+  catalog_mock.payload_fixer.validate_and_fix.side_effect = Exception(
       "'text' is a required property"
   )
   tool = SendA2uiToClientToolset._SendA2uiJsonToClientTool(catalog_mock, "examples")

a2a_agents/python/a2ui_agent/tests/inference/test_payload_fixer.py

@@ -68,7 +68,7 @@ def test_fix_payload_success_first_time():
   fixer = A2uiPayloadFixer(catalog_mock)
 
   valid_json = '[{"type": "Text", "text": "Hello"}]'
-  result = fixer.fix(valid_json)
+  result = fixer.validate_and_fix(valid_json)
 
   assert result == [{"type": "Text", "text": "Hello"}]
   catalog_mock.validator.validate.assert_called_once()
@@ -90,7 +90,7 @@ def side_effect(instance):
   fixer = A2uiPayloadFixer(catalog_mock)
 
   malformed_json = '[{"type": "Text", "text": "Hello"},]'
-  result = fixer.fix(malformed_json)
+  result = fixer.validate_and_fix(malformed_json)
 
   assert result == [{"type": "Text", "text": "Hello"}]
   assert "Initial A2UI payload validation failed" in caplog.text

a2a_agents/python/a2ui_agent/tests/integration/verify_load_real.py

@@ -23,7 +23,7 @@ def verify():
   print('Verifying A2uiSchemaManager...')
   try:
     manager = A2uiSchemaManager('0.8', schema_modifiers=[remove_strict_validation])
-    catalog = manager.get_effective_catalog()
+    catalog = manager.get_selected_catalog()
     catalog_components = catalog.catalog_schema[CATALOG_COMPONENTS_KEY]
     print(f'Successfully loaded 0.8: {len(catalog_components)} components')
     print(f'Components found: {list(catalog_components.keys())[:5]}...')
@@ -384,7 +384,7 @@ def verify():
 
   try:
     manager = A2uiSchemaManager('0.9', schema_modifiers=[remove_strict_validation])
-    catalog = manager.get_effective_catalog()
+    catalog = manager.get_selected_catalog()
     catalog_components = catalog.catalog_schema[CATALOG_COMPONENTS_KEY]
     print(f'Successfully loaded 0.9: {len(catalog_components)} components')
     print(f'Components found: {list(catalog_components.keys())}...')

samples/agent/adk/contact_lookup/agent.py

@@ -142,8 +142,8 @@ async def stream(self, query, session_id) -> AsyncIterable[dict[str, Any]]:
     current_query_text = query
 
     # Ensure catalog schema was loaded
-    effective_catalog = self._schema_manager.get_effective_catalog()
-    if self.use_ui and not effective_catalog.catalog_schema:
+    selected_catalog = self._schema_manager.get_selected_catalog()
+    if self.use_ui and not selected_catalog.catalog_schema:
       logger.error(
           "--- ContactAgent.stream: A2UI_SCHEMA is not loaded. "
           "Cannot perform UI validation. ---"
@@ -245,7 +245,7 @@ async def stream(self, query, session_id) -> AsyncIterable[dict[str, Any]]:
             logger.info(
                 "--- ContactAgent.stream: Validating against A2UI_SCHEMA... ---"
             )
-            effective_catalog.validator.validate(parsed_json_data)
+            selected_catalog.validator.validate(parsed_json_data)
             # --- End New Validation Steps ---
 
             logger.info(

samples/agent/adk/contact_multiple_surfaces/agent.py

@@ -151,8 +151,8 @@ async def stream(self, query, session_id) -> AsyncIterable[dict[str, Any]]:
     current_query_text = query
 
     # Ensure schema was loaded
-    effective_catalog = self.schema_manager.get_effective_catalog()
-    if self.use_ui and not effective_catalog.catalog_schema:
+    selected_catalog = self.schema_manager.get_selected_catalog()
+    if self.use_ui and not selected_catalog.catalog_schema:
       logger.error(
           "--- ContactAgent.stream: A2UI_SCHEMA is not loaded. "
           "Cannot perform UI validation. ---"
@@ -323,7 +323,7 @@ async def stream(self, query, session_id) -> AsyncIterable[dict[str, Any]]:
               # logger.info(
               #     "--- ContactAgent.stream: Validating against A2UI_SCHEMA... ---"
               # )
-              # effective_catalog.validator.validate(parsed_json_data)
+              # selected_catalog.validator.validate(parsed_json_data)
               # --- End New Validation Steps ---
 
             logger.info(

samples/agent/adk/contact_multiple_surfaces/agent_executor.py

@@ -92,7 +92,7 @@ async def execute(
               client_ui_capabilities = part.root.data["metadata"][
                   "a2uiClientCapabilities"
               ]
-              catalog = agent.schema_manager.get_effective_catalog(
+              catalog = agent.schema_manager.get_selected_catalog(
                   client_ui_capabilities=client_ui_capabilities
               )
               catalog_schema_str = catalog.render_as_llm_instructions()