feat(live): let a live streaming tool send messages to the user directly

A streaming tool used to only talk to the model: every value it yielded came
back as a FunctionResponse. Therefore it costs model context and could derail the model's reasoning.

An Event with message yielded by a streaming tool is now addressed to the user instead. It is enqueued on the invocation's event queue, so the runner appends it to the session and streams it to the client, and it is never sent over the live model
connection. Plain values are still sent to the model as FunctionResponse. A tool can mix and match any number of each, in any order.

Runner's run_live also now initializes the invocation's event queue and merges it with the live agent's own event stream. Without a queue, anything running under
the live agent that enqueues an event -- a streaming tool, or a node -- fails
with "_event_queue is not set".

Co-authored-by: Liang Wu <wuliang@google.com>
PiperOrigin-RevId: 968533249

Author: wuliang229

contributing/samples/live/streaming_tool_events/README.md

@@ -0,0 +1,72 @@
+# Streaming Tool Events
+
+**In a streaming tool, `yield Event(message=...)` to talk to the user directly,
+and `yield <value>` to give the model a result. Mix and match, in any order.**
+
+## Overview
+
+A streaming tool reports progress to the user while streaming results to the
+model, so narrating a long-running tool costs no model turn. Only supported in
+streaming (live) agents/api.
+
+## Sample Inputs
+
+- `Help me monitor the stock price for $XYZ stock.`
+
+  *The tool tells you directly that it connected to the feed, without going
+  through the model. The price alerts do go to the model, and it reports them
+  in its own words.*
+
+- `Stop monitoring $XYZ.`
+
+  *The model calls `stop_streaming`, which cancels the background monitor.*
+
+## Graph
+
+```mermaid
+graph TD
+    Agent[streaming_tool_events_agent] -->|calls| Monitor(monitor_stock_price)
+    Agent -->|calls| Stop(stop_streaming)
+```
+
+## How To
+
+Write an `async` generator and put it in `tools`. The yielded type picks the
+audience:
+
+```python
+async def monitor_stock_price(stock_symbol: str) -> AsyncGenerator[Any, None]:
+  """Starts a background monitor for the price of the given stock_symbol."""
+  yield Event(message=f"Connected to the {stock_symbol} price feed.")
+  yield f"the price for {stock_symbol} is 300"
+  yield f"the price for {stock_symbol} is 900"
+  yield Event(message="That is my last update for now.")
+```
+
+Key points:
+
+- **User updates**: yield `Event(message=...)` to send a message straight to
+  the client. `message` takes a string, a `types.Part` or a `types.Content`.
+  Framework metadata (`author`, `branch`, `invocation_id`, the content role)
+  is filled in for you; any other field you set on the event is ignored with a
+  warning, and the message is still delivered.
+- **Model results**: yield a plain value (`str`, `dict`, ...) to send a
+  `FunctionResponse` back to the model.
+- **Side effects**: use `tool_context.actions`, not the event.
+
+### Where the message goes
+
+The message is streamed to your client and appended to the session. It does
+not go over the live connection, so it consumes no model turns or tokens
+during the active turn and cannot derail the model's reasoning mid-task. It is
+ordinary session history, though, so the model does see it once the history is
+replayed on the next connect.
+
+## Related Guides
+
+- [Event and NodeInfo](../../../../docs/guides/events/event/index.md) - How
+  `Event` carries content, actions and metadata, including the `message` field
+  used here.
+- [live_bidi_streaming_tools_agent](../live_bidi_streaming_tools_agent/readme.md) -
+  The streaming tool basics this sample builds on, including `input_stream` and
+  `stop_streaming`.

contributing/samples/live/streaming_tool_events/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import agent

contributing/samples/live/streaming_tool_events/agent.py

@@ -0,0 +1,92 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from typing import Any
+from typing import AsyncGenerator
+
+from google.adk import Event
+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[Any, None]:
+  """Starts a background monitor for the price of the given stock_symbol.
+
+  Call this function ONLY ONCE to initiate monitoring. Once started, it runs
+  continuously in the background and automatically streams price alerts.
+
+  CRITICAL: Do NOT call this function again to "check" or "poll" for updates.
+  Simply wait for the background task to yield new values and report them.
+  Calling this again while running will launch a duplicate background task.
+  """
+  print(f"Start monitor stock price for {stock_symbol}!")
+
+  # An Event goes straight to the user, without going over the live connection
+  # to the model, so status chatter costs no model turn and cannot derail the
+  # model's reasoning mid-task.
+  yield Event(message=f"Connected to the {stock_symbol} price feed.")
+
+  # A plain value goes back to the model as a FunctionResponse, and the model
+  # decides how to report it. This is what a streaming tool has always done.
+  await asyncio.sleep(4)
+  yield f"the price for {stock_symbol} is 300"
+
+  yield Event(message="Trading is getting busy, prices are moving fast.")
+
+  await asyncio.sleep(4)
+  yield f"the price for {stock_symbol} is 400"
+
+  # Each yield is addressed to exactly one audience, so narrating and
+  # reporting at the same moment is simply two yields.
+  await asyncio.sleep(10)
+  yield f"the price for {stock_symbol} is 900"
+  yield Event(message=f"That is my last {stock_symbol} update for now.")
+
+
+# Use this exact function to help ADK stop your streaming tools when requested.
+# For example, to stop `monitor_stock_price` the model calls
+# stop_streaming(function_name="monitor_stock_price").
+def stop_streaming(function_name: str):
+  """Stop the streaming.
+
+  The body is intentionally empty: ADK intercepts this call and cancels the
+  named tool's background task itself. Copy it as is.
+
+  Args:
+    function_name: The name of the streaming function to stop.
+  """
+
+
+root_agent = Agent(
+    # Find supported models in Vertex here: https://docs.cloud.google.com/vertex-ai/generative-ai/docs/live-api
+    model="gemini-live-2.5-flash-native-audio",  # Vertex
+    # Find supported models in Gemini API here: https://ai.google.dev/gemini-api/docs/models
+    # model='gemini-2.5-flash-native-audio-preview-12-2025',  # Gemini API
+    name="streaming_tool_events_agent",
+    instruction="""
+      You are a monitoring agent. You can monitor a stock price using
+      monitor_stock_price.
+      CRITICAL: Only call monitor_stock_price at most once per request. Once
+      called, it runs continuously in the background. Do NOT call it again to
+      "poll" or "check" for updates. Simply wait for it to stream a new price
+      alert to you, and then report that alert to the user.
+      If you need to stop the monitor, call stop_streaming.
+      Don't ask too many questions. Don't be too talkative.
+    """,
+    tools=[
+        monitor_stock_price,
+        FunctionTool(stop_streaming),
+    ],
+)

src/google/adk/flows/llm_flows/functions.py

@@ -1026,6 +1026,93 @@ async def _background_task() -> None:
       return tel_ctx.function_response_event
 
 
+_MESSAGE_EVENT_FIELDS = frozenset({'content', 'id', 'timestamp'})
+"""The Event fields a streaming tool's message is built from.
+
+``id`` and ``timestamp`` are stamped at construction, so every Event carries
+them and neither says anything about what the tool asked for.
+"""
+
+
+def _message_content_for_user(
+    event: Event, *, tool: BaseTool
+) -> Optional[types.Content]:
+  """Returns the content to deliver, or None if the event has no message.
+
+  Only the ``content`` field is considered for delivery. All other fields are
+  ignored. The role is set to "user", overriding any other value.
+
+  Args:
+    event: The event the tool yielded.
+    tool: The tool that yielded it, named in the warning.
+
+  Returns:
+    The content to send to the user, or None if there is nothing to send.
+  """
+  problem = None
+  if not event.content:
+    problem = 'it has no content, so there is nothing to deliver'
+  elif event.model_dump(
+      exclude=set(_MESSAGE_EVENT_FIELDS),
+      exclude_defaults=True,
+      # Load-bearing beside exclude_defaults: a field with a custom serializer
+      # skips the default comparison, so ``long_running_tool_ids`` reports as
+      # set on every event. This reads the raw value instead.
+      exclude_none=True,
+      # Only the presence of a field is read, so a mistyped value is not worth
+      # a warning of its own.
+      warnings=False,
+  ):
+    problem = 'it sets fields beyond the message, which are ignored'
+
+  if problem:
+    logger.warning(
+        'Streaming tool `%s` yielded an Event that is not a purely'
+        ' user-facing message: %s. To send a message, use Event(message=...)',
+        tool.name,
+        problem,
+    )
+  if not event.content:
+    return None
+  return event.content.model_copy(deep=True, update={'role': 'user'})
+
+
+async def _emit_streaming_tool_event(
+    event: Event,
+    *,
+    tool: BaseTool,
+    tool_context: ToolContext,
+    invocation_context: InvocationContext,
+) -> None:
+  """Streams an Event yielded by a streaming tool to the user.
+
+  Args:
+    event: The event the tool yielded.
+    tool: The tool that yielded it, named in the branch and in any warning.
+    tool_context: The context of the call, for its function call id.
+    invocation_context: The invocation to enqueue on.
+  """
+  content = _message_content_for_user(event, tool=tool)
+  if content is None:
+    return
+  # Built fresh rather than copied, so the delivered event carries the message
+  # and nothing else, and each delivery gets its own id and timestamp: a tool
+  # may hold one Event and yield it twice, and the session orders events and
+  # decides what compaction has already summarized by timestamp.
+  await invocation_context._enqueue_event(
+      Event(
+          content=content,
+          author=_require_agent_name(invocation_context),
+          invocation_id=invocation_context.invocation_id,
+          branch=(
+              f'{tool.name}@{tool_context.function_call_id}'
+              if tool_context.function_call_id
+              else tool.name
+          ),
+      )
+  )
+
+
 async def _process_function_live_helper(
     tool: BaseTool,
     tool_context: ToolContext,
@@ -1115,6 +1202,15 @@ async def run_tool_and_update_queue(
         if inspect.isasyncgen(res):
           async with Aclosing(res) as agen:
             async for result in agen:
+              if isinstance(result, Event):
+                await _emit_streaming_tool_event(
+                    result,
+                    tool=tool,
+                    tool_context=tool_context,
+                    invocation_context=invocation_context,
+                )
+                continue
+
               updated_content = _build_function_response_content(
                   tool, result, tool_context.function_call_id
               )

src/google/adk/runners.py

@@ -1961,6 +1961,9 @@ async def run_live(
         live_request_queue=live_request_queue,
         run_config=run_config,
     )
+    # A streaming tool emits its user-facing events here instead of returning
+    # them inline; without a queue those enqueues raise.
+    invocation_context._event_queue = asyncio.Queue()
 
     invocation_context.agent = self._find_agent_to_run(
         invocation_context.session, root_agent
@@ -1975,19 +1978,82 @@ async def execute(ctx: InvocationContext) -> AsyncGenerator[Event, None]:
           yield event
 
     async with aclosing(
-        _with_caller_context(
-            self._exec_with_plugin(
-                invocation_context=invocation_context,
-                session=invocation_context.session,
-                execute_fn=execute,
-                is_live_call=True,
+        self._merge_live_event_streams(
+            invocation_context,
+            _with_caller_context(
+                self._exec_with_plugin(
+                    invocation_context=invocation_context,
+                    session=invocation_context.session,
+                    execute_fn=execute,
+                    is_live_call=True,
+                ),
+                caller_ctx,
             ),
-            caller_ctx,
         )
     ) as agen:
       async for event in agen:
         yield event
 
+  async def _merge_live_event_streams(
+      self,
+      ic: InvocationContext,
+      agent_events: AsyncGenerator[Event, None],
+  ) -> AsyncGenerator[Event, None]:
+    """Interleaves the live agent's events with events from ``ic._event_queue``.
+
+    Code running underneath the live agent — a streaming tool, or a node — has
+    no way to yield an event back through the agent's own stream, so it
+    enqueues on ``ic._event_queue`` instead. Both sources are drained
+    concurrently into one queue and surfaced in the order they are produced.
+
+    Each source keeps its own post-processing: the agent's events are already
+    persisted and plugin-processed by ``_exec_with_plugin``, and the queued
+    events by ``_consume_event_queue``, so nothing is handled twice.
+    """
+    if ic._event_queue is None:
+      raise RuntimeError(
+          'Live event stream merging requires an initialized event queue.'
+      )
+    # Bind the queue to a local: the narrowing above does not reach into the
+    # nested pumps below.
+    event_queue = ic._event_queue
+    done_sentinel = object()
+    merged: asyncio.Queue[Any] = asyncio.Queue(maxsize=1)
+
+    async def _pump_agent_events() -> None:
+      try:
+        async with aclosing(agent_events) as agen:
+          async for event in agen:
+            await merged.put(event)
+      finally:
+        # The queue consumer owns the merged sentinel, so end its stream
+        # rather than the merged one; that also lets already-enqueued events
+        # drain before the merge finishes.
+        await event_queue.put((done_sentinel, None))
+
+    async def _pump_queued_events() -> None:
+      try:
+        async with aclosing(
+            self._consume_event_queue(ic, done_sentinel)
+        ) as agen:
+          async for event in agen:
+            await merged.put(event)
+      finally:
+        await merged.put(done_sentinel)
+
+    agent_task = asyncio.create_task(_pump_agent_events())
+    queue_task = asyncio.create_task(_pump_queued_events())
+    try:
+      while True:
+        event_or_done = await merged.get()
+        if event_or_done is done_sentinel:
+          break
+        yield event_or_done
+    finally:
+      # _cleanup_root_task re-raises a failure from either pump.
+      await self._cleanup_root_task(agent_task, self.agent.name)
+      await self._cleanup_root_task(queue_task, self.agent.name)
+
   def _find_agent_to_run(
       self, session: Session, root_agent: BaseAgent
   ) -> BaseAgent: