refactor(gemini-live): bring tool-result handling in line with the canonical realtime pattern

Lays groundwork for cancel_on_interruption=False support on Gemini Live by
restructuring _process_completed_function_calls to match the shape used by
AWSNovaSonicLLMService and OpenAIRealtimeLLMService in #4441: a single-pass
forward iteration over raw context messages that detects async-tool
messages via async_tool_messages.parse_message and routes them — started
skipped silently, intermediate logged-as-error and surfaced via push_error,
final delivered via the formal FunctionResponse channel.

Replaces the prior two-pass structure that went through the adapter for
sync results — the service now uses a lightweight self._tool_call_id_to_name
map (populated when the model issues tool calls) for the name lookup the
adapter used to provide. Extracts a new GeminiLLMAdapter.to_function_response_dict
static method for the dict-coercion logic that wraps non-dict tool returns
as {value: <result>} for Gemini's FunctionResponse.response field; the
adapter's existing inline copy in _from_standard_message uses it too.

Example consolidation:

- Folds realtime-gemini-live-function-calling.py into the base
  realtime-gemini-live.py example so the base exercises function calling
  out of the box (matching realtime-openai.py and realtime-aws-nova-sonic.py).
- Renames realtime-gemini-live-vertex-function-calling.py to
  realtime-gemini-live-vertex.py, mirroring the consolidation.
- Adds realtime-gemini-live-async-tool.py.
- Updates scripts/evals/run-release-evals.py for the renames.

This commit alone doesn't make cancel_on_interruption=False fully work on
Gemini Live — additional investigation is pending. This is foundational
work to be built on.

examples/realtime/realtime-gemini-live-async-tool.py

@@ -4,15 +4,25 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+"""Example: async function call with the Gemini Live LLM service.
 
+The ``get_current_weather`` tool is registered with
+``cancel_on_interruption=False`` and simulates a slow API call (10s sleep).
+While the call is in flight the conversation continues; the result arrives
+later via the async-tool mechanism and is forwarded to Gemini Live as a
+FunctionResponse so the model can integrate it naturally into its next turn.
+"""
+
+import asyncio
 import os
+import random
 from datetime import datetime
 
 from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -31,33 +41,55 @@ load_dotenv(override=True)
 
 
 async def fetch_weather_from_api(params: FunctionCallParams):
-    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    # Simulate a long-running API call so we can demonstrate that the
+    # conversation continues while the tool is in flight.
+    await asyncio.sleep(10)
+    temperature = (
+        random.randint(60, 85)
+        if params.arguments["format"] == "fahrenheit"
+        else random.randint(15, 30)
+    )
     await params.result_callback(
         {
             "conditions": "nice",
             "temperature": temperature,
+            "location": params.arguments["location"],
             "format": params.arguments["format"],
             "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
         }
     )
 
 
-async def fetch_restaurant_recommendation(params: FunctionCallParams):
-    await params.result_callback({"name": "The Golden Dragon"})
+weather_function = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
+        },
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the user's location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+tools = ToolsSchema(standard_tools=[weather_function])
 
 
-system_instruction = """
-You are a helpful assistant who can answer questions and use tools.
-
-You have three tools available to you:
-1. get_current_weather: Use this tool to get the current weather in a specific location.
-2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location.
-3. google_search: Use this tool to search the web for information.
-"""
+system_instruction = (
+    "You are a friendly assistant. The user and you will engage in a spoken "
+    "dialog exchanging the transcripts of a natural real-time conversation. "
+    "Keep your responses short, generally two or three sentences for chatty "
+    "scenarios. When the user asks for the weather, call get_current_weather. "
+    "While you wait for the result, keep chatting with the user. When the "
+    "result arrives, share it with the user naturally."
+)
 
 
-# We use lambdas to defer transport parameter creation until the transport
-# type is selected at runtime.
 transport_params = {
     "daily": lambda: DailyParams(
         audio_in_enabled=True,
@@ -77,42 +109,6 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    weather_function = FunctionSchema(
-        name="get_current_weather",
-        description="Get the current weather",
-        properties={
-            "location": {
-                "type": "string",
-                "description": "The city and state, e.g. San Francisco, CA",
-            },
-            "format": {
-                "type": "string",
-                "enum": ["celsius", "fahrenheit"],
-                "description": "The temperature unit to use. Infer this from the user's location.",
-            },
-        },
-        required=["location", "format"],
-    )
-    restaurant_function = FunctionSchema(
-        name="get_restaurant_recommendation",
-        description="Get a restaurant recommendation",
-        properties={
-            "location": {
-                "type": "string",
-                "description": "The city and state, e.g. San Francisco, CA",
-            },
-        },
-        required=["location"],
-    )
-    search_tool = {"google_search": {}}
-    # KNOWN ISSUE: If using GeminiVertexLiveLLMService, it appears
-    # you cannot use the "google_search" tool alongside other tools.
-    # See https://github.com/googleapis/python-genai/issues/941.
-    tools = ToolsSchema(
-        standard_tools=[weather_function, restaurant_function],
-        custom_tools={AdapterType.GEMINI: [search_tool]},
-    )
-
     llm = GeminiLiveLLMService(
         api_key=os.environ["GOOGLE_API_KEY"],
         settings=GeminiLiveLLMService.Settings(
@@ -121,13 +117,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         tools=tools,
     )
 
-    llm.register_function("get_current_weather", fetch_weather_from_api)
-    llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+    llm.register_function(
+        "get_current_weather",
+        fetch_weather_from_api,
+        cancel_on_interruption=False,
+    )
 
-    # You can provide the system instructions and tools in the context rather
-    # than as arguments to GeminiLiveLLMService, but note that doing so will
-    # trigger a (fast) reconnection when the GeminiLiveLLMService first
-    # receives the context (i.e. when we send the LLMRunFrame below).
     context = LLMContext()
     # Server-side VAD is enabled by default; no local VAD is added.
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)
@@ -154,7 +149,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
-        # Kick off the conversation.
         context.add_message(
             {"role": "developer", "content": "Please introduce yourself to the user."}
         )
@@ -166,7 +160,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         await task.cancel()
 
     runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
-
     await runner.run(task)
 
 

examples/realtime/realtime-gemini-live.py

@@ -6,10 +6,13 @@
 
 
 import os
+from datetime import datetime
 
 from dotenv import load_dotenv
 from loguru import logger
 
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -23,6 +26,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
+from pipecat.services.llm_service import FunctionCallParams
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
@@ -30,6 +34,32 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
 load_dotenv(override=True)
 
 
+async def fetch_weather_from_api(params: FunctionCallParams):
+    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": params.arguments["format"],
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+async def fetch_restaurant_recommendation(params: FunctionCallParams):
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
+system_instruction = """
+You are a helpful assistant who can answer questions and use tools.
+
+You have three tools available to you:
+1. get_current_weather: Use this tool to get the current weather in a specific location.
+2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location.
+3. google_search: Use this tool to search the web for information.
+"""
+
+
 # We use lambdas to defer transport parameter creation until the transport
 # type is selected at runtime.
 transport_params = {
@@ -51,23 +81,55 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    restaurant_function = FunctionSchema(
+        name="get_restaurant_recommendation",
+        description="Get a restaurant recommendation",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+        },
+        required=["location"],
+    )
+    search_tool = {"google_search": {}}
+    # KNOWN ISSUE: If using GeminiVertexLiveLLMService, it appears
+    # you cannot use the "google_search" tool alongside other tools.
+    # See https://github.com/googleapis/python-genai/issues/941.
+    tools = ToolsSchema(
+        standard_tools=[weather_function, restaurant_function],
+        custom_tools={AdapterType.GEMINI: [search_tool]},
+    )
+
     llm = GeminiLiveLLMService(
         api_key=os.environ["GOOGLE_API_KEY"],
         settings=GeminiLiveLLMService.Settings(
+            system_instruction=system_instruction,
             voice="Aoede",  # Puck, Charon, Kore, Fenrir, Aoede
-            # system_instruction="Talk like a pirate."
         ),
-        # inference_on_context_initialization=False,
+        tools=tools,
     )
 
-    context = LLMContext(
-        [
-            {
-                "role": "user",
-                "content": "Say hello. Then ask if I want to hear a joke.",
-            },
-        ],
-    )
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+
+    context = LLMContext()
     # Server-side VAD is enabled by default; no local VAD is added.
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)
 
@@ -94,6 +156,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
         # Kick off the conversation.
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
         await task.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_client_disconnected")

scripts/evals/run-release-evals.py

@@ -223,12 +223,11 @@ TESTS_REALTIME = [
     # ("realtime/realtime-azure.py", EVAL_WEATHER),
     ("realtime/realtime-openai-text.py", EVAL_WEATHER),
     ("realtime/realtime-openai-live-video.py", EVAL_VISION_CAMERA),
-    ("realtime/realtime-gemini-live.py", EVAL_SIMPLE_MATH),
+    ("realtime/realtime-gemini-live.py", EVAL_WEATHER),
     ("realtime/realtime-gemini-live-local-vad.py", EVAL_SIMPLE_MATH),
-    ("realtime/realtime-gemini-live-function-calling.py", EVAL_WEATHER),
     ("realtime/realtime-gemini-live-video.py", EVAL_VISION_CAMERA),
     ("realtime/realtime-gemini-live-google-search.py", EVAL_ONLINE_SEARCH),
-    ("realtime/realtime-gemini-live-vertex-function-calling.py", EVAL_WEATHER),
+    ("realtime/realtime-gemini-live-vertex.py", EVAL_WEATHER),
     ("realtime/realtime-aws-nova-sonic.py", EVAL_SIMPLE_MATH),
     ("realtime/realtime-ultravox.py", EVAL_ORDER),
     ("realtime/realtime-grok.py", EVAL_WEATHER),

src/pipecat/adapters/services/gemini_adapter.py

@@ -139,6 +139,36 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
 
         return formatted_standard_tools + custom_gemini_tools
 
+    @staticmethod
+    def to_function_response_dict(content: Any) -> dict[str, Any]:
+        """Convert a tool-result content value to Gemini's FunctionResponse.response shape.
+
+        Gemini's ``FunctionResponse.response`` field requires a dict, so
+        non-dict values (e.g. plain strings, JSON-encoded scalars, or
+        sentinel strings like ``"COMPLETED"`` used when a function returned
+        no value) are wrapped as ``{"value": <value>}``. JSON strings that
+        decode to a dict are passed through as-is.
+
+        Args:
+            content: The tool-result content. Typically the JSON-encoded
+                return value of a function, but can also be a plain string
+                (e.g. ``"COMPLETED"``) or already-parsed dict.
+
+        Returns:
+            A dict suitable for ``FunctionResponse.response``.
+        """
+        if isinstance(content, dict):
+            return content
+        if not isinstance(content, str):
+            return {"value": content}
+        try:
+            decoded = json.loads(content)
+        except (json.JSONDecodeError, ValueError):
+            return {"value": content}
+        if isinstance(decoded, dict):
+            return decoded
+        return {"value": decoded}
+
     def get_messages_for_logging(self, context: LLMContext) -> list[dict[str, Any]]:
         """Get messages from a universal LLM context in a format ready for logging about Gemini.
 
@@ -382,16 +412,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 )
         elif role == "tool":
             role = "user"
-            try:
-                response = json.loads(msg["content"])
-                if isinstance(response, dict):
-                    response_dict = response
-                else:
-                    response_dict = {"value": response}
-            except Exception as e:
-                # Response might not be JSON-deserializable.
-                # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
-                response_dict = {"value": msg["content"]}
+            response_dict = self.to_function_response_dict(msg["content"])
 
             # Get function name from mapping using tool_call_id, or fallback
             tool_call_id = msg.get("tool_call_id")

src/pipecat/services/google/gemini_live/llm.py

@@ -14,6 +14,7 @@ voice transcription, streaming responses, and tool usage.
 import asyncio
 import base64
 import io
+import json
 import time
 import uuid
 from dataclasses import dataclass, field
@@ -56,7 +57,8 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators import async_tool_messages
+from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
 from pipecat.services.google.utils import update_google_client_http_options
@@ -557,6 +559,11 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
 
         # Bookkeeping for tool calls
         self._completed_tool_calls = set()
+        # tool_call_id -> tool_name, populated as the model issues tool
+        # calls. Used to look up the function name when sending an async
+        # tool's final result back to the provider, since the async-tool
+        # message in the context only carries the id.
+        self._tool_call_id_to_name: dict[str, str] = {}
 
     def create_client(self):
         """Create the Gemini API client instance. Subclasses can override this."""
@@ -842,27 +849,58 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
 
     async def _process_completed_function_calls(self, send_new_results: bool):
         # Check for set of completed function calls in the context
-        adapter = self.get_llm_adapter()
-        messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
-        for message in messages:
-            if message.parts:
-                for part in message.parts:
-                    if part.function_response:
-                        tool_call_id = part.function_response.id
-                        tool_name = part.function_response.name
-                        response = part.function_response.response
-                        if (
-                            tool_call_id
-                            and tool_call_id not in self._completed_tool_calls
-                            and response
-                            and response.get("value") != "IN_PROGRESS"
-                        ):
-                            # Found a newly-completed function call - send the result to the service
-                            if send_new_results:
-                                await self._tool_result(
-                                    tool_call_id, tool_name, part.function_response.response
-                                )
-                            self._completed_tool_calls.add(tool_call_id)
+        for message in self._context.get_messages():
+            # LLMSpecificMessages are opaque provider-specific payloads, not
+            # standard tool-result messages — skip them.
+            if isinstance(message, LLMSpecificMessage):
+                continue
+
+            # Async-tool messages live alongside regular tool messages in the
+            # context; detect and route them before the regular logic so we
+            # don't try to send the async-tool envelope JSON as a tool result.
+            async_payload = async_tool_messages.parse_message(message)
+            if async_payload is not None:
+                if async_payload.tool_call_id in self._completed_tool_calls:
+                    continue
+                if async_payload.kind == "started":
+                    # The provider already issued the tool call and natively
+                    # awaits a result; nothing to send for the started marker.
+                    continue
+                if async_payload.kind == "intermediate":
+                    logger.error(
+                        f"{self}: Gemini Live does not support streamed async "
+                        f"tool results; dropping intermediate result for "
+                        f"tool_call_id={async_payload.tool_call_id}. Use a "
+                        f"non-realtime LLM service if your tool needs to "
+                        f"stream intermediate results."
+                    )
+                    await self.push_error(
+                        error_msg="Gemini Live does not support streamed async tool results.",
+                    )
+                    continue
+                # kind == "final": deliver via the formal tool-response channel
+                # — same path as a synchronous tool result, just delayed.
+                tool_name = self._tool_call_id_to_name.get(
+                    async_payload.tool_call_id, "tool_call_result"
+                )
+                response_dict = GeminiLLMAdapter.to_function_response_dict(async_payload.result)
+                if send_new_results:
+                    await self._tool_result(async_payload.tool_call_id, tool_name, response_dict)
+                self._completed_tool_calls.add(async_payload.tool_call_id)
+                continue
+
+            # Look for newly-completed "regular" (as opposed to async-tool) results
+            if message.get("role") == "tool" and message.get("content") != "IN_PROGRESS":
+                tool_call_id = message.get("tool_call_id")
+                if tool_call_id and tool_call_id not in self._completed_tool_calls:
+                    # Found a newly-completed function call - send the result to the service
+                    tool_name = self._tool_call_id_to_name.get(tool_call_id, "tool_call_result")
+                    response_dict = GeminiLLMAdapter.to_function_response_dict(
+                        message.get("content")
+                    )
+                    if send_new_results:
+                        await self._tool_result(tool_call_id, tool_name, response_dict)
+                    self._completed_tool_calls.add(tool_call_id)
 
     async def _set_bot_is_responding(self, responding: bool):
         if self._bot_is_responding == responding:
@@ -1193,6 +1231,7 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
                 await self._session.close()
                 self._session = None
             self._completed_tool_calls = set()
+            self._tool_call_id_to_name = {}
             self._ready_for_realtime_input = False
             self._disconnecting = False
         except Exception as e:
@@ -1420,6 +1459,10 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
         if self._disconnecting or not self._session:
             return
 
+        logger.debug(
+            f"Sending tool result to Gemini Live for tool_call_id={tool_call_id}, tool_result_message={tool_result_message}"
+        )
+
         # For now we're shoving the name into the tool_call_id field, so this
         # will work until we revisit that.
         response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message)
@@ -1555,6 +1598,9 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
             for f in function_calls
         ]
 
+        for fc in function_calls_llm:
+            self._tool_call_id_to_name[fc.tool_call_id] = fc.function_name
+
         await self.run_function_calls(function_calls_llm)
 
     @traced_gemini_live(operation="llm_response")