andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-04-12 05:52:39 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

feat: align a2ui extension with protocol spec; add v0.9 support

Browse Source
This commit is contained in:
Greyson Lalonde
2026-03-27 19:56:50 +08:00
parent 01210d582f
commit 36600fa0ff
17 changed files with 3673 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
lib/crewai/src/crewai/a2a/extensions/a2ui/__init__.py
View File

@@ -31,42 +31,118 @@ from crewai.a2a.extensions.a2ui.models import (
SurfaceUpdate,
UserAction,
)
from crewai.a2a.extensions.a2ui.server_extension import A2UIServerExtension
from crewai.a2a.extensions.a2ui.server_extension import (
A2UI_STANDARD_CATALOG_ID,
A2UI_V09_BASIC_CATALOG_ID,
A2UI_V09_EXTENSION_URI,
A2UIServerExtension,
)
from crewai.a2a.extensions.a2ui.v0_9 import (
A2UIEventV09,
A2UIMessageV09,
ActionEvent,
ActionV09,
AudioPlayerV09,
ButtonV09,
CardV09,
CheckBoxV09,
ChoicePickerV09,
ClientDataModel,
ClientErrorV09,
ColumnV09,
CreateSurface,
DateTimeInputV09,
DeleteSurfaceV09,
DividerV09,
IconV09,
ImageV09,
ListV09,
ModalV09,
RowV09,
SliderV09,
TabsV09,
TextFieldV09,
TextV09,
Theme,
UpdateComponents,
UpdateDataModel,
VideoV09,
)
from crewai.a2a.extensions.a2ui.validator import (
validate_a2ui_event,
validate_a2ui_event_v09,
validate_a2ui_message,
validate_a2ui_message_v09,
validate_catalog_components,
validate_catalog_components_v09,
)
__all__ = [
"A2UI_STANDARD_CATALOG_ID",
"A2UI_V09_BASIC_CATALOG_ID",
"A2UI_V09_EXTENSION_URI",
"A2UIClientExtension",
"A2UIEvent",
"A2UIEventV09",
"A2UIMessage",
"A2UIMessageV09",
"A2UIResponse",
"A2UIServerExtension",
"ActionEvent",
"ActionV09",
"AudioPlayer",
"AudioPlayerV09",
"BeginRendering",
"Button",
"ButtonV09",
"Card",
"CardV09",
"CheckBox",
"CheckBoxV09",
"ChoicePickerV09",
"ClientDataModel",
"ClientErrorV09",
"Column",
"ColumnV09",
"CreateSurface",
"DataModelUpdate",
"DateTimeInput",
"DateTimeInputV09",
"DeleteSurface",
"DeleteSurfaceV09",
"Divider",
"DividerV09",
"Icon",
"IconV09",
"Image",
"ImageV09",
"List",
"ListV09",
"Modal",
"ModalV09",
"MultipleChoice",
"Row",
"RowV09",
"Slider",
"SliderV09",
"SurfaceUpdate",
"Tabs",
"TabsV09",
"Text",
"TextField",
"TextFieldV09",
"TextV09",
"Theme",
"UpdateComponents",
"UpdateDataModel",
"UserAction",
"Video",
"VideoV09",
"validate_a2ui_event",
"validate_a2ui_event_v09",
"validate_a2ui_message",
"validate_a2ui_message_v09",
"validate_catalog_components",
"validate_catalog_components_v09",
]

257
lib/crewai/src/crewai/a2a/extensions/a2ui/client_extension.py
View File

@@ -4,18 +4,27 @@ from __future__ import annotations
from collections.abc import Sequence
import logging
from typing import TYPE_CHECKING, Any, cast
from typing import TYPE_CHECKING, Any, Literal, cast
from pydantic import Field
from pydantic.dataclasses import dataclass
from typing_extensions import TypedDict
from typing_extensions import TypeIs, TypedDict
from crewai.a2a.extensions.a2ui.models import extract_a2ui_json_objects
from crewai.a2a.extensions.a2ui.prompt import build_a2ui_system_prompt
from crewai.a2a.extensions.a2ui.server_extension import A2UI_MIME_TYPE
from crewai.a2a.extensions.a2ui.prompt import (
build_a2ui_system_prompt,
build_a2ui_v09_system_prompt,
)
from crewai.a2a.extensions.a2ui.server_extension import (
A2UI_MIME_TYPE,
A2UI_STANDARD_CATALOG_ID,
A2UI_V09_BASIC_CATALOG_ID,
)
from crewai.a2a.extensions.a2ui.v0_9 import extract_a2ui_v09_json_objects
from crewai.a2a.extensions.a2ui.validator import (
A2UIValidationError,
validate_a2ui_message,
validate_a2ui_message_v09,
)
@@ -84,7 +93,7 @@ class DeleteSurfaceDict(TypedDict):
class A2UIMessageDict(TypedDict, total=False):
"""Serialized A2UI server-to-client message with exactly one key set."""
"""Serialized A2UI v0.8 server-to-client message with exactly one key set."""
beginRendering: BeginRenderingDict
surfaceUpdate: SurfaceUpdateDict
@@ -92,17 +101,79 @@ class A2UIMessageDict(TypedDict, total=False):
deleteSurface: DeleteSurfaceDict
class ThemeDict(TypedDict, total=False):
"""Serialized v0.9 theme."""
primaryColor: str
iconUrl: str
agentDisplayName: str
class CreateSurfaceDict(TypedDict, total=False):
"""Serialized createSurface payload."""
surfaceId: str
catalogId: str
theme: ThemeDict
sendDataModel: bool
class UpdateComponentsDict(TypedDict, total=False):
"""Serialized updateComponents payload."""
surfaceId: str
components: list[dict[str, Any]]
class UpdateDataModelDict(TypedDict, total=False):
"""Serialized updateDataModel payload."""
surfaceId: str
path: str
value: Any
class DeleteSurfaceV09Dict(TypedDict):
"""Serialized v0.9 deleteSurface payload."""
surfaceId: str
class A2UIMessageV09Dict(TypedDict, total=False):
"""Serialized A2UI v0.9 server-to-client message with version and exactly one key set."""
version: Literal["v0.9"]
createSurface: CreateSurfaceDict
updateComponents: UpdateComponentsDict
updateDataModel: UpdateDataModelDict
deleteSurface: DeleteSurfaceV09Dict
A2UIAnyMessageDict = A2UIMessageDict | A2UIMessageV09Dict
def is_v09_message(msg: A2UIAnyMessageDict) -> TypeIs[A2UIMessageV09Dict]:
"""Narrow a message dict to the v0.9 variant."""
return msg.get("version") == "v0.9"
def is_v08_message(msg: A2UIAnyMessageDict) -> TypeIs[A2UIMessageDict]:
"""Narrow a message dict to the v0.8 variant."""
return "version" not in msg
@dataclass
class A2UIConversationState:
"""Tracks active A2UI surfaces and data models across a conversation."""
active_surfaces: dict[str, dict[str, Any]] = Field(default_factory=dict)
data_models: dict[str, list[dict[str, Any]]] = Field(default_factory=dict)
last_a2ui_messages: list[A2UIMessageDict] = Field(default_factory=list)
last_a2ui_messages: list[A2UIAnyMessageDict] = Field(default_factory=list)
initialized_surfaces: set[str] = Field(default_factory=set)
def is_ready(self) -> bool:
"""Return True when at least one surface is active."""
return bool(self.active_surfaces)
"""Return True when at least one surface has been initialized via beginRendering."""
return bool(self.initialized_surfaces)
class A2UIClientExtension:
@@ -125,15 +196,18 @@ class A2UIClientExtension:
self,
catalog_id: str | None = None,
allowed_components: list[str] | None = None,
version: str = "v0.8",
) -> None:
"""Initialize the A2UI client extension.
Args:
catalog_id: Catalog identifier to use for prompt generation.
allowed_components: Subset of component names to expose to the agent.
version: Protocol version, ``"v0.8"`` or ``"v0.9"``.
"""
self._catalog_id = catalog_id
self._allowed_components = allowed_components
self._version = version
def inject_tools(self, agent: Agent) -> None:
"""No-op — A2UI uses prompt augmentation rather than tool injection."""
@@ -169,17 +243,41 @@ class A2UIClientExtension:
catalog_id = data["beginRendering"].get("catalogId")
if catalog_id and catalog_id != self._catalog_id:
continue
if self._catalog_id and "createSurface" in data:
catalog_id = data["createSurface"].get("catalogId")
if catalog_id and catalog_id != self._catalog_id:
continue
if "deleteSurface" in data:
state.active_surfaces.pop(surface_id, None)
state.data_models.pop(surface_id, None)
state.initialized_surfaces.discard(surface_id)
elif "beginRendering" in data:
state.initialized_surfaces.add(surface_id)
state.active_surfaces[surface_id] = data["beginRendering"]
elif "createSurface" in data:
state.initialized_surfaces.add(surface_id)
state.active_surfaces[surface_id] = data["createSurface"]
elif "surfaceUpdate" in data:
if surface_id not in state.initialized_surfaces:
logger.warning(
"surfaceUpdate for uninitialized surface %s",
surface_id,
)
state.active_surfaces[surface_id] = data["surfaceUpdate"]
elif "updateComponents" in data:
if surface_id not in state.initialized_surfaces:
logger.warning(
"updateComponents for uninitialized surface %s",
surface_id,
)
state.active_surfaces[surface_id] = data["updateComponents"]
elif "dataModelUpdate" in data:
contents = data["dataModelUpdate"].get("contents", [])
state.data_models.setdefault(surface_id, []).extend(contents)
elif "updateDataModel" in data:
update = data["updateDataModel"]
state.data_models.setdefault(surface_id, []).append(update)
if not state.active_surfaces and not state.data_models:
return None
@@ -191,10 +289,16 @@ class A2UIClientExtension:
_conversation_state: A2UIConversationState | None,
) -> str:
"""Append A2UI system prompt instructions to the base prompt."""
a2ui_prompt = build_a2ui_system_prompt(
catalog_id=self._catalog_id,
allowed_components=self._allowed_components,
)
if self._version == "v0.9":
a2ui_prompt = build_a2ui_v09_system_prompt(
catalog_id=self._catalog_id,
allowed_components=self._allowed_components,
)
else:
a2ui_prompt = build_a2ui_system_prompt(
catalog_id=self._catalog_id,
allowed_components=self._allowed_components,
)
return f"{base_prompt}\n\n{a2ui_prompt}"
def process_response(
@@ -211,21 +315,77 @@ class A2UIClientExtension:
text = (
agent_response if isinstance(agent_response, str) else str(agent_response)
)
a2ui_messages = _extract_and_validate(text)
results: list[A2UIAnyMessageDict]
if self._version == "v0.9":
results = list(_extract_and_validate_v09(text))
if self._allowed_components:
allowed = set(self._allowed_components)
results = [
_filter_components_v09(m, allowed)
for m in results
if is_v09_message(m)
]
else:
results = list(_extract_and_validate(text))
if self._allowed_components:
allowed = set(self._allowed_components)
results = [
_filter_components(msg, allowed)
for msg in results
if is_v08_message(msg)
]
if self._allowed_components:
allowed = set(self._allowed_components)
a2ui_messages = [_filter_components(msg, allowed) for msg in a2ui_messages]
if a2ui_messages and conversation_state is not None:
conversation_state.last_a2ui_messages = a2ui_messages
if results and conversation_state is not None:
conversation_state.last_a2ui_messages = results
return agent_response
def prepare_message_metadata(
self,
_conversation_state: A2UIConversationState | None,
) -> dict[str, Any]:
"""Inject a2uiClientCapabilities into outbound A2A message metadata.
Per the A2UI extension spec, clients must declare supported catalog
IDs in every outbound message's metadata. v0.9 nests capabilities
under a ``"v0.9"`` key per ``client_capabilities.json``.
"""
if self._version == "v0.9":
default_catalog = A2UI_V09_BASIC_CATALOG_ID
catalog_ids = [default_catalog]
if self._catalog_id and self._catalog_id != default_catalog:
catalog_ids.append(self._catalog_id)
return {
"a2uiClientCapabilities": {
"v0.9": {
"supportedCatalogIds": catalog_ids,
},
},
}
catalog_ids = [A2UI_STANDARD_CATALOG_ID]
if self._catalog_id and self._catalog_id != A2UI_STANDARD_CATALOG_ID:
catalog_ids.append(self._catalog_id)
return {
"a2uiClientCapabilities": {
"supportedCatalogIds": catalog_ids,
},
}
_ALL_SURFACE_ID_KEYS = (
"beginRendering",
"surfaceUpdate",
"dataModelUpdate",
"deleteSurface",
"createSurface",
"updateComponents",
"updateDataModel",
)
def _get_surface_id(data: dict[str, Any]) -> str | None:
"""Extract surfaceId from any A2UI message type."""
for key in ("beginRendering", "surfaceUpdate", "dataModelUpdate", "deleteSurface"):
"""Extract surfaceId from any A2UI v0.8 or v0.9 message type."""
for key in _ALL_SURFACE_ID_KEYS:
inner = data.get(key)
if isinstance(inner, dict):
sid = inner.get("surfaceId")
@@ -263,8 +423,37 @@ def _filter_components(msg: A2UIMessageDict, allowed: set[str]) -> A2UIMessageDi
return {**msg, "surfaceUpdate": {**surface_update, "components": filtered}}
def _filter_components_v09(
msg: A2UIMessageV09Dict, allowed: set[str]
) -> A2UIMessageV09Dict:
"""Strip v0.9 components whose type is not in *allowed* from updateComponents.
v0.9 components use a flat structure where ``component`` is a type-name string.
"""
update = msg.get("updateComponents")
if not isinstance(update, dict):
return msg
components = update.get("components")
if not isinstance(components, list):
return msg
filtered = []
for entry in components:
comp_type = entry.get("component") if isinstance(entry, dict) else None
if isinstance(comp_type, str) and comp_type not in allowed:
logger.debug("Stripping disallowed v0.9 component type %s", comp_type)
continue
filtered.append(entry)
if len(filtered) == len(components):
return msg
return {**msg, "updateComponents": {**update, "components": filtered}}
def _extract_and_validate(text: str) -> list[A2UIMessageDict]:
"""Extract A2UI JSON objects from text and validate them."""
"""Extract A2UI v0.8 JSON objects from text and validate them."""
return [
dumped
for candidate in extract_a2ui_json_objects(text)
@@ -273,7 +462,7 @@ def _extract_and_validate(text: str) -> list[A2UIMessageDict]:
def _try_validate(candidate: dict[str, Any]) -> A2UIMessageDict | None:
"""Validate a single A2UI candidate, returning None on failure."""
"""Validate a single v0.8 A2UI candidate, returning None on failure."""
try:
msg = validate_a2ui_message(candidate)
except A2UIValidationError:
@@ -283,3 +472,25 @@ def _try_validate(candidate: dict[str, Any]) -> A2UIMessageDict | None:
)
return None
return cast(A2UIMessageDict, msg.model_dump(by_alias=True, exclude_none=True))
def _extract_and_validate_v09(text: str) -> list[A2UIMessageV09Dict]:
"""Extract and validate v0.9 A2UI JSON objects from text."""
return [
dumped
for candidate in extract_a2ui_v09_json_objects(text)
if (dumped := _try_validate_v09(candidate)) is not None
]
def _try_validate_v09(candidate: dict[str, Any]) -> A2UIMessageV09Dict | None:
"""Validate a single v0.9 A2UI candidate, returning None on failure."""
try:
msg = validate_a2ui_message_v09(candidate)
except A2UIValidationError:
logger.debug(
"Skipping invalid A2UI v0.9 candidate in agent output",
exc_info=True,
)
return None
return cast(A2UIMessageV09Dict, msg.model_dump(by_alias=True, exclude_none=True))

6
lib/crewai/src/crewai/a2a/extensions/a2ui/models.py
View File

@@ -42,10 +42,6 @@ class MapEntry(BaseModel):
value_boolean: bool | None = Field(
default=None, alias="valueBoolean", description="Boolean value."
)
value_map: list[MapEntry] | None = Field(
default=None, alias="valueMap", description="Nested map entries."
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
@@ -83,7 +79,7 @@ class Styles(BaseModel):
description="Primary color as a hex string.",
)
model_config = ConfigDict(populate_by_name=True, extra="allow")
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class ComponentEntry(BaseModel):

88
lib/crewai/src/crewai/a2a/extensions/a2ui/prompt.py
View File

@@ -6,18 +6,21 @@ import json
from crewai.a2a.extensions.a2ui.catalog import STANDARD_CATALOG_COMPONENTS
from crewai.a2a.extensions.a2ui.schema import load_schema
from crewai.a2a.extensions.a2ui.server_extension import A2UI_EXTENSION_URI
from crewai.a2a.extensions.a2ui.server_extension import (
A2UI_EXTENSION_URI,
A2UI_V09_BASIC_CATALOG_ID,
)
from crewai.a2a.extensions.a2ui.v0_9 import (
BASIC_CATALOG_COMPONENTS as V09_CATALOG_COMPONENTS,
BASIC_CATALOG_FUNCTIONS,
)
def build_a2ui_system_prompt(
catalog_id: str | None = None,
allowed_components: list[str] | None = None,
) -> str:
"""Build a system prompt fragment instructing the LLM to produce A2UI output.
The prompt describes the A2UI message format, available components, and
data binding rules. It includes the resolved schema so the LLM can
generate structured output.
"""Build a v0.8 system prompt fragment instructing the LLM to produce A2UI output.
Args:
catalog_id: Catalog identifier to reference. Defaults to the
@@ -36,7 +39,9 @@ def build_a2ui_system_prompt(
catalog_label = catalog_id or f"standard ({A2UI_EXTENSION_URI.rsplit('/', 1)[-1]})"
resolved_schema = load_schema("server_to_client_with_standard_catalog")
resolved_schema = load_schema(
"server_to_client_with_standard_catalog", version="v0.8"
)
schema_json = json.dumps(resolved_schema, indent=2)
return f"""\
@@ -74,3 +79,72 @@ optionally dataModelUpdate messages to populate data-bound values.
SCHEMA:
{schema_json}
</A2UI_INSTRUCTIONS>"""
def build_a2ui_v09_system_prompt(
catalog_id: str | None = None,
allowed_components: list[str] | None = None,
) -> str:
"""Build a v0.9 system prompt fragment instructing the LLM to produce A2UI output.
Args:
catalog_id: Catalog identifier to reference. Defaults to the
v0.9 basic catalog.
allowed_components: Subset of component names to expose. When
``None``, all basic catalog components are available.
Returns:
A system prompt string to append to the agent's instructions.
"""
components = sorted(
allowed_components if allowed_components is not None else V09_CATALOG_COMPONENTS
)
catalog_label = catalog_id or A2UI_V09_BASIC_CATALOG_ID
functions = sorted(BASIC_CATALOG_FUNCTIONS)
envelope_schema = load_schema("server_to_client", version="v0.9")
schema_json = json.dumps(envelope_schema, indent=2)
return f"""\
<A2UI_INSTRUCTIONS>
You can generate rich, declarative UI by emitting A2UI v0.9 JSON messages.
Every message MUST include "version": "v0.9".
CATALOG: {catalog_label}
AVAILABLE COMPONENTS: {", ".join(components)}
AVAILABLE FUNCTIONS: {", ".join(functions)}
MESSAGE TYPES (emit exactly ONE per message alongside "version": "v0.9"):
- createSurface: Create a new surface. Requires surfaceId and catalogId. \
Optionally includes theme (primaryColor, iconUrl, agentDisplayName) and \
sendDataModel (boolean).
- updateComponents: Send/update components for a surface. Each component is a flat \
object with "id", "component" (type name string), and type-specific properties at the \
top level. One component MUST have id "root".
- updateDataModel: Update the data model. Uses "path" (JSON Pointer) and "value" \
(any JSON type). Omit "value" to delete the key at path.
- deleteSurface: Remove a surface by surfaceId.
COMPONENT FORMAT (flat, NOT nested):
{{"id": "myText", "component": "Text", "text": "Hello world", "variant": "h1"}}
{{"id": "myBtn", "component": "Button", "child": "myText", "action": {{"event": \
{{"name": "click"}}}}}}
DATA BINDING:
- Use plain values for literals: "text": "Hello" or "value": 42
- Use {{"path": "/data/model/path"}} to bind to data model values.
- Use {{"call": "functionName", "args": {{...}}}} for client-side functions.
ACTIONS:
- Server event: {{"event": {{"name": "actionName", "context": {{"key": "value"}}}}}}
- Local function: {{"functionCall": {{"call": "openUrl", "args": {{"url": "..."}}}}}}
OUTPUT FORMAT:
Emit each A2UI message as a valid JSON object. When generating UI, first emit a \
createSurface message with the catalogId, then updateComponents messages with \
components (one must have id "root"), and optionally updateDataModel messages.
ENVELOPE SCHEMA:
{schema_json}
</A2UI_INSTRUCTIONS>"""

48
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/__init__.py
View File

@@ -7,7 +7,8 @@ from pathlib import Path
from typing import Any
_SCHEMA_DIR = Path(__file__).parent / "v0_8"
_V08_DIR = Path(__file__).parent / "v0_8"
_V09_DIR = Path(__file__).parent / "v0_9"
_SCHEMA_CACHE: dict[str, dict[str, Any]] = {}
@@ -20,29 +21,54 @@ SCHEMA_NAMES: frozenset[str] = frozenset(
}
)
V09_SCHEMA_NAMES: frozenset[str] = frozenset(
{
"server_to_client",
"client_to_server",
"common_types",
"basic_catalog",
"client_capabilities",
"server_capabilities",
"client_data_model",
}
)
def load_schema(name: str) -> dict[str, Any]:
"""Load a vendored A2UI JSON schema by name.
def load_schema(name: str, *, version: str = "v0.8") -> dict[str, Any]:
"""Load a vendored A2UI JSON schema by name and version.
Args:
name: Schema name without extension (e.g. ``"server_to_client"``).
name: Schema name without extension, e.g. ``"server_to_client"``.
version: Protocol version, ``"v0.8"`` or ``"v0.9"``.
Returns:
Parsed JSON schema dict.
Raises:
ValueError: If the schema name is not recognized.
ValueError: If the schema name or version is not recognized.
FileNotFoundError: If the schema file is missing from the package.
"""
if name not in SCHEMA_NAMES:
raise ValueError(f"Unknown schema {name!r}. Available: {sorted(SCHEMA_NAMES)}")
if version == "v0.8":
valid_names = SCHEMA_NAMES
schema_dir = _V08_DIR
elif version == "v0.9":
valid_names = V09_SCHEMA_NAMES
schema_dir = _V09_DIR
else:
raise ValueError(f"Unknown version {version!r}. Available: v0.8, v0.9")
if name in _SCHEMA_CACHE:
return _SCHEMA_CACHE[name]
if name not in valid_names:
raise ValueError(
f"Unknown schema {name!r} for {version}. Available: {sorted(valid_names)}"
)
path = _SCHEMA_DIR / f"{name}.json"
cache_key = f"{version}/{name}"
if cache_key in _SCHEMA_CACHE:
return _SCHEMA_CACHE[cache_key]
path = schema_dir / f"{name}.json"
with path.open() as f:
schema: dict[str, Any] = json.load(f)
_SCHEMA_CACHE[name] = schema
_SCHEMA_CACHE[cache_key] = schema
return schema

1387
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/basic_catalog.json Normal file
View File

File diff suppressed because it is too large Load Diff

97
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/client_capabilities.json Normal file
View File

@@ -0,0 +1,97 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://a2ui.org/specification/v0_9/client_capabilities.json",
"title": "A2UI Client Capabilities Schema",
"description": "A schema for the a2uiClientCapabilities object, which is sent from the client to the server as part of the A2A metadata to describe the client's UI rendering capabilities.",
"type": "object",
"properties": {
"v0.9": {
"type": "object",
"description": "The capabilities structure for version 0.9 of the A2UI protocol.",
"properties": {
"supportedCatalogIds": {
"type": "array",
"description": "The URI of each of the component and function catalogs that is supported by the client.",
"items": { "type": "string" }
},
"inlineCatalogs": {
"type": "array",
"description": "An array of inline catalog definitions, which can contain both components and functions. This should only be provided if the agent declares 'acceptsInlineCatalogs: true' in its capabilities.",
"items": { "$ref": "#/$defs/Catalog" }
}
},
"required": ["supportedCatalogIds"]
}
},
"required": ["v0.9"],
"$defs": {
"FunctionDefinition": {
"type": "object",
"description": "Describes a function's interface.",
"properties": {
"name": {
"type": "string",
"description": "The unique name of the function."
},
"description": {
"type": "string",
"description": "A human-readable description of what the function does and how to use it."
},
"parameters": {
"type": "object",
"description": "A JSON Schema describing the expected arguments (args) for this function.",
"$ref": "https://json-schema.org/draft/2020-12/schema"
},
"returnType": {
"type": "string",
"enum": [
"string",
"number",
"boolean",
"array",
"object",
"any",
"void"
],
"description": "The type of value this function returns."
}
},
"required": ["name", "parameters", "returnType"],
"additionalProperties": false
},
"Catalog": {
"type": "object",
"description": "A collection of component and function definitions.",
"properties": {
"catalogId": {
"type": "string",
"description": "Unique identifier for this catalog."
},
"components": {
"type": "object",
"description": "Definitions for UI components supported by this catalog.",
"additionalProperties": {
"$ref": "https://json-schema.org/draft/2020-12/schema"
}
},
"functions": {
"type": "array",
"description": "Definitions for functions supported by this catalog.",
"items": {
"$ref": "#/$defs/FunctionDefinition"
}
},
"theme": {
"title": "A2UI Theme",
"description": "A schema that defines a catalog of A2UI theme properties. Each key is a theme property name (e.g. 'primaryColor'), and each value is the JSON schema for that property.",
"type": "object",
"additionalProperties": {
"$ref": "https://json-schema.org/draft/2020-12/schema"
}
}
},
"required": ["catalogId"],
"additionalProperties": false
}
}
}

22
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/client_data_model.json Normal file
View File

@@ -0,0 +1,22 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://a2ui.org/specification/v0_9/client_data_model.json",
"title": "A2UI Client Data Model Schema",
"description": "Schema for attaching the client data model to A2A message metadata. This object should be placed in the `a2uiClientDataModel` field of the metadata.",
"type": "object",
"properties": {
"version": {
"const": "v0.9"
},
"surfaces": {
"type": "object",
"description": "A map of surface IDs to their current data models.",
"additionalProperties": {
"type": "object",
"description": "The current data model for the surface, as a standard JSON object."
}
}
},
"required": ["version", "surfaces"],
"additionalProperties": false
}

104
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/client_to_server.json Normal file
View File

@@ -0,0 +1,104 @@
{
"title": "A2UI (Agent to UI) Client-to-Server Event Schema",
"description": "Describes a JSON payload for a client-to-server event message.",
"type": "object",
"minProperties": 2,
"maxProperties": 2,
"properties": {
"version": {
"const": "v0.9"
},
"action": {
"type": "object",
"description": "Reports a user-initiated action from a component.",
"properties": {
"name": {
"type": "string",
"description": "The name of the action, taken from the component's action.event.name property."
},
"surfaceId": {
"type": "string",
"description": "The id of the surface where the event originated."
},
"sourceComponentId": {
"type": "string",
"description": "The id of the component that triggered the event."
},
"timestamp": {
"type": "string",
"format": "date-time",
"description": "An ISO 8601 timestamp of when the event occurred."
},
"context": {
"type": "object",
"description": "A JSON object containing the key-value pairs from the component's action.event.context, after resolving all data bindings.",
"additionalProperties": true
}
},
"required": [
"name",
"surfaceId",
"sourceComponentId",
"timestamp",
"context"
]
},
"error": {
"description": "Reports a client-side error.",
"oneOf": [
{
"type": "object",
"title": "Validation Failed Error",
"properties": {
"code": {
"const": "VALIDATION_FAILED"
},
"surfaceId": {
"type": "string",
"description": "The id of the surface where the error occurred."
},
"path": {
"type": "string",
"description": "The JSON pointer to the field that failed validation (e.g. '/components/0/text')."
},
"message": {
"type": "string",
"description": "A short one or two sentence description of why validation failed."
}
},
"required": ["code", "path", "message", "surfaceId"],
"additionalProperties": false
},
{
"type": "object",
"title": "Generic Error",
"properties": {
"code": {
"not": {
"const": "VALIDATION_FAILED"
}
},
"message": {
"type": "string",
"description": "A short one or two sentence description of why the error occurred."
},
"surfaceId": {
"type": "string",
"description": "The id of the surface where the error occurred."
}
},
"required": ["code", "surfaceId", "message"],
"additionalProperties": true
}
]
}
},
"oneOf": [
{
"required": ["action", "version"]
},
{
"required": ["error", "version"]
}
]
}

315
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/common_types.json Normal file
View File

@@ -0,0 +1,315 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://a2ui.org/specification/v0_9/common_types.json",
"title": "A2UI Common Types",
"description": "Common type definitions used across A2UI schemas.",
"$defs": {
"ComponentId": {
"type": "string",
"description": "The unique identifier for a component, used for both definitions and references within the same surface."
},
"AccessibilityAttributes": {
"type": "object",
"description": "Attributes to enhance accessibility when using assistive technologies like screen readers.",
"properties": {
"label": {
"$ref": "#/$defs/DynamicString",
"description": "A short string, typically 1 to 3 words, used by assistive technologies to convey the purpose or intent of an element. For example, an input field might have an accessible label of 'User ID' or a button might be labeled 'Submit'."
},
"description": {
"$ref": "#/$defs/DynamicString",
"description": "Additional information provided by assistive technologies about an element such as instructions, format requirements, or result of an action. For example, a mute button might have a label of 'Mute' and a description of 'Silences notifications about this conversation'."
}
}
},
"ComponentCommon": {
"type": "object",
"properties": {
"id": {
"$ref": "#/$defs/ComponentId"
},
"accessibility": {
"$ref": "#/$defs/AccessibilityAttributes"
}
},
"required": ["id"]
},
"ChildList": {
"oneOf": [
{
"type": "array",
"items": {
"$ref": "#/$defs/ComponentId"
},
"description": "A static list of child component IDs."
},
{
"type": "object",
"description": "A template for generating a dynamic list of children from a data model list. The `componentId` is the component to use as a template.",
"properties": {
"componentId": {
"$ref": "#/$defs/ComponentId"
},
"path": {
"type": "string",
"description": "The path to the list of component property objects in the data model."
}
},
"required": ["componentId", "path"],
"additionalProperties": false
}
]
},
"DataBinding": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "A JSON Pointer path to a value in the data model."
}
},
"required": ["path"],
"additionalProperties": false
},
"DynamicValue": {
"description": "A value that can be a literal, a path, or a function call returning any type.",
"oneOf": [
{
"type": "string"
},
{
"type": "number"
},
{
"type": "boolean"
},
{
"type": "array"
},
{
"$ref": "#/$defs/DataBinding"
},
{
"$ref": "#/$defs/FunctionCall"
}
]
},
"DynamicString": {
"description": "Represents a string",
"oneOf": [
{
"type": "string"
},
{
"$ref": "#/$defs/DataBinding"
},
{
"allOf": [
{
"$ref": "#/$defs/FunctionCall"
},
{
"properties": {
"returnType": {
"const": "string"
}
}
}
]
}
]
},
"DynamicNumber": {
"description": "Represents a value that can be either a literal number, a path to a number in the data model, or a function call returning a number.",
"oneOf": [
{
"type": "number"
},
{
"$ref": "#/$defs/DataBinding"
},
{
"allOf": [
{
"$ref": "#/$defs/FunctionCall"
},
{
"properties": {
"returnType": {
"const": "number"
}
}
}
]
}
]
},
"DynamicBoolean": {
"description": "A boolean value that can be a literal, a path, or a function call returning a boolean.",
"oneOf": [
{
"type": "boolean"
},
{
"$ref": "#/$defs/DataBinding"
},
{
"allOf": [
{
"$ref": "#/$defs/FunctionCall"
},
{
"properties": {
"returnType": {
"const": "boolean"
}
}
}
]
}
]
},
"DynamicStringList": {
"description": "Represents a value that can be either a literal array of strings, a path to a string array in the data model, or a function call returning a string array.",
"oneOf": [
{
"type": "array",
"items": {
"type": "string"
}
},
{
"$ref": "#/$defs/DataBinding"
},
{
"allOf": [
{
"$ref": "#/$defs/FunctionCall"
},
{
"properties": {
"returnType": {
"const": "array"
}
}
}
]
}
]
},
"FunctionCall": {
"type": "object",
"description": "Invokes a named function on the client.",
"properties": {
"call": {
"type": "string",
"description": "The name of the function to call."
},
"args": {
"type": "object",
"description": "Arguments passed to the function.",
"additionalProperties": {
"anyOf": [
{
"$ref": "#/$defs/DynamicValue"
},
{
"type": "object",
"description": "A literal object argument (e.g. configuration)."
}
]
}
},
"returnType": {
"type": "string",
"description": "The expected return type of the function call.",
"enum": [
"string",
"number",
"boolean",
"array",
"object",
"any",
"void"
],
"default": "boolean"
}
},
"required": ["call"],
"oneOf": [
{ "$ref": "catalog.json#/$defs/anyFunction" }
]
},
"CheckRule": {
"type": "object",
"description": "A single validation rule applied to an input component.",
"properties": {
"condition": {
"$ref": "#/$defs/DynamicBoolean"
},
"message": {
"type": "string",
"description": "The error message to display if the check fails."
}
},
"required": ["condition", "message"],
"additionalProperties": false
},
"Checkable": {
"description": "Properties for components that support client-side checks.",
"type": "object",
"properties": {
"checks": {
"type": "array",
"description": "A list of checks to perform. These are function calls that must return a boolean indicating validity.",
"items": {
"$ref": "#/$defs/CheckRule"
}
}
}
},
"Action": {
"description": "Defines an interaction handler that can either trigger a server-side event or execute a local client-side function.",
"oneOf": [
{
"type": "object",
"description": "Triggers a server-side event.",
"properties": {
"event": {
"type": "object",
"description": "The event to dispatch to the server.",
"properties": {
"name": {
"type": "string",
"description": "The name of the action to be dispatched to the server."
},
"context": {
"type": "object",
"description": "A JSON object containing the key-value pairs for the action context. Values can be literals or paths. Use literal values unless the value must be dynamically bound to the data model. Do NOT use paths for static IDs.",
"additionalProperties": {
"$ref": "#/$defs/DynamicValue"
}
}
},
"required": ["name"],
"additionalProperties": false
}
},
"required": ["event"],
"additionalProperties": false
},
{
"type": "object",
"description": "Executes a local client-side function.",
"properties": {
"functionCall": {
"$ref": "#/$defs/FunctionCall"
}
},
"required": ["functionCall"],
"additionalProperties": false
}
]
}
}
}

26
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/server_capabilities.json Normal file
View File

@@ -0,0 +1,26 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://a2ui.org/specification/v0_9/server_capabilities.json",
"title": "A2UI Server Capabilities Schema",
"description": "A schema for the server capabilities object, which is used by an A2UI server (or Agent) to advertise its supported UI features to clients. This can be embedded in an Agent Card for A2A or used in other transport protocols like MCP.",
"type": "object",
"properties": {
"v0.9": {
"type": "object",
"description": "The server capabilities structure for version 0.9 of the A2UI protocol.",
"properties": {
"supportedCatalogIds": {
"type": "array",
"description": "An array of strings, where each string is an ID identifying a Catalog Definition Schema that the server can generate. This is not necessarily a resolvable URI.",
"items": { "type": "string" }
},
"acceptsInlineCatalogs": {
"type": "boolean",
"description": "A boolean indicating if the server can accept an 'inlineCatalogs' array in the client's a2uiClientCapabilities. If omitted, this defaults to false.",
"default": false
}
}
}
},
"required": ["v0.9"]
}

132
lib/crewai/src/crewai/a2a/extensions/a2ui/schema/v0_9/server_to_client.json Normal file
View File

@@ -0,0 +1,132 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://a2ui.org/specification/v0_9/server_to_client.json",
"title": "A2UI Message Schema",
"description": "Describes a JSON payload for an A2UI (Agent to UI) message, which is used to dynamically construct and update user interfaces.",
"type": "object",
"oneOf": [
{ "$ref": "#/$defs/CreateSurfaceMessage" },
{ "$ref": "#/$defs/UpdateComponentsMessage" },
{ "$ref": "#/$defs/UpdateDataModelMessage" },
{ "$ref": "#/$defs/DeleteSurfaceMessage" }
],
"$defs": {
"CreateSurfaceMessage": {
"type": "object",
"properties": {
"version": {
"const": "v0.9"
},
"createSurface": {
"type": "object",
"description": "Signals the client to create a new surface and begin rendering it. When this message is sent, the client will expect 'updateComponents' and/or 'updateDataModel' messages for the same surfaceId that define the component tree.",
"properties": {
"surfaceId": {
"type": "string",
"description": "The unique identifier for the UI surface to be rendered."
},
"catalogId": {
"description": "A string that uniquely identifies this catalog. It is recommended to prefix this with an internet domain that you own, to avoid conflicts e.g. mycompany.com:somecatalog'.",
"type": "string"
},
"theme": {
"$ref": "catalog.json#/$defs/theme",
"description": "Theme parameters for the surface (e.g., {'primaryColor': '#FF0000'}). These must validate against the 'theme' schema defined in the catalog."
},
"sendDataModel": {
"type": "boolean",
"description": "If true, the client will send the full data model of this surface in the metadata of every A2A message sent to the server that created the surface. Defaults to false."
}
},
"required": ["surfaceId", "catalogId"],
"additionalProperties": false
}
},
"required": ["createSurface", "version"],
"additionalProperties": false
},
"UpdateComponentsMessage": {
"type": "object",
"properties": {
"version": {
"const": "v0.9"
},
"updateComponents": {
"type": "object",
"description": "Updates a surface with a new set of components. This message can be sent multiple times to update the component tree of an existing surface. One of the components in one of the components lists MUST have an 'id' of 'root' to serve as the root of the component tree. The createSurface message MUST have been previously sent with the 'catalogId' that is in this message.",
"properties": {
"surfaceId": {
"type": "string",
"description": "The unique identifier for the UI surface to be updated."
},
"components": {
"type": "array",
"description": "A list containing all UI components for the surface.",
"minItems": 1,
"items": {
"$ref": "catalog.json#/$defs/anyComponent"
}
}
},
"required": ["surfaceId", "components"],
"additionalProperties": false
}
},
"required": ["updateComponents", "version"],
"additionalProperties": false
},
"UpdateDataModelMessage": {
"type": "object",
"properties": {
"version": {
"const": "v0.9"
},
"updateDataModel": {
"type": "object",
"description": "Updates the data model for an existing surface. This message can be sent multiple times to update the data model. The createSurface message MUST have been previously sent with the 'catalogId' that is in this message.",
"properties": {
"surfaceId": {
"type": "string",
"description": "The unique identifier for the UI surface this data model update applies to."
},
"path": {
"type": "string",
"description": "An optional path to a location within the data model (e.g., '/user/name'). If omitted, or set to '/', refers to the entire data model."
},
"value": {
"description": "The data to be updated in the data model. If present, the value at 'path' is replaced (or created). If omitted, the key at 'path' is removed.",
"additionalProperties": true
}
},
"required": ["surfaceId"],
"additionalProperties": false
}
},
"required": ["updateDataModel", "version"],
"additionalProperties": false
},
"DeleteSurfaceMessage": {
"type": "object",
"properties": {
"version": {
"const": "v0.9"
},
"deleteSurface": {
"type": "object",
"description": "Signals the client to delete the surface identified by 'surfaceId'. The createSurface message MUST have been previously sent with the 'catalogId' that is in this message.",
"properties": {
"surfaceId": {
"type": "string",
"description": "The unique identifier for the UI surface to be deleted."
}
},
"required": ["surfaceId"],
"additionalProperties": false
}
},
"required": ["deleteSurface", "version"],
"additionalProperties": false
}
}
}

51
lib/crewai/src/crewai/a2a/extensions/a2ui/server_extension.py
View File

@@ -6,9 +6,13 @@ import logging
from typing import Any
from crewai.a2a.extensions.a2ui.models import A2UIResponse, extract_a2ui_json_objects
from crewai.a2a.extensions.a2ui.v0_9 import (
extract_a2ui_v09_json_objects,
)
from crewai.a2a.extensions.a2ui.validator import (
A2UIValidationError,
validate_a2ui_message,
validate_a2ui_message_v09,
)
from crewai.a2a.extensions.server import ExtensionContext, ServerExtension
@@ -17,13 +21,18 @@ logger = logging.getLogger(__name__)
A2UI_MIME_TYPE = "application/json+a2ui"
A2UI_EXTENSION_URI = "https://a2ui.org/a2a-extension/a2ui/v0.8"
A2UI_STANDARD_CATALOG_ID = (
"https://a2ui.org/specification/v0_8/standard_catalog_definition.json"
)
A2UI_V09_EXTENSION_URI = "https://a2ui.org/a2a-extension/a2ui/v0.9"
A2UI_V09_BASIC_CATALOG_ID = "https://a2ui.org/specification/v0_9/basic_catalog.json"
class A2UIServerExtension(ServerExtension):
"""A2A server extension that enables A2UI declarative UI generation.
When activated by a client that declares A2UI v0.8 support,
this extension:
Supports both v0.8 and v0.9 of the A2UI protocol via the ``version``
parameter. When activated by a client, this extension:
* Negotiates catalog preferences during ``on_request``.
* Wraps A2UI messages in the agent response as A2A DataParts with
@@ -31,10 +40,9 @@ class A2UIServerExtension(ServerExtension):
Example::
A2AServerConfig(
server_extensions=[A2UIServerExtension()],
A2AServerConfig
server_extensions=[A2UIServerExtension],
default_output_modes=["text/plain", "application/json+a2ui"],
)
"""
uri: str = A2UI_EXTENSION_URI
@@ -45,15 +53,20 @@ class A2UIServerExtension(ServerExtension):
self,
catalog_ids: list[str] | None = None,
accept_inline_catalogs: bool = False,
version: str = "v0.8",
) -> None:
"""Initialize the A2UI server extension.
Args:
catalog_ids: Catalog identifiers this server supports.
accept_inline_catalogs: Whether inline catalog definitions are accepted.
version: Protocol version, ``"v0.8"`` or ``"v0.9"``.
"""
self._catalog_ids = catalog_ids or []
self._accept_inline_catalogs = accept_inline_catalogs
self._version = version
if version == "v0.9":
self.uri = A2UI_V09_EXTENSION_URI
@property
def params(self) -> dict[str, Any]:
@@ -86,6 +99,7 @@ class A2UIServerExtension(ServerExtension):
Scans the result for A2UI JSON payloads and converts them into
DataParts with ``application/json+a2ui`` MIME type and A2UI metadata.
Dispatches to the correct extractor and validator based on version.
"""
if not context.state.get("a2ui_active"):
return result
@@ -93,13 +107,18 @@ class A2UIServerExtension(ServerExtension):
if not isinstance(result, str):
return result
a2ui_messages = extract_a2ui_json_objects(result)
if self._version == "v0.9":
a2ui_messages = extract_a2ui_v09_json_objects(result)
else:
a2ui_messages = extract_a2ui_json_objects(result)
if not a2ui_messages:
return result
build_fn = _build_data_part_v09 if self._version == "v0.9" else _build_data_part
data_parts = [
part
for part in (_build_data_part(msg_data) for msg_data in a2ui_messages)
for part in (build_fn(msg_data) for msg_data in a2ui_messages)
if part is not None
]
@@ -110,7 +129,7 @@ class A2UIServerExtension(ServerExtension):
def _build_data_part(msg_data: dict[str, Any]) -> dict[str, Any] | None:
"""Validate a single A2UI message and wrap it as a DataPart dict."""
"""Validate a v0.8 A2UI message and wrap it as a DataPart dict."""
try:
validated = validate_a2ui_message(msg_data)
except A2UIValidationError:
@@ -123,3 +142,19 @@ def _build_data_part(msg_data: dict[str, Any]) -> dict[str, Any] | None:
"mimeType": A2UI_MIME_TYPE,
},
}
def _build_data_part_v09(msg_data: dict[str, Any]) -> dict[str, Any] | None:
"""Validate a v0.9 A2UI message and wrap it as a DataPart dict."""
try:
validated = validate_a2ui_message_v09(msg_data)
except A2UIValidationError:
logger.warning("Skipping invalid A2UI v0.9 message in response", exc_info=True)
return None
return {
"kind": "data",
"data": validated.model_dump(by_alias=True, exclude_none=True),
"metadata": {
"mimeType": A2UI_MIME_TYPE,
},
}

831
lib/crewai/src/crewai/a2a/extensions/a2ui/v0_9.py Normal file
View File

@@ -0,0 +1,831 @@
"""Pydantic models for A2UI v0.9 protocol messages and types.
This module provides v0.9 counterparts to the v0.8 models in ``models.py``.
Key differences from v0.8:
* ``beginRendering`` → ``createSurface`` — adds ``theme``, ``sendDataModel``,
requires ``catalogId``.
* ``surfaceUpdate`` → ``updateComponents`` — component structure is flat:
``component`` is a type-name string, properties live at the top level.
* ``dataModelUpdate`` → ``updateDataModel`` — ``contents`` adjacency list
replaced by a single ``value`` of any JSON type; ``path`` uses JSON Pointers.
* All messages carry a ``version: "v0.9"`` discriminator.
* Data binding uses plain JSON values, ``DataBinding`` objects, or
``FunctionCall`` objects instead of ``literalString`` / ``path`` wrappers.
* ``MultipleChoice`` is replaced by ``ChoicePicker``.
* ``Styles`` is replaced by ``Theme`` — adds ``iconUrl``, ``agentDisplayName``.
* Client-to-server ``userAction`` is renamed to ``action``; ``error`` gains
structured ``code`` / ``path`` fields.
"""
from __future__ import annotations
import json
from typing import Any, Literal, get_args
from pydantic import BaseModel, ConfigDict, Field, model_validator
ComponentName = Literal[
"Text",
"Image",
"Icon",
"Video",
"AudioPlayer",
"Row",
"Column",
"List",
"Card",
"Tabs",
"Modal",
"Divider",
"Button",
"TextField",
"CheckBox",
"ChoicePicker",
"Slider",
"DateTimeInput",
]
BASIC_CATALOG_COMPONENTS: frozenset[ComponentName] = frozenset(get_args(ComponentName))
FunctionName = Literal[
"required",
"regex",
"length",
"numeric",
"email",
"formatString",
"formatNumber",
"formatCurrency",
"formatDate",
"pluralize",
"openUrl",
"and",
"or",
"not",
]
BASIC_CATALOG_FUNCTIONS: frozenset[FunctionName] = frozenset(get_args(FunctionName))
IconNameV09 = Literal[
"accountCircle",
"add",
"arrowBack",
"arrowForward",
"attachFile",
"calendarToday",
"call",
"camera",
"check",
"close",
"delete",
"download",
"edit",
"event",
"error",
"fastForward",
"favorite",
"favoriteOff",
"folder",
"help",
"home",
"info",
"locationOn",
"lock",
"lockOpen",
"mail",
"menu",
"moreVert",
"moreHoriz",
"notificationsOff",
"notifications",
"pause",
"payment",
"person",
"phone",
"photo",
"play",
"print",
"refresh",
"rewind",
"search",
"send",
"settings",
"share",
"shoppingCart",
"skipNext",
"skipPrevious",
"star",
"starHalf",
"starOff",
"stop",
"upload",
"visibility",
"visibilityOff",
"volumeDown",
"volumeMute",
"volumeOff",
"volumeUp",
"warning",
]
V09_ICON_NAMES: frozenset[IconNameV09] = frozenset(get_args(IconNameV09))
class DataBinding(BaseModel):
"""JSON Pointer path reference to the data model."""
path: str = Field(description="A JSON Pointer path to a value in the data model.")
model_config = ConfigDict(extra="forbid")
class FunctionCall(BaseModel):
"""Client-side function invocation."""
call: str = Field(description="The name of the function to call.")
args: dict[str, DynamicValue] | None = Field(
default=None, description="Arguments passed to the function."
)
return_type: (
Literal["string", "number", "boolean", "array", "object", "any", "void"] | None
) = Field(
default=None,
alias="returnType",
description="Expected return type of the function call.",
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
DynamicValue = str | float | int | bool | list[Any] | DataBinding | FunctionCall
DynamicString = str | DataBinding | FunctionCall
DynamicNumber = float | int | DataBinding | FunctionCall
DynamicBoolean = bool | DataBinding | FunctionCall
DynamicStringList = list[str] | DataBinding | FunctionCall
class CheckRule(BaseModel):
"""A single validation rule for an input component."""
condition: DynamicBoolean = Field(
description="Condition that must evaluate to true for the check to pass."
)
message: str = Field(description="Error message displayed if the check fails.")
model_config = ConfigDict(extra="forbid")
class AccessibilityAttributes(BaseModel):
"""Accessibility attributes for assistive technologies."""
label: DynamicString | None = Field(
default=None, description="Short label for screen readers."
)
description: DynamicString | None = Field(
default=None, description="Extended description for screen readers."
)
model_config = ConfigDict(extra="forbid")
class ChildTemplate(BaseModel):
"""Template for generating dynamic children from a data model list."""
component_id: str = Field(
alias="componentId", description="Component to repeat per list item."
)
path: str = Field(description="Data model path to the list of items.")
model_config = ConfigDict(populate_by_name=True, extra="forbid")
ChildListV09 = list[str] | ChildTemplate
class EventAction(BaseModel):
"""Server-side event triggered by a component interaction."""
name: str = Field(description="Action name dispatched to the server.")
context: dict[str, DynamicValue] | None = Field(
default=None, description="Key-value pairs sent with the event."
)
model_config = ConfigDict(extra="forbid")
class ActionV09(BaseModel):
"""Interaction handler: server event or local function call.
Exactly one of ``event`` or ``function_call`` must be set.
"""
event: EventAction | None = Field(
default=None, description="Triggers a server-side event."
)
function_call: FunctionCall | None = Field(
default=None,
alias="functionCall",
description="Executes a local client-side function.",
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
@model_validator(mode="after")
def _check_exactly_one(self) -> ActionV09:
"""Enforce exactly one of event or functionCall."""
count = sum(f is not None for f in (self.event, self.function_call))
if count != 1:
raise ValueError(
f"Exactly one of event or functionCall must be set, got {count}"
)
return self
class TextV09(BaseModel):
"""Displays text content."""
id: str = Field(description="Unique component identifier.")
component: Literal["Text"] = "Text"
text: DynamicString = Field(description="Text content to display.")
variant: Literal["h1", "h2", "h3", "h4", "h5", "caption", "body"] | None = Field(
default=None, description="Semantic text style hint."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ImageV09(BaseModel):
"""Displays an image."""
id: str = Field(description="Unique component identifier.")
component: Literal["Image"] = "Image"
url: DynamicString = Field(description="Image source URL.")
description: DynamicString | None = Field(
default=None, description="Accessibility text."
)
fit: Literal["contain", "cover", "fill", "none", "scaleDown"] | None = Field(
default=None, description="Object-fit behavior."
)
variant: (
Literal[
"icon", "avatar", "smallFeature", "mediumFeature", "largeFeature", "header"
]
| None
) = Field(default=None, description="Image size hint.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class IconV09(BaseModel):
"""Displays a named icon."""
id: str = Field(description="Unique component identifier.")
component: Literal["Icon"] = "Icon"
name: IconNameV09 | DataBinding = Field(description="Icon name or data binding.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class VideoV09(BaseModel):
"""Displays a video player."""
id: str = Field(description="Unique component identifier.")
component: Literal["Video"] = "Video"
url: DynamicString = Field(description="Video source URL.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class AudioPlayerV09(BaseModel):
"""Displays an audio player."""
id: str = Field(description="Unique component identifier.")
component: Literal["AudioPlayer"] = "AudioPlayer"
url: DynamicString = Field(description="Audio source URL.")
description: DynamicString | None = Field(
default=None, description="Audio content description."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class RowV09(BaseModel):
"""Horizontal layout container."""
id: str = Field(description="Unique component identifier.")
component: Literal["Row"] = "Row"
children: ChildListV09 = Field(description="Child components.")
justify: (
Literal[
"center",
"end",
"spaceAround",
"spaceBetween",
"spaceEvenly",
"start",
"stretch",
]
| None
) = Field(default=None, description="Main-axis distribution.")
align: Literal["start", "center", "end", "stretch"] | None = Field(
default=None, description="Cross-axis alignment."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ColumnV09(BaseModel):
"""Vertical layout container."""
id: str = Field(description="Unique component identifier.")
component: Literal["Column"] = "Column"
children: ChildListV09 = Field(description="Child components.")
justify: (
Literal[
"start",
"center",
"end",
"spaceBetween",
"spaceAround",
"spaceEvenly",
"stretch",
]
| None
) = Field(default=None, description="Main-axis distribution.")
align: Literal["center", "end", "start", "stretch"] | None = Field(
default=None, description="Cross-axis alignment."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ListV09(BaseModel):
"""Scrollable list container."""
id: str = Field(description="Unique component identifier.")
component: Literal["List"] = "List"
children: ChildListV09 = Field(description="Child components.")
direction: Literal["vertical", "horizontal"] | None = Field(
default=None, description="Scroll direction."
)
align: Literal["start", "center", "end", "stretch"] | None = Field(
default=None, description="Cross-axis alignment."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class CardV09(BaseModel):
"""Card container wrapping a single child."""
id: str = Field(description="Unique component identifier.")
component: Literal["Card"] = "Card"
child: str = Field(description="ID of the child component.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class TabItemV09(BaseModel):
"""A single tab definition."""
title: DynamicString = Field(description="Tab title.")
child: str = Field(description="ID of the tab content component.")
model_config = ConfigDict(extra="forbid")
class TabsV09(BaseModel):
"""Tabbed navigation container."""
id: str = Field(description="Unique component identifier.")
component: Literal["Tabs"] = "Tabs"
tabs: list[TabItemV09] = Field(min_length=1, description="Tab definitions.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ModalV09(BaseModel):
"""Modal dialog with a trigger and content."""
id: str = Field(description="Unique component identifier.")
component: Literal["Modal"] = "Modal"
trigger: str = Field(description="ID of the component that opens the modal.")
content: str = Field(description="ID of the component inside the modal.")
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class DividerV09(BaseModel):
"""Visual divider line."""
id: str = Field(description="Unique component identifier.")
component: Literal["Divider"] = "Divider"
axis: Literal["horizontal", "vertical"] | None = Field(
default=None, description="Divider orientation."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ButtonV09(BaseModel):
"""Interactive button."""
id: str = Field(description="Unique component identifier.")
component: Literal["Button"] = "Button"
child: str = Field(description="ID of the button label component.")
action: ActionV09 = Field(description="Action dispatched on click.")
variant: Literal["default", "primary", "borderless"] | None = Field(
default=None, description="Button style variant."
)
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class TextFieldV09(BaseModel):
"""Text input field."""
id: str = Field(description="Unique component identifier.")
component: Literal["TextField"] = "TextField"
label: DynamicString = Field(description="Input label.")
value: DynamicString | None = Field(default=None, description="Current text value.")
variant: Literal["longText", "number", "shortText", "obscured"] | None = Field(
default=None, description="Input type variant."
)
validation_regexp: str | None = Field(
default=None,
alias="validationRegexp",
description="Regex for client-side validation.",
)
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class CheckBoxV09(BaseModel):
"""Checkbox input."""
id: str = Field(description="Unique component identifier.")
component: Literal["CheckBox"] = "CheckBox"
label: DynamicString = Field(description="Checkbox label.")
value: DynamicBoolean = Field(description="Checked state.")
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class ChoicePickerOption(BaseModel):
"""A single option in a ChoicePicker."""
label: DynamicString = Field(description="Display label.")
value: str = Field(description="Value when selected.")
model_config = ConfigDict(extra="forbid")
class ChoicePickerV09(BaseModel):
"""Selection component replacing v0.8 MultipleChoice."""
id: str = Field(description="Unique component identifier.")
component: Literal["ChoicePicker"] = "ChoicePicker"
options: list[ChoicePickerOption] = Field(description="Available choices.")
value: DynamicStringList = Field(description="Currently selected values.")
label: DynamicString | None = Field(default=None, description="Group label.")
variant: Literal["multipleSelection", "mutuallyExclusive"] | None = Field(
default=None, description="Selection behavior."
)
display_style: Literal["checkbox", "chips"] | None = Field(
default=None, alias="displayStyle", description="Visual display style."
)
filterable: bool | None = Field(
default=None, description="Whether options can be filtered."
)
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class SliderV09(BaseModel):
"""Numeric slider input."""
id: str = Field(description="Unique component identifier.")
component: Literal["Slider"] = "Slider"
value: DynamicNumber = Field(description="Current slider value.")
max: float = Field(description="Maximum slider value.")
min: float | None = Field(default=None, description="Minimum slider value.")
label: DynamicString | None = Field(default=None, description="Slider label.")
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(extra="forbid")
class DateTimeInputV09(BaseModel):
"""Date and/or time picker."""
id: str = Field(description="Unique component identifier.")
component: Literal["DateTimeInput"] = "DateTimeInput"
value: DynamicString = Field(description="ISO 8601 date/time value.")
enable_date: bool | None = Field(
default=None, alias="enableDate", description="Enable date selection."
)
enable_time: bool | None = Field(
default=None, alias="enableTime", description="Enable time selection."
)
min: DynamicString | None = Field(
default=None, description="Minimum allowed date/time."
)
max: DynamicString | None = Field(
default=None, description="Maximum allowed date/time."
)
label: DynamicString | None = Field(default=None, description="Input label.")
checks: list[CheckRule] | None = Field(
default=None, description="Validation rules."
)
weight: float | None = Field(default=None, description="Flex weight.")
accessibility: AccessibilityAttributes | None = None
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class Theme(BaseModel):
"""Surface theme configuration for v0.9.
Replaces v0.8 ``Styles``. Adds ``iconUrl`` and ``agentDisplayName``
for agent attribution; drops ``font``.
"""
primary_color: str | None = Field(
default=None,
alias="primaryColor",
pattern=r"^#[0-9a-fA-F]{6}$",
description="Primary brand color as a hex string.",
)
icon_url: str | None = Field(
default=None,
alias="iconUrl",
description="URL for an image identifying the agent or tool.",
)
agent_display_name: str | None = Field(
default=None,
alias="agentDisplayName",
description="Text label identifying the agent or tool.",
)
model_config = ConfigDict(populate_by_name=True, extra="allow")
class CreateSurface(BaseModel):
"""Signals the client to create a new surface and begin rendering.
Replaces v0.8 ``BeginRendering``. ``catalogId`` is now required and
``theme`` / ``sendDataModel`` are new.
"""
surface_id: str = Field(alias="surfaceId", description="Unique surface identifier.")
catalog_id: str = Field(
alias="catalogId", description="Catalog identifier for this surface."
)
theme: Theme | None = Field(default=None, description="Theme parameters.")
send_data_model: bool | None = Field(
default=None,
alias="sendDataModel",
description="If true, client sends data model in action metadata.",
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class UpdateComponents(BaseModel):
"""Updates a surface with a new set of components.
Replaces v0.8 ``SurfaceUpdate``. Components use a flat structure where
``component`` is a type-name string and properties sit at the top level.
"""
surface_id: str = Field(alias="surfaceId", description="Target surface identifier.")
components: list[dict[str, Any]] = Field(
min_length=1, description="Components to render on the surface."
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class UpdateDataModel(BaseModel):
"""Updates the data model for a surface.
Replaces v0.8 ``DataModelUpdate``. The ``contents`` adjacency list is
replaced by a single ``value`` of any JSON type. ``path`` uses JSON
Pointer syntax — e.g. ``/user/name``.
"""
surface_id: str = Field(alias="surfaceId", description="Target surface identifier.")
path: str | None = Field(
default=None, description="JSON Pointer path for the update."
)
value: Any = Field(
default=None, description="Value to set. Omit to delete the key."
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class DeleteSurfaceV09(BaseModel):
"""Signals the client to delete a surface."""
surface_id: str = Field(alias="surfaceId", description="Surface to delete.")
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class A2UIMessageV09(BaseModel):
"""Union wrapper for v0.9 server-to-client message types.
Exactly one message field must be set alongside the ``version`` field.
"""
version: Literal["v0.9"] = "v0.9"
create_surface: CreateSurface | None = Field(
default=None, alias="createSurface", description="Create a new surface."
)
update_components: UpdateComponents | None = Field(
default=None,
alias="updateComponents",
description="Update components on a surface.",
)
update_data_model: UpdateDataModel | None = Field(
default=None,
alias="updateDataModel",
description="Update the surface data model.",
)
delete_surface: DeleteSurfaceV09 | None = Field(
default=None, alias="deleteSurface", description="Delete a surface."
)
model_config = ConfigDict(populate_by_name=True, extra="forbid")
@model_validator(mode="after")
def _check_exactly_one(self) -> A2UIMessageV09:
"""Enforce the spec's exactly-one-of constraint."""
fields = [
self.create_surface,
self.update_components,
self.update_data_model,
self.delete_surface,
]
count = sum(f is not None for f in fields)
if count != 1:
raise ValueError(
f"Exactly one A2UI v0.9 message type must be set, got {count}"
)
return self
class ActionEvent(BaseModel):
"""User-initiated action from a component.
Replaces v0.8 ``UserAction``. The event field is renamed from
``userAction`` to ``action``.
"""
name: str = Field(description="Action name.")
surface_id: str = Field(alias="surfaceId", description="Source surface identifier.")
source_component_id: str = Field(
alias="sourceComponentId",
description="Component that triggered the action.",
)
timestamp: str = Field(description="ISO 8601 timestamp of the action.")
context: dict[str, Any] = Field(description="Resolved action context payload.")
model_config = ConfigDict(populate_by_name=True)
class ClientErrorV09(BaseModel):
"""Structured client-side error report.
Replaces v0.8's flexible ``ClientError`` with required ``code``,
``surfaceId``, and ``message`` fields.
"""
code: str = Field(description="Error code (e.g. VALIDATION_FAILED).")
surface_id: str = Field(
alias="surfaceId", description="Surface where the error occurred."
)
message: str = Field(description="Human-readable error description.")
path: str | None = Field(
default=None, description="JSON Pointer to the failing field."
)
model_config = ConfigDict(populate_by_name=True, extra="allow")
class A2UIEventV09(BaseModel):
"""Union wrapper for v0.9 client-to-server events."""
version: Literal["v0.9"] = "v0.9"
action: ActionEvent | None = Field(
default=None, description="User-initiated action event."
)
error: ClientErrorV09 | None = Field(
default=None, description="Client-side error report."
)
model_config = ConfigDict(populate_by_name=True)
@model_validator(mode="after")
def _check_exactly_one(self) -> A2UIEventV09:
"""Enforce the spec's exactly-one-of constraint."""
fields = [self.action, self.error]
count = sum(f is not None for f in fields)
if count != 1:
raise ValueError(
f"Exactly one A2UI v0.9 event type must be set, got {count}"
)
return self
class ClientDataModel(BaseModel):
"""Client data model payload for A2A message metadata.
When ``sendDataModel`` is ``true`` on ``createSurface``, the client
attaches this object to every outbound A2A message as
``a2uiClientDataModel`` in the metadata.
"""
version: Literal["v0.9"] = "v0.9"
surfaces: dict[str, dict[str, Any]] = Field(
description="Map of surface IDs to their current data models."
)
model_config = ConfigDict(extra="forbid")
_V09_KEYS = {"createSurface", "updateComponents", "updateDataModel", "deleteSurface"}
def extract_a2ui_v09_json_objects(text: str) -> list[dict[str, Any]]:
"""Extract JSON objects containing A2UI v0.9 keys from text.
Uses ``json.JSONDecoder.raw_decode`` for robust parsing that correctly
handles braces inside string literals.
"""
decoder = json.JSONDecoder()
results: list[dict[str, Any]] = []
idx = 0
while idx < len(text):
idx = text.find("{", idx)
if idx == -1:
break
try:
obj, end_idx = decoder.raw_decode(text, idx)
if isinstance(obj, dict) and _V09_KEYS & obj.keys():
results.append(obj)
idx = end_idx
except json.JSONDecodeError:
idx += 1
return results

236
lib/crewai/src/crewai/a2a/extensions/a2ui/validator.py
View File

@@ -4,9 +4,73 @@ from __future__ import annotations
from typing import Any
from pydantic import ValidationError
from pydantic import BaseModel, ValidationError
from crewai.a2a.extensions.a2ui.catalog import (
AudioPlayer,
Button,
Card,
CheckBox,
Column,
DateTimeInput,
Divider,
Icon,
Image,
List,
Modal,
MultipleChoice,
Row,
Slider,
Tabs,
Text,
TextField,
Video,
)
from crewai.a2a.extensions.a2ui.models import A2UIEvent, A2UIMessage
from crewai.a2a.extensions.a2ui.v0_9 import (
A2UIEventV09,
A2UIMessageV09,
AudioPlayerV09,
ButtonV09,
CardV09,
CheckBoxV09,
ChoicePickerV09,
ColumnV09,
DateTimeInputV09,
DividerV09,
IconV09,
ImageV09,
ListV09,
ModalV09,
RowV09,
SliderV09,
TabsV09,
TextFieldV09,
TextV09,
VideoV09,
)
_STANDARD_CATALOG_MODELS: dict[str, type[BaseModel]] = {
"AudioPlayer": AudioPlayer,
"Button": Button,
"Card": Card,
"CheckBox": CheckBox,
"Column": Column,
"DateTimeInput": DateTimeInput,
"Divider": Divider,
"Icon": Icon,
"Image": Image,
"List": List,
"Modal": Modal,
"MultipleChoice": MultipleChoice,
"Row": Row,
"Slider": Slider,
"Tabs": Tabs,
"Text": Text,
"TextField": TextField,
"Video": Video,
}
class A2UIValidationError(Exception):
@@ -17,11 +81,17 @@ class A2UIValidationError(Exception):
self.errors = errors or []
def validate_a2ui_message(data: dict[str, Any]) -> A2UIMessage:
def validate_a2ui_message(
data: dict[str, Any],
*,
validate_catalog: bool = False,
) -> A2UIMessage:
"""Parse and validate an A2UI server-to-client message.
Args:
data: Raw message dict (JSON-decoded).
data: Raw JSON-decoded message dict.
validate_catalog: If True, also validate component properties
against the standard catalog.
Returns:
Validated ``A2UIMessage`` instance.
@@ -30,19 +100,24 @@ def validate_a2ui_message(data: dict[str, Any]) -> A2UIMessage:
A2UIValidationError: If the data does not conform to the A2UI schema.
"""
try:
return A2UIMessage.model_validate(data)
message = A2UIMessage.model_validate(data)
except ValidationError as exc:
raise A2UIValidationError(
f"Invalid A2UI message: {exc.error_count()} validation error(s)",
errors=exc.errors(),
) from exc
if validate_catalog:
validate_catalog_components(message)
return message
def validate_a2ui_event(data: dict[str, Any]) -> A2UIEvent:
"""Parse and validate an A2UI client-to-server event.
Args:
data: Raw event dict (JSON-decoded).
data: Raw JSON-decoded event dict.
Returns:
Validated ``A2UIEvent`` instance.
@@ -57,3 +132,154 @@ def validate_a2ui_event(data: dict[str, Any]) -> A2UIEvent:
f"Invalid A2UI event: {exc.error_count()} validation error(s)",
errors=exc.errors(),
) from exc
def validate_a2ui_message_v09(data: dict[str, Any]) -> A2UIMessageV09:
"""Parse and validate an A2UI v0.9 server-to-client message.
Args:
data: Raw JSON-decoded message dict.
Returns:
Validated ``A2UIMessageV09`` instance.
Raises:
A2UIValidationError: If the data does not conform to the v0.9 schema.
"""
try:
return A2UIMessageV09.model_validate(data)
except ValidationError as exc:
raise A2UIValidationError(
f"Invalid A2UI v0.9 message: {exc.error_count()} validation error(s)",
errors=exc.errors(),
) from exc
def validate_a2ui_event_v09(data: dict[str, Any]) -> A2UIEventV09:
"""Parse and validate an A2UI v0.9 client-to-server event.
Args:
data: Raw JSON-decoded event dict.
Returns:
Validated ``A2UIEventV09`` instance.
Raises:
A2UIValidationError: If the data does not conform to the v0.9 schema.
"""
try:
return A2UIEventV09.model_validate(data)
except ValidationError as exc:
raise A2UIValidationError(
f"Invalid A2UI v0.9 event: {exc.error_count()} validation error(s)",
errors=exc.errors(),
) from exc
def validate_catalog_components(message: A2UIMessage) -> None:
"""Validate component properties in a surfaceUpdate against the standard catalog.
Only applies to surfaceUpdate messages. Components whose type is not
in the standard catalog are skipped without error.
Args:
message: A validated A2UIMessage.
Raises:
A2UIValidationError: If any component fails catalog validation.
"""
if message.surface_update is None:
return
errors: list[Any] = []
for entry in message.surface_update.components:
for type_name, props in entry.component.items():
model = _STANDARD_CATALOG_MODELS.get(type_name)
if model is None:
continue
try:
model.model_validate(props)
except ValidationError as exc:
errors.extend(
{
"component_id": entry.id,
"component_type": type_name,
**err,
}
for err in exc.errors()
)
if errors:
raise A2UIValidationError(
f"Catalog validation failed: {len(errors)} error(s)",
errors=errors,
)
_V09_BASIC_CATALOG_MODELS: dict[str, type[BaseModel]] = {
"AudioPlayer": AudioPlayerV09,
"Button": ButtonV09,
"Card": CardV09,
"CheckBox": CheckBoxV09,
"ChoicePicker": ChoicePickerV09,
"Column": ColumnV09,
"DateTimeInput": DateTimeInputV09,
"Divider": DividerV09,
"Icon": IconV09,
"Image": ImageV09,
"List": ListV09,
"Modal": ModalV09,
"Row": RowV09,
"Slider": SliderV09,
"Tabs": TabsV09,
"Text": TextV09,
"TextField": TextFieldV09,
"Video": VideoV09,
}
def validate_catalog_components_v09(message: A2UIMessageV09) -> None:
"""Validate component properties in an updateComponents against the basic catalog.
v0.9 components use a flat structure where ``component`` is a type-name
string and properties sit at the top level of the component dict.
Only applies to updateComponents messages. Components whose type is not
in the basic catalog are skipped without error.
Args:
message: A validated A2UIMessageV09.
Raises:
A2UIValidationError: If any component fails catalog validation.
"""
if message.update_components is None:
return
errors: list[Any] = []
for entry in message.update_components.components:
if not isinstance(entry, dict):
continue
type_name = entry.get("component")
if not isinstance(type_name, str):
continue
model = _V09_BASIC_CATALOG_MODELS.get(type_name)
if model is None:
continue
try:
model.model_validate(entry)
except ValidationError as exc:
errors.extend(
{
"component_id": entry.get("id", "<unknown>"),
"component_type": type_name,
**err,
}
for err in exc.errors()
)
if errors:
raise A2UIValidationError(
f"v0.9 catalog validation failed: {len(errors)} error(s)",
errors=errors,
)

35
lib/crewai/src/crewai/a2a/extensions/base.py
View File

@@ -150,6 +150,23 @@ class A2AExtension(Protocol):
"""
...
def prepare_message_metadata(
self,
conversation_state: ConversationState | None,
) -> dict[str, Any]:
"""Prepare extension-specific metadata for outbound A2A messages.
Called when constructing A2A messages to inject extension-specific
metadata such as client capabilities declarations.
Args:
conversation_state: Extension-specific state from extract_state_from_history.
Returns:
Dict of metadata key-value pairs to merge into the message metadata.
"""
...
class ExtensionRegistry:
"""Registry for managing A2A extensions.
@@ -236,3 +253,21 @@ class ExtensionRegistry:
state = extension_states.get(type(extension))
processed = extension.process_response(processed, state)
return processed
def prepare_all_metadata(
self,
extension_states: dict[type[A2AExtension], ConversationState],
) -> dict[str, Any]:
"""Collect metadata from all registered extensions for outbound messages.
Args:
extension_states: Mapping of extension types to conversation states.
Returns:
Merged metadata dict from all extensions.
"""
metadata: dict[str, Any] = {}
for extension in self._extensions:
state = extension_states.get(type(extension))
metadata.update(extension.prepare_message_metadata(state))
return metadata

22
lib/crewai/src/crewai/a2a/wrapper.py
View File

@@ -1273,6 +1273,15 @@ def _delegate_to_a2a(
for turn_num in range(ctx.max_turns):
agent_branch, accepted_output_modes = _get_turn_context(ctx.agent_config)
merged_metadata = dict(ctx.metadata) if ctx.metadata else {}
if _extension_registry and conversation_history:
_ext_states = _extension_registry.extract_all_states(
conversation_history
)
merged_metadata.update(
_extension_registry.prepare_all_metadata(_ext_states)
)
a2a_result = execute_a2a_delegation(
endpoint=ctx.agent_config.endpoint,
auth=ctx.agent_config.auth,
@@ -1281,7 +1290,7 @@ def _delegate_to_a2a(
context_id=context_id,
task_id=task_id,
reference_task_ids=reference_task_ids,
metadata=ctx.metadata,
metadata=merged_metadata or None,
extensions=ctx.extensions,
conversation_history=conversation_history,
agent_id=ctx.agent_id,
@@ -1619,6 +1628,15 @@ async def _adelegate_to_a2a(
for turn_num in range(ctx.max_turns):
agent_branch, accepted_output_modes = _get_turn_context(ctx.agent_config)
merged_metadata = dict(ctx.metadata) if ctx.metadata else {}
if _extension_registry and conversation_history:
_ext_states = _extension_registry.extract_all_states(
conversation_history
)
merged_metadata.update(
_extension_registry.prepare_all_metadata(_ext_states)
)
a2a_result = await aexecute_a2a_delegation(
endpoint=ctx.agent_config.endpoint,
auth=ctx.agent_config.auth,
@@ -1627,7 +1645,7 @@ async def _adelegate_to_a2a(
context_id=context_id,
task_id=task_id,
reference_task_ids=reference_task_ids,
metadata=ctx.metadata,
metadata=merged_metadata or None,
extensions=ctx.extensions,
conversation_history=conversation_history,
agent_id=ctx.agent_id,