No answer received when streaming reasoning + enforcing output schema + using tools

[ramcav]

Describe the bug
When using output_schema with streaming mode (StreamingMode.SSE), function calls from set_model_response are being lost or processed incorrectly. This means that the final response contains the complete set of streamed thoughts but an empty string for the answer, causing:

1. Function calls not being registered: Despite the raw GenAI API response containing function calls, they do not appear in the final Event objects
2. Early stream termination: The streaming loop breaks early before receiving the function call response
3. Parsing errors: When trying to parse the final response into the expected Pydantic model, the missing function calls makes it so an empty string tries to get parsed (resulting in a ValidationError)

To Reproduce

Steps to reproduce the behavior:

1. Install google-adk latest version

2. Create an agent with output_schema and enable streaming:

   from pydantic import BaseModel, Field
   from google.adk.agents import LlmAgent
   from google.adk.runners import Runner
   from google.adk.agents.run_config import RunConfig, StreamingMode
   from google.adk.sessions import InMemorySessionService
   from google.adk.planners import BuiltInPlanner
   from google.genai.types import GenerateContentConfig, ThinkingConfig, Content, Part
   
   # Define output schema
   class ShoppingItem(BaseModel):
       name: str
       quantity: int
       price: float
   
   class ShoppingList(BaseModel):
       items: list[ShoppingItem]
       total_price: float
       complexity: str = Field(description="The complexity of the shopping list, either 'easy', 'medium', or 'hard'")
   
   # Create agent with output_schema
   planner = BuiltInPlanner(
       thinking_config=ThinkingConfig(include_thoughts=True, thinking_budget=-1),
   )
   
   agent = LlmAgent(
       name="shopper",
       model="gemini-2.0-flash-exp",  # or any model that supports function calling and will
       description="AI shopping list assistant",
       instruction="You are a helpful shopping list assistant.",
       generate_content_config=GenerateContentConfig(temperature=0.2, seed=42),
       output_schema=ShoppingList,
       planner=planner,
       tools=[tool1, tool2, ...]
   )
   
   # Create runner and stream
   session_service = InMemorySessionService()
   await session_service.create_session(
       app_name="test",
       user_id="test",
       session_id="test",
   )
   
   runner = Runner(
       agent=agent,
       session_service=session_service,
       app_name="test",
   )
   
   event_stream = runner.run_async(
       user_id="test",
       session_id="test",
       new_message=Content(role="user", parts=[Part(text="I need you to give me shopping list for reina pepiada")]),
       run_config=RunConfig(streaming_mode=StreamingMode.SSE),
   )

3. Iterate through the event stream and check for function calls:

   event_count = 0
   function_call_count = 0
   
   async for event in event_stream:
       event_count += 1
       function_calls = event.get_function_calls()
       
       if function_calls:
           function_call_count += 1
           print(f"✅ Event {event_count}: Found {len(function_calls)} function call(s)")
       else:
           # Check for thoughts or text
           if event.content and event.content.parts:
               for part in event.content.parts:
                   if hasattr(part, 'text') and part.text:
                       is_thought = getattr(part, 'thought', False)
                       print(f"📝 Event {event_count}: {'Thought' if is_thought else 'Text'}")
   
   print(f"Total events: {event_count}")
   print(f"Events with function_calls: {function_call_count}")
   # Output: Events with function_calls: 0
   # Expected: Events with function_calls: 1 (with set_model_response)

4. Observed behavior: The stream completes with 0 function calls detected, even though the raw GenAI API response contains a set_model_response function call with the structured JSON output.

5. Error or stacktrace: A Pydantic Validation error should arise when trying to parse the response

Expected behavior

When using output_schema with streaming:

• Function calls should be properly detected and yielded in the event stream
• The streaming loop should not terminate prematurely before receiving the function call
• Function call responses should not be duplicated
• The final structured response from set_model_response should be accessible via event.get_function_calls() or event.content.parts with function_response

[ramcav]

I already made a patch locally for the issue and will open a PR for it!

[ramcav]

@DeanChensj

I was looking over the bug to create a PR as I said earlier. However, I saw that there were a bunch of changes made to the streaming_utils.py file which is the one I made the fix for.

These are the changes I had planned to submit:

# Copyright 2025 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 __future__ import annotations

from typing import AsyncGenerator
from typing import Optional

from google.genai import types

from ..models.llm_response import LlmResponse


class StreamingResponseAggregator:
    """Aggregates partial streaming responses.

    It aggregates content from partial responses, and generates LlmResponses for
    individual (partial) model responses, as well as for aggregated content.
    """

    def __init__(self):
        self._text = ""
        self._thought_text = ""
        self._usage_metadata = None
        self._response = None

    async def process_response(
        self, response: types.GenerateContentResponse
    ) -> AsyncGenerator[LlmResponse, None]:
        """Processes a single model response.

        Args:
          response: The response to process.

        Yields:
          The generated LlmResponse(s), for the partial response, and the aggregated
          response if needed.
        """
        # results = []
        self._response = response
        llm_response = LlmResponse.create(response)
        self._usage_metadata = llm_response.usage_metadata

        # Check if this response has function calls
        has_function_call = (
            llm_response.content
            and llm_response.content.parts
            and any(
                getattr(p, "function_call", None) for p in llm_response.content.parts
            )
        )

        if (
            llm_response.content
            and llm_response.content.parts
            and llm_response.content.parts[0].text
        ):
            part0 = llm_response.content.parts[0]
            if part0.thought:
                self._thought_text += part0.text
            else:
                self._text += part0.text
            llm_response.partial = True
        elif (
            (self._thought_text or self._text)
            and (
                not llm_response.content
                or not llm_response.content.parts
                # don't yield the merged text event when receiving audio data
                or not llm_response.content.parts[0].inline_data
            )
            and not has_function_call
        ):  # Don't yield accumulated thoughts if we have a function call
            parts = []
            if self._thought_text:
                parts.append(types.Part(text=self._thought_text, thought=True))
            if self._text:
                parts.append(types.Part.from_text(text=self._text))
            yield LlmResponse(
                content=types.ModelContent(parts=parts),
                usage_metadata=llm_response.usage_metadata,
            )
            self._thought_text = ""
            self._text = ""

        # If this is a function call response, clear accumulated thoughts/text since it's the final response
        if has_function_call:
            self._thought_text = ""
            self._text = ""
        yield llm_response

    def close(self) -> Optional[LlmResponse]:
        """Generate an aggregated response at the end, if needed.

        This should be called after all the model responses are processed.

        Returns:
          The aggregated LlmResponse.
        """
        if (
            (self._text or self._thought_text)
            and self._response
            and self._response.candidates
        ):
            parts = []
            if self._thought_text:
                parts.append(types.Part(text=self._thought_text, thought=True))
            if self._text:
                parts.append(types.Part.from_text(text=self._text))
            candidate = self._response.candidates[0]
            return LlmResponse(
                content=types.ModelContent(parts=parts),
                error_code=None
                if candidate.finish_reason == types.FinishReason.STOP
                else candidate.finish_reason,
                error_message=None
                if candidate.finish_reason == types.FinishReason.STOP
                else candidate.finish_message,
                usage_metadata=self._usage_metadata,
            )

Submitting them just in case you think it would be valuable for me to introduce this into the new behavior, as this solved the bug we were having of receiving empty answers when reasoning + tools +streaming

[surajksharma07]

@ramcav The fix for this issue already exists in the codebase (streaming_utils.py, commit 786aaed from Nov 26, 2025) as the PROGRESSIVE_SSE_STREAMING feature, but it's disabled by default.

As an immediate workaround, add from google.adk.features import FeatureName, override_feature_enabled and override_feature_enabled(FeatureName.PROGRESSIVE_SSE_STREAMING, True) at the top of your script before creating the agent.

This will properly handle function calls during streaming with output_schema.

The fix handles both streaming function call arguments and preserves the structured output, so your ShoppingList will be populated correctly instead of returning empty.

Team is looking more into it.

In the meantime, this workaround should unblock you completely.

Thanks!

[surajksharma07]

@seanzhou1023 Please have a look into it.

[surajksharma07]

Confirmed resolved in latest ADK (post v1.23.0).

The PROGRESSIVE_SSE_STREAMING feature is now enabled by default (commit 0b1cff29, Dec 19, 2025) — _feature_registry.py:117-119 sets default_on=True.

No manual override_feature_enabled() call is needed anymore.

The StreamingResponseAggregator in streaming_utils.py now properly accumulates streaming function call arguments and handles set_model_response correctly with output_schema.

An additional edge case fix for will_continue flag handling was also merged (commit 2220d885, Feb 3, 2026).