feat(agent-sdk): Add streaming support to Python agent SDK formats

[gspencergoog]

Important

Stacked PR Series:

1. #1981 (Foundation & Experiments API)
2. #1982 (Inference Strategies & Eval Solver)
3. #1996 (Streaming Support) 👈 This PR

Summary

Implements stateful incremental parsing and fallback streaming capabilities on Python agent SDK strategies, enabling real-time token rendering for Express and Elemental formats.

Changes

• Stateful Streaming Support:
  • Implemented stateful streaming parser logic on Parser subclasses to support incremental chunk processing.
  • Implemented StreamingPartConverter (under a2ui/a2a/parts.py) supporting active state tracking of the stream.
  • Created a robust plain text fallback parser strategy that suppresses partially rendered A2UI tags and parses incomplete blocks.
• Robustness Refinements:
  • Updated tag prefix matching rules to support open tags with custom attributes (e.g. <a2ui id="main">).
  • Added robust validation type checks and descriptive logging when parsing streams.
• Testing:
  • Added tests/test_streaming.py verifying stateful streaming parsing across various chunk segment variations.

Impact & Risks

• None. Streaming parsing is backward-compatible and optimizes the response token flow.

Testing

Verify the changes by running all Python tests:

uv run pytest agent_sdks/python/a2ui_agent/tests/

[gemini-code-assist]

Code Review

This pull request introduces stateful streaming support for converting token streams directly into A2A Parts by adding the StreamingPartConverter class, integrating json-repair to handle truncated JSON blocks, and adding comprehensive streaming tests. Feedback on these changes suggests suppressing trailing partial prefixes of the opening tag to prevent raw tag leakage when split across chunk boundaries, and optimizing validation by only running it on the final chunk to reduce CPU overhead and ensure a consistent user experience.

[gemini-code-assist]

During streaming, if a chunk boundary splits the opening tag (e.g., is split into ), the partial tag will leak and be rendered to the user as plain text. To prevent this raw tag leakage and ensure a smooth user experience, we should suppress any trailing partial prefix of the opening tag at the end of the buffer. Additionally, we must enforce a minimum length of at least 3 characters for partial matches to prevent incorrect or overly loose matches.

Suggested change

	except Exception:
	open_prefix = self.parser.open_tag_prefix
	open_idx = self._buffer.find(open_prefix)
	if open_idx != -1:
	response_parts = [ResponsePart(text=self._buffer[:open_idx])]
	else:
	response_parts = [ResponsePart(text=self._buffer)]
	except Exception:
	open_prefix = self.parser.open_tag_prefix
	open_idx = self._buffer.find(open_prefix)
	if open_idx != -1:
	response_parts = [ResponsePart(text=self._buffer[:open_idx])]
	else:
	text = self._buffer
	for i in range(len(open_prefix), 2, -1):
	if text.endswith(open_prefix[:i]):
	text = text[:-i]
	break
	response_parts = [ResponsePart(text=text)]

References

1. When implementing substring or prefix matching logic, prioritize exact matches first and enforce a minimum length (e.g., at least 3 characters) for partial matches to prevent incorrect or overly loose matches.

[gemini-code-assist]

Running validation on every intermediate chunk during streaming is highly inefficient and can cause significant CPU overhead and latency spikes, especially since any validation errors are silently ignored anyway. Additionally, discarding the final UI part on validation failure (via continue) while having displayed intermediate versions of it during streaming creates an inconsistent and jarring user experience.

We should only run the validator when is_final is True, and log the warning without discarding the part.

                if self.validator and is_final:
                    try:
                        self.validator.validate(json_data)
                    except Exception as e:
                        logger.warning(
                            f"Failed to validate final A2UI response: {e}"
                        )