feat(python/adk): wire SendA2uiToClientToolset and converters to A2uiRequestProcessor and RendererState

[nan-yu]

Summary

Connects python/a2ui_adk to the A2uiRequestProcessor, RendererState, and ResponseStream architecture introduced in #3268–#3271:

1. SendA2uiToClientToolset:
   • Accepts structured example turns (Sequence[Sequence[AgentToRendererMessage]], such as processor.examples) or None in addition to markdown strings, fixing the TypeError when passing processor.examples.
   • Avoids duplicating schema and example instructions in process_llm_request() when they are already present in llm_request.config.system_instruction.
   • Accepts an optional a2ui_state (RendererState or A2uiStateProvider) to validate and apply tool payloads against a Canvas RendererState.
2. A2uiPartConverter & parse_content_to_parts:
   • parse_content_to_parts accepts an optional state: RendererState | None.
   • A2uiPartConverter accepts optional processor: A2uiRequestProcessor | None and state: RendererState | None, and infers version from the active catalogs' protocol_version when version is omitted.
3. A2uiEventConverter:
   • Reads system:a2ui_processor, system:a2ui_state, and system:a2ui_stream from session state alongside system:a2ui_catalogs.
   • Reuses A2uiPartConverter across events instead of instantiating a throw-away converter and parser on every ADK event.
   • Supports incremental streaming via ResponseStream when partial ADK events (event.partial == True) arrive.

Test Plan

• uv run pytest python/a2ui_a2a/tests python/a2ui_adk/tests
• uv run mypy python/a2ui_a2a/src python/a2ui_adk/src

[gemini-code-assist]

Code Review

This pull request enhances the A2UI ADK library to support Canvas UI states, structured example turns, and incremental streaming. Key updates include allowing SendA2uiToClientToolset to accept structured example turns and an optional a2ui_state for validation, updating A2uiPartConverter to accept a processor and state, and enabling A2uiEventConverter to support incremental streaming via ResponseStream. Feedback on these changes suggests avoiding Python object ID recycling issues by tracking ResponseStream objects directly instead of using id(stream), and improving type flexibility by checking for callable and Iterable instead of Sequence when handling collections of examples.

[gemini-code-assist]

Using id(stream) as an identifier in a set can lead to subtle bugs due to object ID recycling in Python. If a stream object is garbage collected and a new one is allocated at the same memory address, a non-streaming response might be incorrectly identified as an active stream. It is safer to store the stream objects directly in a set (or a WeakSet from weakref to avoid keeping aborted streams alive).

Suggested change

	self._active_stream_ids: set[int] = set()
	self._active_streams: set[ResponseStream] = set()

[gemini-code-assist]

Update the stream tracking logic to use the stream object directly instead of id(stream) to prevent ID recycling issues.

Suggested change

	stream_id = id(stream)
	flushed_final = False
	def _convert_part(part: genai_types.Part) -> list[a2a_types.Part]:
	nonlocal flushed_final
	if part.text is not None:
	if is_partial:
	self._active_stream_ids.add(stream_id)
	try:
	return to_a2a_parts(
	stream.feed(part.text),
	version=base_converter._version,
	)
	except Exception as e:
	logger.warning("Failed to stream A2UI chunk: %s", e)
	return []
	if stream_id in self._active_stream_ids:
	self._active_stream_ids.discard(stream_id)
	flushed_final = False
	def _convert_part(part: genai_types.Part) -> list[a2a_types.Part]:
	nonlocal flushed_final
	if part.text is not None:
	if is_partial:
	self._active_streams.add(stream)
	try:
	return to_a2a_parts(
	stream.feed(part.text),
	version=base_converter._version,
	)
	except Exception as e:
	logger.warning("Failed to stream A2UI chunk: %s", e)
	return []
	if stream in self._active_streams:
	self._active_streams.discard(stream)

[gemini-code-assist]

When distinguishing between a callable provider and a collection of items, check callable() to identify the provider rather than checking isinstance(..., Sequence). If a collection is passed directly, convert it to a list once (e.g., in the constructor) to support generators/sets without exhausting them, and explicitly reject single items, strings, or bytes with a clear TypeError.

            if a2ui_examples is not None:
                if callable(a2ui_examples):
                    self._a2ui_examples = a2ui_examples
                elif isinstance(a2ui_examples, (str, bytes)):
                    raise TypeError("a2ui_examples cannot be a string or bytes")
                elif isinstance(a2ui_examples, collections.abc.Iterable):
                    self._a2ui_examples = list(a2ui_examples)
                else:
                    raise TypeError("a2ui_examples must be a callable provider or an iterable collection")

References

1. When distinguishing between a callable provider and a collection of items in Python, check callable() to identify the provider rather than checking isinstance(..., Sequence). If a collection is passed directly, convert it to a list once (e.g., in the constructor) to support generators/sets without exhausting them, and explicitly reject single items, strings, or bytes with a clear TypeError.

[gemini-code-assist]

To support any iterable examples (such as generators or sets) returned by dynamic providers, check isinstance(examples, Iterable) instead of Sequence, and convert it to a list before processing.

            if isinstance(examples, Iterable) and not isinstance(
                examples, (str, bytes)
            ):
                examples_list = list(examples)
                if not examples_list:
                    return ""
                decompiler = DirectJsonDecompiler()
                blocks = [
                    f"<a2ui-json>\n{decompiler.decompile(turn)}\n</a2ui-json>"
                    for turn in examples_list
                ]
                return "### Examples:\n" + "\n\n".join(blocks)