Remove notion of "custom catalogs" from agent SDK

The basic catalog maintained by the A2UI team has no difference from
third-party catalogs.

This PR removes the notion of custom catalogs. Each catalog provided at
runtime should be independent and immutable. At build time, catalogs can
refer to components from other catalogs. They need to be bundled into a
free-standing one using the `tools/build_catalog/build_catalog.py`
script.

Fixes a2ui-project#650

Author: nan-yu

a2a_agents/python/a2ui_agent/agent_development.md

@@ -6,7 +6,7 @@ This guide explains how to build AI agents that generate A2UI interfaces using t
 
 The `a2ui_agent` SDK revolves around three main classes:
 
-*   **`CustomCatalogConfig`**: Defines the metadata for a component catalog (name, schema path, examples path).
+*   **`CatalogConfig`**: 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.
 
@@ -17,24 +17,25 @@ The `a2ui_agent` SDK revolves around three main classes:
 The first step in any A2UI-enabled agent is initializing the `A2uiSchemaManager`.
 
 ```python
-from a2ui.inference.schema.manager import A2uiSchemaManager, CustomCatalogConfig
+from a2ui.inference.schema.constants import VERSION_0_8
+from a2ui.inference.schema.manager import A2uiSchemaManager, CatalogConfig
 
 schema_manager = A2uiSchemaManager(
-    version="0.8",
-    basic_examples_path="path/to/basic/examples",
-    custom_catalogs=[
-        CustomCatalogConfig(
+    version=VERSION_0_8,
+    catalogs=[
+        CatalogConfig.bundled(version=VERSION_0_8, examples_path="examples"),
+        CatalogConfig.from_path(
             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.
+- The `catalogs` parameter is optional. If not provided, the schema manager will use the basic catalog maintained by the A2UI team.
+- The provided catalogs must be freestanding, i.e. they 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

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

@@ -94,7 +94,7 @@ def get_a2ui_agent_extension(
   """Creates the A2UI AgentExtension configuration.
 
   Args:
-      accepts_inline_catalogs: Whether the agent accepts inline custom catalogs.
+      accepts_inline_catalogs: Whether the agent accepts inline catalogs.
       supported_catalog_ids: All pre-defined catalogs the agent is known to support.
 
   Returns:

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

@@ -19,22 +19,50 @@
 from dataclasses import dataclass, field, replace
 from typing import Any, Dict, List, Optional, TYPE_CHECKING
 
-from .constants import CATALOG_COMPONENTS_KEY, CATALOG_ID_KEY
-from referencing import Registry, Resource
-
-if TYPE_CHECKING:
-  from .validator import A2uiValidator
-  from .payload_fixer import A2uiPayloadFixer
+from .catalog_provider import A2uiCatalogProvider, FileSystemCatalogProvider, BundledCatalogProvider
+from .constants import CATALOG_COMPONENTS_KEY, CATALOG_ID_KEY, BASIC_CATALOG_NAME
 
 
 @dataclass
-class CustomCatalogConfig:
-  """Configuration for a custom component catalog."""
+class CatalogConfig:
+  """
+  Configuration for a catalog of components.
+
+  A catalog consists of a provider that knows how to load the schema,
+  and optionally a path to examples.
+
+  Attributes:
+    name: The name of the catalog.
+    provider: The provider to use to load the catalog schema.
+    examples_path: The path to the examples directory.
+  """
 
   name: str
-  catalog_path: str
+  provider: A2uiCatalogProvider
   examples_path: Optional[str] = None
 
+  @classmethod
+  def bundled(
+      cls, version: str, examples_path: Optional[str] = None
+  ) -> "CatalogConfig":
+    """Returns a CatalogConfig for the basic bundled catalog."""
+    return cls(
+        name=BASIC_CATALOG_NAME,
+        provider=BundledCatalogProvider(version),
+        examples_path=examples_path,
+    )
+
+  @classmethod
+  def from_path(
+      cls, name: str, catalog_path: str, examples_path: Optional[str] = None
+  ) -> "CatalogConfig":
+    """Returns a CatalogConfig that loads from a file path."""
+    return cls(
+        name=name,
+        provider=FileSystemCatalogProvider(catalog_path),
+        examples_path=examples_path,
+    )
+
 
 @dataclass(frozen=True)
 class A2uiCatalog:
@@ -165,113 +193,6 @@ def load_examples(self, path: Optional[str], validate: bool = False) -> str:
       return ""
     return "\n\n".join(merged_examples)
 
-  @staticmethod
-  def resolve_schema(basic: Dict[str, Any], custom: Dict[str, Any]) -> Dict[str, Any]:
-    """Resolves references in custom catalog schema against the basic catalog.
-
-    Args:
-      basic: The basic catalog schema.
-      custom: The custom catalog schema.
-
-    Returns:
-      A new dictionary with references resolved.
-    """
-    result = copy.deepcopy(custom)
-
-    # Initialize registry with basic catalog and maybe others from basic's $id
-    registry = Registry()
-    if CATALOG_ID_KEY in basic:
-      basic_resource = Resource.from_contents(basic)
-      registry = registry.with_resource(basic[CATALOG_ID_KEY], basic_resource)
-
-    def resolve_ref(ref_uri: str) -> Any:
-      try:
-        resolver = registry.resolver()
-        resolved = resolver.lookup(ref_uri)
-        return resolved.contents
-      except Exception as e:
-        logging.warning("Could not resolve reference %s: %s", ref_uri, e)
-        return None
-
-    def merge_into(target: Dict[str, Any], source: Dict[str, Any]):
-      for key, value in source.items():
-        if key not in target:
-          target[key] = copy.deepcopy(value)
-
-    # Process components
-    if CATALOG_COMPONENTS_KEY in result:
-      comp_dict = result[CATALOG_COMPONENTS_KEY]
-      if "$ref" in comp_dict:
-        resolved = resolve_ref(comp_dict["$ref"])
-        if isinstance(resolved, dict):
-          merge_into(comp_dict, resolved)
-        del comp_dict["$ref"]
-
-    # Process functions
-    if "functions" in result:
-      func_dict = result["functions"]
-      if "$ref" in func_dict:
-        resolved = resolve_ref(func_dict["$ref"])
-        if isinstance(resolved, dict):
-          merge_into(func_dict, resolved)
-        del func_dict["$ref"]
-
-    # Process $defs
-    if "$defs" in result:
-      res_defs = result["$defs"]
-      if "$ref" in res_defs:
-        resolved = resolve_ref(res_defs["$ref"])
-        if isinstance(resolved, dict):
-          merge_into(res_defs, resolved)
-        del res_defs["$ref"]
-
-      for name in ["anyComponent", "anyFunction"]:
-        if name in res_defs:
-          target = res_defs[name]
-          one_of = target.get("oneOf", [])
-          new_one_of = []
-          for item in one_of:
-            if isinstance(item, dict) and "$ref" in item:
-              ref_uri = item["$ref"]
-              # Check if it points to basic collector
-              resolved = resolve_ref(ref_uri)
-              if isinstance(resolved, dict) and "oneOf" in resolved:
-                # Merge oneOf items and resolve transitive refs to components/functions
-                for sub_item in resolved["oneOf"]:
-                  if sub_item not in new_one_of:
-                    new_one_of.append(copy.deepcopy(sub_item))
-                    # Transitive merge: if sub_item is a ref to a component/function
-                    if isinstance(sub_item, dict) and "$ref" in sub_item:
-                      sub_ref = sub_item["$ref"]
-                      if (
-                          sub_ref.startswith("#/components/")
-                          and CATALOG_COMPONENTS_KEY in basic
-                      ):
-                        comp_name = sub_ref.split("/")[-1]
-                        if comp_name in basic[CATALOG_COMPONENTS_KEY]:
-                          if CATALOG_COMPONENTS_KEY not in result:
-                            result[CATALOG_COMPONENTS_KEY] = {}
-                          if comp_name not in result[CATALOG_COMPONENTS_KEY]:
-                            result[CATALOG_COMPONENTS_KEY][comp_name] = copy.deepcopy(
-                                basic[CATALOG_COMPONENTS_KEY][comp_name]
-                            )
-                      elif sub_ref.startswith("#/functions/") and "functions" in basic:
-                        func_name = sub_ref.split("/")[-1]
-                        if func_name in basic["functions"]:
-                          if "functions" not in result:
-                            result["functions"] = {}
-                          if func_name not in result["functions"]:
-                            result["functions"][func_name] = copy.deepcopy(
-                                basic["functions"][func_name]
-                            )
-              else:
-                new_one_of.append(item)
-            else:
-              new_one_of.append(item)
-          target["oneOf"] = new_one_of
-
-    return result
-
   def _validate_example(self, full_path: str, basename: str, content: str) -> bool:
     try:
       json_data = json.loads(content)

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

@@ -0,0 +1,142 @@
+# 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.
+
+"""Module for providing A2UI catalog schemas and resources."""
+
+import json
+import logging
+import os
+import importlib.resources
+from abc import ABC, abstractmethod
+from json.decoder import JSONDecodeError
+from typing import Any, Dict, Optional
+
+from .constants import (
+    A2UI_ASSET_PACKAGE,
+    BASE_SCHEMA_URL,
+    CATALOG_ID_KEY,
+    CATALOG_SCHEMA_KEY,
+    SPEC_VERSION_MAP,
+    find_repo_root,
+)
+
+ENCODING = "utf-8"
+
+
+class A2uiCatalogProvider(ABC):
+  """Abstract base class for providing A2UI schemas and catalogs."""
+
+  @abstractmethod
+  def load(self) -> Dict[str, Any]:
+    """Loads a schema resource.
+
+    Returns:
+      The loaded schema as a dictionary.
+    """
+    pass
+
+
+class FileSystemCatalogProvider(A2uiCatalogProvider):
+  """Loads schemas from the local filesystem."""
+
+  def __init__(self, path: str):
+    self.path = path
+
+  def load(self) -> Dict[str, Any]:
+    try:
+      with open(self.path, "r", encoding=ENCODING) as f:
+        return json.load(f)
+    except (FileNotFoundError, JSONDecodeError) as e:
+      raise IOError(f"Could not load schema from {self.path}: {e}") from e
+
+
+def load_from_bundled_resource(version: str, resource_key: str) -> Dict[str, Any]:
+  """Loads a schema resource from bundled package resources."""
+  spec_map = SPEC_VERSION_MAP.get(version)
+  if not spec_map:
+    raise ValueError(f"Unknown A2UI version: {version}")
+
+  if resource_key not in spec_map:
+    return None
+
+  rel_path = spec_map[resource_key]
+  filename = os.path.basename(rel_path)
+
+  # 1. Try to load from the bundled package resources.
+  try:
+    traversable = importlib.resources.files(A2UI_ASSET_PACKAGE)
+    traversable = traversable.joinpath(version).joinpath(filename)
+    with traversable.open("r", encoding=ENCODING) as f:
+      return json.load(f)
+  except Exception as e:
+    logging.debug("Could not load '%s' from package resources: %s", filename, e)
+
+  # 2. Fallback to local assets
+  # This handles cases where assets might be present in src but not installed
+  try:
+    potential_path = os.path.abspath(
+        os.path.join(
+            os.path.dirname(__file__),
+            "..",
+            "assets",
+            version,
+            filename,
+        )
+    )
+    if os.path.exists(potential_path):
+      provider = FileSystemCatalogProvider(potential_path)
+      return provider.load()
+  except Exception as e:
+    logging.debug("Could not load schema '%s' from local assets: %s", filename, e)
+
+  # 3. Fallback: Source Repository (specification/...)
+  # This handles cases where we are running directly from source tree
+  # And assets are not yet copied to src/a2ui/assets
+  # manager.py is at a2a_agents/python/a2ui_agent/src/a2ui/inference/schema/manager.py
+  # Dynamically find repo root by looking for "specification" directory
+  try:
+    repo_root = find_repo_root(os.path.dirname(__file__))
+    if repo_root:
+      source_path = os.path.join(repo_root, rel_path)
+      if os.path.exists(source_path):
+        provider = FileSystemCatalogProvider(source_path)
+        return provider.load()
+  except Exception as e:
+    logging.debug("Could not load schema from source repo: %s", e)
+
+  raise IOError(f"Could not load schema {filename} for version {version}")
+
+
+class BundledCatalogProvider(A2uiCatalogProvider):
+  """Loads schemas from bundled package resources with fallbacks."""
+
+  def __init__(self, version: str):
+    self.version = version
+
+  def load(self) -> Dict[str, Any]:
+
+    resource = load_from_bundled_resource(self.version, CATALOG_SCHEMA_KEY)
+
+    # Post-load processing for catalogs
+    if CATALOG_ID_KEY not in resource:
+      spec_map = SPEC_VERSION_MAP.get(self.version)
+      rel_path = spec_map[CATALOG_SCHEMA_KEY]
+      # Strip the `json/` part from the catalog file path for the ID.
+      catalog_file = rel_path.replace("/json/", "/")
+      resource[CATALOG_ID_KEY] = BASE_SCHEMA_URL + catalog_file
+
+    if "$schema" not in resource:
+      resource["$schema"] = "https://json-schema.org/draft/2020-12/schema"
+
+    return resource