Support out-of-band progress callbacks for streaming tools in live mode

[kazunori279]

Description

Related: #5947 (streaming tool yields cause re-invocation loop)

Issue #5947 addresses the immediate bug where streaming tool yields interrupt the model via turn_complete=True. The turn_complete=False fix (in progress) solves the re-invocation loop, but intermediate yields still land in the model's context window — even when they're purely presentational status updates the model doesn't need to reason about.

This feature request proposes a complementary mechanism: an out-of-band progress callback that lets streaming tools report status directly to the application layer, bypassing the model's context entirely.

Motivation

Many streaming tool use cases involve two distinct types of intermediate output:

Type	Example	Model needs it?
Context-relevant	"Found 3 matching documents"	Yes — model should narrate or reason about it
Presentational	"Step 2/5: processing...", spinner updates	No — only the UI needs it

After #5947's turn_complete=False fix, context-relevant yields work correctly — the model receives them without interruption. But presentational yields still consume context window space unnecessarily and can accumulate noise over long-running tools.

An out-of-band callback gives tool authors explicit control: yield for model-facing updates, progress_callback() for UI-only status.

Proposed Design

1. Parameter injection (following ToolContext pattern)

FunctionTool already detects tool_context in the function signature via find_context_parameter() and injects it at call time (function_tool.py:88). A progress_callback would follow the same pattern:

async def long_running_task(
    query: str,
    tool_context: ToolContext,
    progress_callback: Callable[[Any], Awaitable[None]] | None = None,
):
    """A tool that takes time and reports progress."""
    # OOB: goes directly to the app layer, not to the model
    if progress_callback:
        await progress_callback({"status": "running", "message": "Step 1: fetching data..."})
    
    await asyncio.sleep(5)
    
    if progress_callback:
        await progress_callback({"status": "running", "message": "Step 2: processing..."})
    
    await asyncio.sleep(5)
    
    # Model-facing: returned as the tool result for the model to narrate
    return {"status": "completed", "result": "Done!"}

For async generator (streaming) tools, both mechanisms coexist:

async def long_running_task(
    query: str,
    progress_callback: Callable[[Any], Awaitable[None]] | None = None,
):
    # OOB progress — UI only
    if progress_callback:
        await progress_callback({"step": 1, "message": "Searching..."})
    
    results = await search(query)
    
    # Model-facing yield — model can narrate "I found 3 results"
    yield {"status": "found", "count": len(results)}
    
    if progress_callback:
        await progress_callback({"step": 2, "message": "Analyzing..."})
    
    analysis = await analyze(results)
    yield {"status": "completed", "analysis": analysis}

2. Injection point

In FunctionTool._call_live() (function_tool.py:320-321), where tool_context is already injected for async generator tools:

async def _call_live(self, *, args, tool_context, invocation_context) -> Any:
    args_to_call = args.copy()
    signature = inspect.signature(self.func)
    
    if 'tool_context' in signature.parameters:
        args_to_call['tool_context'] = tool_context
    
    # New: inject progress_callback if the tool declares it
    if 'progress_callback' in signature.parameters:
        args_to_call['progress_callback'] = self._make_progress_callback(
            invocation_context, tool_context
        )
    
    async with Aclosing(self.func(**args_to_call)) as agen:
        async for item in agen:
            yield item

Same injection in run_async() for non-generator tools.

3. Callback routing

The bound callback routes progress to a handler registered on the runner or invocation context — never entering live_request_queue:

def _make_progress_callback(self, invocation_context, tool_context):
    async def callback(data: Any) -> None:
        if invocation_context.progress_handler:
            await invocation_context.progress_handler(
                tool_name=self.name,
                function_call_id=tool_context.function_call_id,
                data=data,
            )
    return callback

4. Runner-level handler registration

runner = Runner(
    agent=agent,
    app_name="my_app",
    session_service=session_service,
)

# Application registers a handler for OOB progress
runner.on_tool_progress = my_progress_handler

async def my_progress_handler(tool_name: str, function_call_id: str, data: Any):
    # Route to WebSocket, SSE, logging, etc.
    await websocket.send_json({
        "type": "tool_progress",
        "tool": tool_name,
        "data": data,
    })

Surface area

File	Change
InvocationContext	Add progress_handler: Optional[Callable] field
FunctionTool._call_live()	Detect and inject progress_callback
FunctionTool.run_async()	Same injection for non-generator tools
FunctionTool._ignore_params	Add 'progress_callback' to exclude from function declaration
Runner	Add on_tool_progress registration, thread handler into InvocationContext

Non-goals

• This does not replace turn_complete=False from Streaming tool yields in live mode cause model to re-invoke the tool in a loop #5947 — both mechanisms serve different purposes and coexist
• This does not change the behavior of yield in streaming tools — yields remain model-facing

Environment

• google-adk: 2.1.0
• Related: Streaming tool yields in live mode cause model to re-invoke the tool in a loop #5947

[adk-bot]

I will read the instruction file for the adk-issue-analyze skill to understand the required steps for analyzing the GitHub issue.I will fetch the details of GitHub issue #5979 using the provided fetch_github_issue tool.I will search for any existing pull requests related to issue #5979 in the repository to check if there is an active implementation underway.I will check if there is an active PR or issue mentioning the related issue #5947 to gain context on the "streaming tool yields cause re-invocation loop" bug and its fix.I will search for the relevant source files (function_tool.py, files containing InvocationContext, and files containing the Runner class/definition) in the local repository to examine their implementations and verify the feasibility of the proposed design.I will search for function_tool.py and other core components of the repository using grep_search or find_by_name to locate where the tool framework and runners are defined.I will perform a grep search to find references to InvocationContext and locate its definition.I will search for the definition of the Runner class as well.I will perform a grep search in src/google/adk/tools/function_tool.py to target the exact lines where tool_context is reference-checked and used, so we can see how the injection currently works.I will view the first 400 lines of src/google/adk/tools/function_tool.py to examine the tool signature parsing and injection mechanisms.I will view the invocation_context.py file to see the fields of InvocationContext and understand how it gets initialized.I will view the rest of representation in invocation_context.py to check for details on initialization and other fields.I will perform a grep search for InvocationContext( to see how it is instantiated and where we might pass the progress_handler into it.I will perform a line-by-line grep search for InvocationContext in src/google/adk/runners.py to identify exactly how and where the runner creates the context.I will view lines 2010 to 2040 in src/google/adk/runners.py to see how _create_invocation_context is implemented and how kwargs are passed.I will view lines 2041 to 2090 in src/google/adk/runners.py to see the full implementation of _new_invocation_context.I will view lines 2091 to 2140 in src/google/adk/runners.py to see the complete implementation of _new_invocation_context_for_live.I will search for the constructor (__init__) and standard properties of the Runner class in src/google/adk/runners.py to understand where user handlers and configurations are registered.I will view the Runner.__init__ constructor and properties from line 190 to 250 in src/google/adk/runners.py.I will view src/google/adk/tools/tool_context.py to check what attributes are available on ToolContext, specifically looking for references to the InvocationContext or other session information that might help access the progress handler.I will view src/google/adk/agents/context.py to see what attributes are available on Context, such as invocation_context.I will search for the definition of ReadonlyContext or references to invocation_context in readonly_context.py to confirm how the underlying InvocationContext is stored/exposed on Context (also referred to as ToolContext).I will view lines 25 to 60 of src/google/adk/agents/readonly_context.py to check if _invocation_context has a property/getter or if it is private and how we can access it.I will view the unit tests for FunctionTool in tests/unittests/tools/test_function_tool.py to see how they verify tool behavior, so we can recommend precise testing updates.I have completed a thorough investigation and code analysis for the requested issue. Below is the detailed analysis report prepared in accordance with the adk-issue-analyze guidelines.

---

GitHub Issue #5979 Analysis: Support out-of-band progress callbacks for streaming tools in live mode

Executive Summary

1. What is broken? There is no mechanism in FunctionTool to allow streaming/generator tools to send out-of-band (OOB) progress updates directly to the application layer. Currently, intermediate tool outputs must be generated via standard yield, which injects those updates directly into the model's chat context. This causes unnecessary context consumption (noise/token inflation) for presentational/UI-only updates (like spinners or process status indicators) that the model does not need to reason about.
2. Is there a linked PR that fixes this issue? None.
3. Recommendation: Should Fix (Medium/Low Priority). Implementing this feature introduces high-quality out-of-band visibility for long-running tools, perfectly matching real-world UI requirements while preserving the LLM's valuable context window. The implementation cost and architectural risk are minimal, following the existing tool_context parameter injection pattern.

Detailed Analysis

1. Root Cause & Design Analysis ("What is broken?")

The core issue is that intermediate feedback from a streaming tool is forced to go through the LLM. In typical application patterns, tools can generate both context-relevant statements (which the LLM must consume/understand, e.g. "Fetched 3 items") and presentational updates (which only the client UI needs to show, e.g. "Querying partition: 2/5...").

Currently, all generator outputs in FunctionTool are piped directly from tool yields/returns into the LLM context flow. To support clean OOB presentational progress updates, we need a parameter injection flow analogous to the existing ToolContext setup.

Technical Analysis & Affected Source Files:

• FunctionTool Parameter Detection:
  In the FunctionTool.__init__ of function_tool.py, the find_context_parameter utility identifies parameters to suppress from the model function definitions. We should append "progress_callback" to self._ignore_params dynamically or declare it of a special ignored type so it doesn't get published into the LLM schema:

  self._ignore_params = [self._context_param_name, 'input_stream', 'progress_callback']

• Sync / Async Tool Execution & Injection:
  In FunctionTool.run_async (function_tool.py), if "progress_callback" is declared in the function's signature, we need to bind and inject a custom progress_callback leveraging the underlying InvocationContext:

  if 'progress_callback' in signature.parameters:
    args_to_call['progress_callback'] = self._make_progress_callback(
        tool_context._invocation_context, tool_context
    )

• Live Tool Execution & Injection:
  In FunctionTool._call_live (function_tool.py), the same detection and injection pattern needs to occur where the generator tool runs in live mode:

  if 'progress_callback' in signature.parameters:
    args_to_call['progress_callback'] = self._make_progress_callback(
        invocation_context, tool_context
    )

• Invocation Context Changes:
  In InvocationContext (invocation_context.py), we should declare a progress_handler field of type Optional[Callable[[str, str, Any], Awaitable[None]]] with a default value of None:

  progress_handler: Optional[Callable[[str, str, Any], Awaitable[None]]] = Field(
      default=None, exclude=True
  )

• Runner Configuration:
  In class Runner (runners.py), expose the registration hook:

  self.on_tool_progress: Optional[Callable[[str, str, Any], Awaitable[None]]] = None

  And thread this function pointer down to the context instantiation in _new_invocation_context (runners.py):

  return self._create_invocation_context(
      ...
      progress_handler=self.on_tool_progress,
  )

---

2. Existing Pull Requests ("Is there a linked PR that fixes this issue?")

• Linked PR: None.
• PR URL: N/A
• Analysis: No active pull requests or open branches currently implement this feature.

---

3. Recommendation

• Recommendation: Should Fix (Medium/Low Priority)
• Rationale:
  • Developer Experience: Significantly improves UX and developers' ability to build sophisticated progress tracking (e.g. web application socket progress reporting) for long-running generative processes without contaminating LLM context blocks.
  • Risk Assessment: Incredibly low risk. It follows exact patterns constructed for parameter injection, is backwards-compatible, and is strictly opt-in (only activated when tools explicitly claim progress_callback in their signatures).

---

Summary of Work Done

• Successfully retrieved details of GitHub issue Support out-of-band progress callbacks for streaming tools in live mode #5979 using fetch_github_issue.
• Handled command constraints and queried GitHub Pull Requests using gh pr list for references to Support out-of-band progress callbacks for streaming tools in live mode #5979/related issue Streaming tool yields in live mode cause model to re-invoke the tool in a loop #5947.
• Explored relevant layout/core orchestration files of the framework: function_tool.py, invocation_context.py, runners.py.
• Provided a highly structured, comprehensive, and clean analysis with explicit line references and code-snippet designs aligning with ADK conventions.