feat(agent-sdk): implement programmatic experiments API and decouple InferenceStrategy/Parser design

Author: gspencergoog

agent_sdks/python/a2ui_agent/agent_development.md

@@ -28,7 +28,8 @@ The first step in any A2UI-enabled agent is initializing the
 
 ```python
 from a2ui.schema.constants import VERSION_0_9
-from a2ui.schema.manager import A2uiSchemaManager, CatalogConfig
+from a2ui.strategies.schema import A2uiSchemaManager
+from a2ui.schema.catalog import CatalogConfig
 from a2ui.basic_catalog.provider import BasicCatalog
 
 # Define your catalogs (basic or bring your own) with optional examples

agent_sdks/python/a2ui_agent/src/a2ui/a2a/parts.py

@@ -14,6 +14,7 @@
 
 import logging
 from typing import Any, Optional, List, AsyncIterable, TYPE_CHECKING
+from a2ui.inference_strategy import Parser
 
 if TYPE_CHECKING:
     from a2ui.parser.streaming import A2uiStreamParser
@@ -87,26 +88,35 @@ def get_a2ui_datapart(part: Part) -> Optional[DataPart]:
 
 def parse_response_to_parts(
     content: str,
+    parser: Optional[Parser] = None,
     validator: Optional[Any] = None,
     fallback_text: Optional[str] = None,
     version: Optional[str] = None,
+    catalog: Optional[Any] = None,
 ) -> List[Part]:
     """Helper to parse LLM response content into A2A Parts, with optional validation.
 
     Args:
         content: The LLM response content, potentially containing A2UI delimiters.
+        parser: Optional Parser instance.
         validator: Optional validator to run against extracted JSON payloads.
         fallback_text: Optional text to return if no parts are successfully created.
         version: Optional version string.
+        catalog: Optional A2uiCatalog for fallback schema parser creation.
 
     Returns:
         A list of A2A Part objects (TextPart and/or DataPart).
     """
-    from a2ui.parser.parser import parse_response
+    if parser is None:
+        if catalog is None:
+            raise ValueError("catalog is required when parser is None.")
+        from a2ui.strategies.schema.parser import A2uiSchemaParser
+
+        parser = A2uiSchemaParser(catalog)
 
     parts = []
     try:
-        response_parts = parse_response(content)
+        response_parts = parser.parse_response(content)
 
         for part in response_parts:
             if part.text:

agent_sdks/python/a2ui_agent/src/a2ui/adk/a2a/part_converter.py

@@ -38,7 +38,8 @@
 
 from a2a import types as a2a_types
 from a2ui.a2a.parts import create_a2ui_part, parse_response_to_parts
-from a2ui.parser.parser import has_a2ui_parts
+from a2ui.inference_strategy import Parser
+from a2ui.strategies.schema.parser import A2uiSchemaParser
 from a2ui.schema import constants
 from a2ui.schema.catalog import A2uiCatalog
 from google.adk.a2a.converters import part_converter
@@ -69,11 +70,13 @@ def __init__(
         bypass_tool_check: bool = False,
         fallback_text: Optional[str] = None,
         version: str = constants.VERSION_0_8,
+        parser: Optional[Parser] = None,
     ):
         self._catalog = a2ui_catalog
         self._bypass_tool_check = bypass_tool_check
         self._fallback_text = fallback_text
         self._version = version
+        self._parser = parser or A2uiSchemaParser(a2ui_catalog)
 
     def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
         """Converts a GenAI part to A2A parts, with A2UI validation.
@@ -117,9 +120,10 @@ def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
 
             if function_response.response:
                 result = function_response.response.get("result")
-                if isinstance(result, str) and has_a2ui_parts(result):
+                if isinstance(result, str) and self._parser.has_a2ui_parts(result):
                     return parse_response_to_parts(
                         result,
+                        parser=self._parser,
                         validator=self._catalog.validator,
                         fallback_text=self._fallback_text,
                         version=self._version,
@@ -133,9 +137,10 @@ def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
 
         # 3. Handle Text-based A2UI (TextPart)
         if text := part.text:
-            if has_a2ui_parts(text):
+            if self._parser.has_a2ui_parts(text):
                 return parse_response_to_parts(
                     text,
+                    parser=self._parser,
                     validator=self._catalog.validator,
                     fallback_text=self._fallback_text,
                     version=self._version,

agent_sdks/python/a2ui_agent/src/a2ui/inference_strategy.py

@@ -13,10 +13,37 @@
 # limitations under the License.
 
 from abc import ABC, abstractmethod
-from typing import Optional, Any
+from typing import Optional, Any, List
+from a2ui.parser.response_part import ResponsePart
+
+
+class Parser(ABC):
+    """Abstract interface defining the response parser and tokenizer."""
+
+    @abstractmethod
+    def parse_response(self, content: str) -> List[ResponsePart]:
+        """Parses full response content into standard JSON payload parts."""
+        pass
+
+    @abstractmethod
+    def process_chunk(self, chunk: str) -> List[ResponsePart]:
+        """Processes a streamed token chunk (incremental parsing)."""
+        pass
+
+    @abstractmethod
+    def has_a2ui_parts(self, content: str) -> bool:
+        """Checks if the content contains formatted structured blocks for this parser."""
+        pass
 
 
 class InferenceStrategy(ABC):
+    """Interface coordinating system prompt generation and response parsing."""
+
+    @property
+    @abstractmethod
+    def parser(self) -> Parser:
+        """The Parser instance associated with this inference strategy."""
+        pass
 
     @abstractmethod
     def generate_system_prompt(
@@ -30,22 +57,9 @@ def generate_system_prompt(
         include_schema: bool = False,
         include_examples: bool = False,
         validate_examples: bool = False,
+        format_strategy: Optional[Any] = None,
     ) -> str:
         """
         Generates a system prompt for all LLM requests.
-
-        Args:
-          role_description: Description of the agent's role.
-          workflow_description: Description of the workflow.
-          ui_description: Description of the UI.
-          client_ui_capabilities: Capabilities reported by the client for targeted schema pruning.
-          allowed_components: List of allowed catalog components.
-          allowed_messages: List of allowed messages.
-          include_schema: Whether to include the schema.
-          include_examples: Whether to include examples.
-          validate_examples: Whether to validate examples.
-
-        Returns:
-          The system prompt.
         """
         pass

agent_sdks/python/a2ui_agent/src/a2ui/parser/streaming.py

@@ -29,11 +29,11 @@
     SURFACE_ID_KEY,
     CATALOG_COMPONENTS_KEY,
 )
-from ..schema.validator import (
+from a2ui.validation.validator import (
     extract_component_ref_fields,
     extract_component_required_fields,
 )
-from ..schema.validator import A2uiValidator
+from a2ui.validation.validator import A2uiValidator
 from a2ui.core.validating import analyze_topology
 from .response_part import ResponsePart
 from a2ui.core.validating.validator import RELAXED_VALIDATION, STRICT_VALIDATION, ValidationConfig

agent_sdks/python/a2ui_agent/src/a2ui/schema/catalog.py

@@ -39,7 +39,7 @@
 )
 
 if TYPE_CHECKING:
-    from .validator import A2uiValidator
+    from a2ui.validation.validator import A2uiValidator
 
 
 @dataclass
@@ -163,6 +163,7 @@ class A2uiCatalog:
     common_types_schema: Dict[str, Any]
     catalog_schema: Dict[str, Any]
     custom_cuttable_keys: Optional[frozenset[str]] = None
+    experiments: Optional[frozenset[str]] = None
 
     @property
     def cuttable_keys(self) -> frozenset[str]:
@@ -181,9 +182,9 @@ def catalog_id(self) -> str:
 
     @property
     def validator(self) -> "A2uiValidator":
-        from .validator import A2uiValidator
+        from a2ui.validation.validator import A2uiValidator
 
-        return A2uiValidator(self)
+        return A2uiValidator(self, experiments=self.experiments)
 
     @property
     def core_catalog(self) -> Catalog[Any, Any]:

agent_sdks/python/a2ui_agent/src/a2ui/strategies/schema/__init__.py

@@ -0,0 +1,23 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .inference_strategy import A2uiSchemaManager
+from .parser import A2uiSchemaParser
+from .prompt_generator import SchemaPromptGenerator
+
+__all__ = [
+    "A2uiSchemaManager",
+    "A2uiSchemaParser",
+    "SchemaPromptGenerator",
+]

…on/a2ui_agent/src/a2ui/schema/manager.py …/strategies/schema/inference_strategy.pyagent_sdks/python/a2ui_agent/src/a2ui/schema/manager.py renamed to agent_sdks/python/a2ui_agent/src/a2ui/strategies/schema/inference_strategy.py agent_sdks/python/a2ui_agent/src/a2ui/schema/manager.py renamed to agent_sdks/python/a2ui_agent/src/a2ui/strategies/schema/inference_strategy.py

@@ -13,17 +13,15 @@
 # limitations under the License.
 
 import copy
-import json
-import logging
-import os
-import importlib.resources
-from typing import Any, Optional, Callable
-from dataclasses import dataclass, field
-from .utils import load_from_bundled_resource
-from ..inference_strategy import InferenceStrategy
-from .constants import *
-from .catalog import CatalogConfig, A2uiCatalog
+from typing import Any, Optional, Callable, Union
+
+from a2ui.schema.utils import load_from_bundled_resource
+from a2ui.inference_strategy import InferenceStrategy, Parser
+from a2ui.schema.constants import *
+from a2ui.schema.catalog import CatalogConfig, A2uiCatalog
 from a2ui.core import A2uiCatalogError
+from a2ui.strategies.schema.parser import A2uiSchemaParser
+from a2ui.strategies.schema.prompt_generator import SchemaPromptGenerator
 
 
 class A2uiSchemaManager(InferenceStrategy):
@@ -37,9 +35,11 @@ def __init__(
         schema_modifiers: Optional[
             list[Callable[[dict[str, Any]], dict[str, Any]]]
         ] = None,
+        experiments: Optional[Union[set[str], frozenset[str]]] = None,
     ):
         self._version = version
         self._accepts_inline_catalogs = accepts_inline_catalogs
+        self.experiments = frozenset(experiments) if experiments else frozenset()
 
         self._server_to_client_schema: dict[str, Any] = {}
         self._common_types_schema: dict[str, Any] = {}
@@ -48,6 +48,19 @@ def __init__(
         self._schema_modifiers = schema_modifiers or []
         self._load_schemas(version, catalogs or [])
 
+    @property
+    def parser(self) -> Parser:
+        """Returns the Parser instance for this strategy."""
+        if not self._supported_catalogs:
+            raise ValueError(
+                "No supported catalogs configured for the schema strategy."
+            )
+        default_catalog = self._supported_catalogs[0]
+        return A2uiSchemaParser(
+            default_catalog,
+            default_catalog.validator,
+        )
+
     @property
     def accepts_inline_catalogs(self) -> bool:
         return self._accepts_inline_catalogs
@@ -98,6 +111,7 @@ def _load_schemas(
                 s2c_schema=self._server_to_client_schema,
                 common_types_schema=self._common_types_schema,
                 custom_cuttable_keys=config.custom_cuttable_keys,
+                experiments=self.experiments,
             )
             self._supported_catalogs.append(catalog)
             if config.examples_path:
@@ -175,6 +189,7 @@ def _select_catalog(
                 catalog_schema=merged_schema,
                 s2c_schema=self._server_to_client_schema,
                 common_types_schema=self._common_types_schema,
+                experiments=self.experiments,
             )
 
         if not client_supported_catalog_ids:
@@ -220,30 +235,24 @@ def generate_system_prompt(
         include_schema: bool = False,
         include_examples: bool = False,
         validate_examples: bool = False,
+        format_strategy: Optional[Any] = None,
     ) -> str:
         """Assembles the final system instruction for the LLM."""
-        parts = [role_description]
-
-        workflow = DEFAULT_WORKFLOW_RULES
-        if workflow_description:
-            workflow += f"\n{workflow_description}"
-        parts.append(f"## Workflow Description:\n{workflow}")
-
-        if ui_description:
-            parts.append(f"## UI Description:\n{ui_description}")
-
         selected_catalog = self.get_selected_catalog(
             client_ui_capabilities, allowed_components, allowed_messages
         )
 
-        if include_schema:
-            parts.append(selected_catalog.render_as_llm_instructions())
-
+        examples_str = ""
         if include_examples:
             examples_str = self.load_examples(
                 selected_catalog, validate=validate_examples
             )
-            if examples_str:
-                parts.append(f"### Examples:\n{examples_str}")
 
-        return "\n\n".join(parts)
+        generator = SchemaPromptGenerator(selected_catalog)
+        return generator.generate(
+            role_description=role_description,
+            workflow_description=workflow_description,
+            ui_description=ui_description,
+            examples=examples_str,
+            include_schema=include_schema,
+        )

agent_sdks/python/a2ui_agent/src/a2ui/strategies/schema/parser.py

@@ -0,0 +1,49 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import List, Optional
+from a2ui.inference_strategy import Parser
+from a2ui.parser.response_part import ResponsePart
+from a2ui.parser.parser import parse_response
+from a2ui.parser.streaming import A2uiStreamParser
+from a2ui.schema.catalog import A2uiCatalog
+from a2ui.validation.validator import A2uiValidator
+
+
+class A2uiSchemaParser(Parser):
+    """Concrete parser implementation for standard A2UI JSON schema responses."""
+
+    def __init__(
+        self,
+        catalog: A2uiCatalog,
+        validator: Optional[A2uiValidator] = None,
+    ):
+        self._catalog = catalog
+        self._validator = validator
+        self._stream_parser: Optional[A2uiStreamParser] = None
+
+    def parse_response(self, content: str) -> List[ResponsePart]:
+        """Parses standard A2UI JSON tags in LLM responses."""
+        return parse_response(content)
+
+    def process_chunk(self, chunk: str) -> List[ResponsePart]:
+        """Processes streamed token chunks incrementally."""
+        if not self._stream_parser:
+            self._stream_parser = A2uiStreamParser(self._catalog)
+        return self._stream_parser.process_chunk(chunk)
+
+    def has_a2ui_parts(self, content: str) -> bool:
+        from a2ui.schema.constants import A2UI_OPEN_TAG
+
+        return A2UI_OPEN_TAG.rstrip(">") in content