feat(python): add A2UI skill generator for managed agents - #2376
Conversation
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the a2ui.skill_generator module, which provides a CLI and core engine for synthesizing custom A2UI agent skills from component catalogs and example payloads. It includes configuration classes, code generation templates, a pre-flight validation script, and comprehensive unit tests. The review feedback highlights several improvement opportunities: updating the CLI help text to clarify that only JSON catalogs are supported, adding robust type checks to prevent crashes when processing non-dictionary JSON schemas or property specifications, and refactoring duplicated example-parsing logic between _generate_references and _collect_examples to improve maintainability.
| action="append", | ||
| dest="catalogs", | ||
| help="Path to component catalog JSON/YAML definition file (can specify multiple)", | ||
| ) |
There was a problem hiding this comment.
The CLI help string mentions support for YAML catalog definition files, but the generator (generator.py) only parses JSON files using json.load. To prevent runtime crashes when users attempt to pass YAML files, we should either implement YAML parsing or update the CLI help string to accurately reflect that only JSON is supported.
help="Path to component catalog JSON definition file (can specify multiple)",| except Exception: | ||
| return |
There was a problem hiding this comment.
If the loaded JSON schema is not a dictionary (e.g., a boolean schema like true or false which is valid in newer JSON Schema drafts, or an array), calling .get() or .items() on it later will raise an AttributeError and crash the generator. We should explicitly verify that schema is a dictionary.
| except Exception: | |
| return | |
| except Exception: | |
| return | |
| if not isinstance(schema, dict): | |
| return |
| for name, spec in (schema.get("properties") or {}).items(): | ||
| if name == "component": # the discriminator, not an authored prop | ||
| continue | ||
| if name not in props: | ||
| kind = spec.get("type") or ("binding" if "$ref" in spec else "any") | ||
| props[name] = {"type": str(kind)} | ||
| if name in required: | ||
| props[name] = {**props[name], "required": True} |
There was a problem hiding this comment.
In JSON Schema, a property schema can be a boolean (e.g., {"properties": {"foo": true}}). If spec is a boolean, calling spec.get() will raise an AttributeError. We should verify that spec is a dictionary before accessing its keys.
| for name, spec in (schema.get("properties") or {}).items(): | |
| if name == "component": # the discriminator, not an authored prop | |
| continue | |
| if name not in props: | |
| kind = spec.get("type") or ("binding" if "$ref" in spec else "any") | |
| props[name] = {"type": str(kind)} | |
| if name in required: | |
| props[name] = {**props[name], "required": True} | |
| for name, spec in (schema.get("properties") or {}).items(): | |
| if name == "component": # the discriminator, not an authored prop | |
| continue | |
| if not isinstance(spec, dict): | |
| continue | |
| if name not in props: | |
| kind = spec.get("type") or ("binding" if "$ref" in spec else "any") | |
| props[name] = {"type": str(kind)} | |
| if name in required: | |
| props[name] = {**props[name], "required": True} |
| collected_examples: List[Dict[str, Any]] = [] | ||
|
|
||
| # 1. Parse example messages provided via list | ||
| for idx, item in enumerate(self.examples_raw, 1): | ||
| if isinstance(item, dict): | ||
| stem = ( | ||
| item.get("name") | ||
| or (item.get("payload", {}).get("type") if isinstance(item.get("payload"), dict) else None) | ||
| or f"example_{idx}" | ||
| ) | ||
| collected_examples.append((str(stem), item)) | ||
| elif isinstance(item, list): | ||
| collected_examples.append((f"example_{idx}", item)) | ||
| elif isinstance(item, str): | ||
| if os.path.exists(item): | ||
| with open(item, "r", encoding="utf-8") as f: | ||
| collected_examples.append(( | ||
| os.path.splitext(os.path.basename(item))[0], json.load(f))) | ||
| else: | ||
| try: | ||
| parsed = json.loads(item) | ||
| stem = ( | ||
| parsed.get("name") | ||
| or (parsed.get("payload", {}).get("type") if isinstance(parsed.get("payload"), dict) else None) | ||
| if isinstance(parsed, dict) else f"example_{idx}" | ||
| ) | ||
| collected_examples.append((str(stem or f"example_{idx}"), parsed)) | ||
| except Exception: | ||
| pass | ||
|
|
||
| # 2. Parse example messages provided via directory path | ||
| if self.examples_path and os.path.exists(self.examples_path): | ||
| if os.path.isdir(self.examples_path): | ||
| ex_files = glob.glob(os.path.join(self.examples_path, "*.json")) | ||
| for ex_file in sorted(ex_files): | ||
| try: | ||
| with open(ex_file, "r", encoding="utf-8") as f: | ||
| collected_examples.append(( | ||
| os.path.splitext(os.path.basename(ex_file))[0], json.load(f))) | ||
| except Exception: | ||
| pass | ||
| elif os.path.isfile(self.examples_path): | ||
| try: | ||
| with open(self.examples_path, "r", encoding="utf-8") as f: | ||
| collected_examples.append(( | ||
| os.path.splitext(os.path.basename(self.examples_path))[0], json.load(f))) | ||
| except Exception: | ||
| pass | ||
|
|
There was a problem hiding this comment.
The logic for parsing and collecting examples from self.examples_raw and self.examples_path is duplicated between _generate_references and _collect_examples. However, the implementation in _collect_examples is less robust (e.g., it does not handle JSON strings or file paths in self.examples_path, and names all list/dict items as "example"). We should unify this logic by making _collect_examples the single source of truth with the robust parsing implementation, and then simply calling it from _generate_references. This also fixes a potential unhandled JSONDecodeError crash when parsing invalid JSON files in self.examples_raw.
| collected_examples: List[Dict[str, Any]] = [] | |
| # 1. Parse example messages provided via list | |
| for idx, item in enumerate(self.examples_raw, 1): | |
| if isinstance(item, dict): | |
| stem = ( | |
| item.get("name") | |
| or (item.get("payload", {}).get("type") if isinstance(item.get("payload"), dict) else None) | |
| or f"example_{idx}" | |
| ) | |
| collected_examples.append((str(stem), item)) | |
| elif isinstance(item, list): | |
| collected_examples.append((f"example_{idx}", item)) | |
| elif isinstance(item, str): | |
| if os.path.exists(item): | |
| with open(item, "r", encoding="utf-8") as f: | |
| collected_examples.append(( | |
| os.path.splitext(os.path.basename(item))[0], json.load(f))) | |
| else: | |
| try: | |
| parsed = json.loads(item) | |
| stem = ( | |
| parsed.get("name") | |
| or (parsed.get("payload", {}).get("type") if isinstance(parsed.get("payload"), dict) else None) | |
| if isinstance(parsed, dict) else f"example_{idx}" | |
| ) | |
| collected_examples.append((str(stem or f"example_{idx}"), parsed)) | |
| except Exception: | |
| pass | |
| # 2. Parse example messages provided via directory path | |
| if self.examples_path and os.path.exists(self.examples_path): | |
| if os.path.isdir(self.examples_path): | |
| ex_files = glob.glob(os.path.join(self.examples_path, "*.json")) | |
| for ex_file in sorted(ex_files): | |
| try: | |
| with open(ex_file, "r", encoding="utf-8") as f: | |
| collected_examples.append(( | |
| os.path.splitext(os.path.basename(ex_file))[0], json.load(f))) | |
| except Exception: | |
| pass | |
| elif os.path.isfile(self.examples_path): | |
| try: | |
| with open(self.examples_path, "r", encoding="utf-8") as f: | |
| collected_examples.append(( | |
| os.path.splitext(os.path.basename(self.examples_path))[0], json.load(f))) | |
| except Exception: | |
| pass | |
| collected_examples = self._collect_examples() |
| def _collect_examples(self): | ||
| """(stem, raw) for every example, keyed by FILE NAME.""" | ||
| out = [] | ||
| for item in self.examples_raw: | ||
| if isinstance(item, (dict, list)): | ||
| out.append(("example", item)) | ||
| elif isinstance(item, str) and os.path.exists(item): | ||
| with open(item, "r", encoding="utf-8") as f: | ||
| out.append((os.path.splitext(os.path.basename(item))[0], json.load(f))) | ||
| path = self.examples_path | ||
| if path and os.path.isdir(path): | ||
| for f_ in sorted(glob.glob(os.path.join(path, "*.json"))): | ||
| try: | ||
| with open(f_, "r", encoding="utf-8") as fh: | ||
| out.append((os.path.splitext(os.path.basename(f_))[0], json.load(fh))) | ||
| except Exception: | ||
| pass | ||
| return out |
There was a problem hiding this comment.
This is the companion suggestion to the refactoring of _generate_references. By implementing the full, robust parsing logic here, we ensure both reference generation and contract documentation use the exact same parsed examples consistently.
def _collect_examples(self) -> List[tuple[str, Any]]:
"""(stem, raw) for every example, keyed by FILE NAME."""
out = []
for idx, item in enumerate(self.examples_raw, 1):
if isinstance(item, dict):
stem = (
item.get("name")
or (item.get("payload", {}).get("type") if isinstance(item.get("payload"), dict) else None)
or f"example_{idx}"
)
out.append((str(stem), item))
elif isinstance(item, list):
out.append((f"example_{idx}", item))
elif isinstance(item, str):
if os.path.exists(item):
try:
with open(item, "r", encoding="utf-8") as f:
out.append((os.path.splitext(os.path.basename(item))[0], json.load(f)))
except Exception:
pass
else:
try:
parsed = json.loads(item)
stem = (
parsed.get("name")
or (parsed.get("payload", {}).get("type") if isinstance(parsed.get("payload"), dict) else None)
if isinstance(parsed, dict) else f"example_{idx}"
)
out.append((str(stem or f"example_{idx}"), parsed))
except Exception:
pass
path = self.examples_path
if path and os.path.exists(path):
if os.path.isdir(path):
for f_ in sorted(glob.glob(os.path.join(path, "*.json"))):
try:
with open(f_, "r", encoding="utf-8") as fh:
out.append((os.path.splitext(os.path.basename(f_))[0], json.load(fh)))
except Exception:
pass
elif os.path.isfile(path):
try:
with open(path, "r", encoding="utf-8") as fh:
out.append((os.path.splitext(os.path.basename(path))[0], json.load(fh)))
except Exception:
pass
return out
3edf9df to
2e6bb9b
Compare
|
Hey @jacobsimionato , here is the prototype for the render-ui skill generator. |
1 participant
Summary
Adds the
a2ui.skill_generatormodule to the Python Agent SDK and updates related agent skill definitions.Key Changes
SkillGeneratorto synthesize bespoke.agents/skills/<name>agent skills from component catalogs and A2UI example payloads.runtime/a2ui-react.js), preflight test validators (scripts/validate_ui.mjs), and runnable reference code (references/).python -m a2ui.skill_generator.SkillGenerator.create-a2ui-renderermeta-skill.