# Streaming Tools

<div class="language-support-tag">
    <span class="lst-supported">Supported in ADK</span><span class="lst-python">Python v0.5.0</span><span class="lst-java">Java v0.2.0</span><span class="lst-preview">Experimental</span>
</div>

Streaming tools allows tools(functions) to stream intermediate results back to agents and agents can respond to those intermediate results. 
For example, we can use streaming tools to monitor the changes of the stock price and have the agent react to it. Another example is we can have the agent monitor the video stream, and when there is changes in video stream, the agent can report the changes.

!!! info

    This is only supported in ADK Gemini Live APIs.

To define a streaming tool, you must adhere to the following:

1.  **Asynchronous Function:** The tool must be an `async` Python function.
2.  **AsyncGenerator Return Type:** The function must be typed to return an `AsyncGenerator`. The first type parameter to `AsyncGenerator` is the type of the data you `yield` (e.g., `str` for text messages, or a custom object for structured data). The second type parameter is typically `None` if the generator doesn't receive values via `send()`.


We support two types of streaming tools:
- Simple type. This is a one type of streaming tools that only take non-video/-audio streams(the streams that you feed to adk web or adk runner) as input.
- Video streaming tools. This only works in video streaming and the video stream(the streams that you feed to adk web or adk runner) will be passed into this function.

Now let's define an agent that can monitor stock price changes and monitor the video stream changes. 

=== "Python"

    ```python
    import asyncio
    from typing import AsyncGenerator

    from google.adk.agents import LiveRequestQueue
    from google.adk.agents.llm_agent import Agent
    from google.adk.tools.function_tool import FunctionTool
    from google.genai import Client
    from google.genai import types as genai_types


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

      # Let's mock stock price change.
      await asyncio.sleep(4)
      price_alert1 = f"the price for {stock_symbol} is 300"
      yield price_alert1
      print(price_alert1)

      await asyncio.sleep(4)
      price_alert1 = f"the price for {stock_symbol} is 400"
      yield price_alert1
      print(price_alert1)

      await asyncio.sleep(20)
      price_alert1 = f"the price for {stock_symbol} is 900"
      yield price_alert1
      print(price_alert1)

      await asyncio.sleep(20)
      price_alert1 = f"the price for {stock_symbol} is 500"
      yield price_alert1
      print(price_alert1)


    # for video streaming, `input_stream: LiveRequestQueue` is required and reserved key parameter for ADK to pass the video streams in.
    async def monitor_video_stream(
        input_stream: LiveRequestQueue,
    ) -> AsyncGenerator[str, None]:
      """Monitor how many people are in the video streams."""
      print("start monitor_video_stream!")
      client = Client(vertexai=False)
      prompt_text = (
          "Count the number of people in this image. Just respond with a numeric"
          " number."
      )
      last_count = None
      while True:
        last_valid_req = None
        print("Start monitoring loop")

        # use this loop to pull the latest images and discard the old ones
        while input_stream._queue.qsize() != 0:
          live_req = await input_stream.get()

          if live_req.blob is not None and live_req.blob.mime_type == "image/jpeg":
            last_valid_req = live_req

        # If we found a valid image, process it
        if last_valid_req is not None:
          print("Processing the most recent frame from the queue")

          # Create an image part using the blob's data and mime type
          image_part = genai_types.Part.from_bytes(
              data=last_valid_req.blob.data, mime_type=last_valid_req.blob.mime_type
          )

          contents = genai_types.Content(
              role="user",
              parts=[image_part, genai_types.Part.from_text(prompt_text)],
          )

          # Call the model to generate content based on the provided image and prompt
          response = client.models.generate_content(
              model="gemini-flash-latest",
              contents=contents,
              config=genai_types.GenerateContentConfig(
                  system_instruction=(
                      "You are a helpful video analysis assistant. You can count"
                      " the number of people in this image or video. Just respond"
                      " with a numeric number."
                  )
              ),
          )
          if not last_count:
            last_count = response.candidates[0].content.parts[0].text
          elif last_count != response.candidates[0].content.parts[0].text:
            last_count = response.candidates[0].content.parts[0].text
            yield response
            print("response:", response)

        # Wait before checking for new images
        await asyncio.sleep(0.5)


    # Use this exact function to help ADK stop your streaming tools when requested.
    # for example, if we want to stop `monitor_stock_price`, then the agent will
    # invoke this function with stop_streaming(function_name=monitor_stock_price).
    def stop_streaming(function_name: str):
      """Stop the streaming

      Args:
        function_name: The name of the streaming function to stop.
      """
      pass


    root_agent = Agent(
        model="gemini-flash-latest",
        name="video_streaming_agent",
        instruction="""
          You are a monitoring agent. You can do video monitoring and stock price monitoring
          using the provided tools/functions.
          When users want to monitor a video stream,
          You can use monitor_video_stream function to do that. When monitor_video_stream
          returns the alert, you should tell the users.
          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=[
            monitor_video_stream,
            monitor_stock_price,
            FunctionTool(stop_streaming),
        ]
    )
    ```

=== "Java"

    ```java
    import com.google.adk.agents.LiveRequestQueue;
    import com.google.adk.agents.LlmAgent;
    import com.google.adk.tools.Annotations.Schema;
    import com.google.adk.tools.FunctionTool;
    import com.google.genai.Client;
    import com.google.genai.types.Content;
    import com.google.genai.types.GenerateContentConfig;
    import com.google.genai.types.GenerateContentResponse;
    import com.google.genai.types.Part;
    import io.reactivex.rxjava3.core.Flowable;
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;

    public class StreamingTools {

      @Schema(description = "This function will monitor the price for the given stock_symbol in a continuous, streaming and asynchronously way.")
      public static Flowable<Map<String, Object>> monitorStockPrice(@Schema(name = "stockSymbol") String stockSymbol) {
        System.out.println("Start monitor stock price for " + stockSymbol + "!");

        return Flowable.concat(
            Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 300")).delay(4, TimeUnit.SECONDS),
            Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 400")).delay(4, TimeUnit.SECONDS),
            Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 900")).delay(20, TimeUnit.SECONDS),
            Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 500")).delay(20, TimeUnit.SECONDS)
        );
      }

      // for video streaming, `inputStream` is required and reserved parameter for ADK to pass the video streams in.
      @Schema(description = "Monitor how many people are in the video streams.")
      public static Flowable<Map<String, Object>> monitorVideoStream(@Schema(name = "inputStream") LiveRequestQueue inputStream) {
        System.out.println("start monitor_video_stream!");
        Client client = Client.builder().build();
        String promptText = "Count the number of people in this image. Just respond with a numeric number.";
        
        // We use RxJava to process the stream
        return inputStream.get()
            .filter(req -> req.blob().isPresent() && "image/jpeg".equals(req.blob().get().mimeType()))
            .sample(500, TimeUnit.MILLISECONDS) // Process one frame every 0.5 seconds
            .map(req -> {
              System.out.println("Processing the most recent frame from the queue");
              Part imagePart = Part.builder().inlineData(req.blob().get()).build();
              Content contents = Content.builder()
                  .role("user")
                  .parts(Arrays.asList(imagePart, Part.fromText(promptText)))
                  .build();

              GenerateContentResponse response = client.models().generateContent(
                  "gemini-flash-latest",
                  contents,
                  GenerateContentConfig.builder()
                      .systemInstruction(Content.builder().parts(Arrays.asList(
                          Part.fromText("You are a helpful video analysis assistant. You can count the number of people in this image or video. Just respond with a numeric number.")
                      )).build())
                      .build()
              );
              return (Map<String, Object>) Collections.<String, Object>singletonMap("result", response.text());
            })
            .distinctUntilChanged()
            .doOnNext(res -> System.out.println("response: " + res));
      }

      // Use this exact function to help ADK stop your streaming tools when requested.
      @Schema(description = "Stop the streaming")
      public static void stopStreaming(
          @Schema(name = "functionName", description = "The name of the streaming function to stop.") String functionName) {
        // Stop the streaming logic
      }

      public static void main(String[] args) {
        LlmAgent rootAgent = LlmAgent.builder()
            .model("gemini-flash-latest")
            .name("video_streaming_agent")
            .instruction(
                "You are a monitoring agent. You can do video monitoring and stock price monitoring\n" +
                "using the provided tools/functions.\n" +
                "When users want to monitor a video stream,\n" +
                "You can use monitorVideoStream function to do that. When monitorVideoStream\n" +
                "returns the alert, you should tell the users.\n" +
                "When users want to monitor a stock price, you can use monitorStockPrice.\n" +
                "Don't ask too many questions. Don't be too talkative."
            )
            .tools(Arrays.asList(
                FunctionTool.create(StreamingTools.class, "monitorVideoStream"),
                FunctionTool.create(StreamingTools.class, "monitorStockPrice"),
                FunctionTool.create(StreamingTools.class, "stopStreaming")
            ))
            .build();
      }
    }
    ```

Here are some sample queries to test:
- Help me monitor the stock price for $XYZ stock.
- Help me monitor how many people are there in the video stream.
