How to stream functionCall content token-by-token?

[mrcaidev]

In the current version of ADK, functionCall deltas are first internally gathered by ADK and then sent as a whole into the SSE stream.

In scenarios like run_code tool call, where functionCall content can be extremely long, this design makes SSE feel laggy/frozen.

If possible, I would like functionCall to be streamed token-by-token, e.g. in AI SDK protocol we have tool-input-start, tool-input-delta and tool-input-available.

Is that possible? And if yes, how?

[adk-bot]

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

TLDR: It is possible to stream functionCall content token-by-token in ADK by defining your tool as a "Streaming Tool." This is an experimental feature available in ADK Python v0.5.0 and later.

---

Based on the information in the document store, you can achieve the desired token-by-token streaming of functionCall content by implementing your tool as a "Streaming Tool." [8, 14]

This will allow your tool to stream intermediate results back to the agent, which can then be sent to the client, preventing the SSE stream from feeling laggy.

To define a streaming tool, you need to adhere to the following conventions [14]:

• The tool must be an async Python function.
• The function's return type must be an AsyncGenerator.

Here is a conceptual example of how you could implement a streaming run_code tool:

from typing import AsyncGenerator

async def run_code(code: str) -> AsyncGenerator[str, None]:
    # Execute the code and stream the output
    for line in execute_and_stream(code):
        yield line

By using yield, the output of the run_code tool will be streamed back to the agent as it becomes available, rather than being sent as a single, large chunk.

This approach is specifically designed for scenarios like yours, where the output of a tool can be lengthy and cause delays in the user experience.