Is streaming working by default in ADK's A2A sample?

[iamthanhn]

Hi, I'm someone who is not very familiar with Python and streaming, so I’d like to ask a basic question.

Currently, I’m using ADK to build an agent, and I'm integrating A2A into my ADK setup.

I’m a bit confused about whether the sample message/stream example actually uses streaming. From what I observe, it seems to wait until the entire response is complete before returning anything.

Do I have to explicitly configure streaming_mode for it to actually stream? For example, I saw this:

class A2ARunConfig(RunConfig):
    """Custom override of ADK RunConfig to smuggle extra data through the event loop."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )
    current_task_updater: TaskUpdater
    streaming_mode: StreamingMode.SSE # 👉 Does this line enable streaming with A2A? Why is it missing in the example?

Does this line enable streaming with A2A? Why is it missing in the example?

Since my knowledge is limited, I’m not sure when or why I should use this. Any clarification or simple explanation would be greatly appreciated!

[matino]

Hi,

From my tests, it looks like you need to set streaming_mode to StreamingMode.SSE, otherwise streaming won't work.
But that's not enough.
In your agent executor, you must yield control back to the event loop (e.g. with asyncio.sleep), otherwise all the respons chunks will be served back at once:

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        ....
        async for event in self.runner.run_async(
            session_id=session.id,
            user_id=self.user_id,
            new_message=message,
            run_config=self.run_config,
        ):
            parts = genai_parts_to_a2a(event.content.parts)
            if event.is_final_response():
                task_updater.add_artifact(parts)
                task_updater.complete()
                break
           
            if not event.get_function_calls():
                task_updater.update_status(
                    TaskState.working,
                    message=task_updater.new_agent_message(parts),
                )
                # This is critical
                await asyncio.sleep(0)

Not sure where exactly is the bug, but without the sleep, streaming doesn't work properly.

[matino]

Small update: the "sleep bug" is fixed now with #151