Streaming tool support for ADK Agent

[velusrinathmr]

I am trying to make use of a tool which can stream the responses back to the agent. But always getting stuck with the Async generator serialisation error. I also tried using the examples given in https://google.github.io/adk-docs/streaming/streaming-tools/.

How to make it work for an ADK agent?

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: The Async generator serialisation error you are encountering was a known issue in older versions of the ADK. Please upgrade your google-adk package to version 1.11.0 or newer to resolve this problem.

---

Hello! It looks like you're running into an issue with serializing AsyncGenerator objects when using streaming tools in the ADK.

This error typically occurs when the framework attempts to serialize the generator object itself, rather than consuming the values it yields. This was a recognized problem in earlier versions of the ADK and was specifically addressed in the v1.11.0 release [2].

To fix this, please update your google-adk library to the latest version:

pip install --upgrade google-adk

For your reference, a streaming tool should be defined as an async function that returns an AsyncGenerator, as you have seen in the documentation [1]. Here is a quick example of the correct structure:

import asyncio
from typing import AsyncGenerator

async def my_streaming_tool() -> AsyncGenerator[str, None]:
    """A tool that streams back a sequence of strings."""
    for i in range(5):
        yield f"This is chunk {i}\n"
        await asyncio.sleep(1)

Upgrading should resolve the serialization error and allow the ADK to correctly handle the streaming output from your tool.

[1] https://google.github.io/adk-docs/streaming/streaming-tools/
[2] https://github.com/google/adk-python/blob/main/CHANGELOG.md

[velusrinathmr]

Google ADK version I am using is 1.14.1

I am adding the code below,

import asyncio
from collections.abc import AsyncGenerator
from google.adk.agents.llm_agent import Agent
from google.adk.tools.function_tool import FunctionTool

async def monitor_stock_price(stock_symbol: str) -> AsyncGenerator[str, None]:
"""This function will monitor the price for the given stock_symbol in a simple way."""
yield f"Start monitor stock price for {stock_symbol}!"

await asyncio.sleep(2)
price = f"Current price for {stock_symbol} is $350"
yield price

root_agent = Agent(
model="gemini-2.0-flash-exp",
name="video_streaming_agent",
instruction="""
You are a monitoring agent. You can do stock price monitoring
using the provided tools/functions.
When users want to monitor a stock price, you can use monitor_stock_price.
Don't ask too many questions. Don't be too talkative.
""",
tools=[
FunctionTool(monitor_stock_price)
],
)