How to access AgentTool's SSE stream?

[nyl199310]

Hi, I found that the SSE stream from the adk api_server only includes FunctionCall and FunctionResponse events.
Is there a way to access the SSE stream for AgentTool as well?
I’d like to receive real-time updates or outputs generated by tools invoked by the agent, not just the function-level events.

Thanks!

[SalvitSimmus7]

Yeah, I have the same issue.

Right now the SSE stream only surfaces top-level FunctionCall/FunctionResponse events. When a parent LLM agent invokes an AgentTool, the sub-agent’s runner is effectively encapsulated—its internal events (thoughts, partial text, tool calls, etc.) aren’t forwarded onto the parent SSE stream from what I've experimented.

Workaround that’s been reliable for me: don’t register the sub-agent as an AgentTool. Instead, wrap it in a plain FunctionTool and manually drive a Runner for the sub-agent inside that function. Consume its async events there, emit whatever logging/logic you want, and return an aggregated result to the parent agent.

Note: FunctionTools can’t return an AsyncGenerator for automatic function calling—if you try, you’ll hit a parsing error "Failed to parse the parameter return_value: AsyncGenerator[str, NoneType] of function...".

So you need to consume the sub-agent stream inside the function and return a final value (string/JSON), and maybe you can yield events to the SSE after the FunctionTool finishes using the FunctionTools response.

from google.adk import types
from google.adk.runner import Runner
from google.adk.tools import FunctionTool
from google.adk.runtime import RunConfig, StreamingMode

async def subagent_as_function_tool(prompt: str) -> str:
    # set up your sub-agent + services here, get from global variables or context variables
    runner = Runner(agent=sub_agent, app_name="my_app", session_service=session_service)
    content = types.Content(role="user", parts=[types.Part(text=prompt)])
    run_cfg = RunConfig(streaming_mode=StreamingMode.SSE, response_modalities=["TEXT"])

    result = []
    async for event in runner.run_async(
        user_id=user_id, session_id=session_id, new_message=content, run_config=run_cfg
    ):
        # Observe sub-agent events here (thoughts, partials, tool calls)
        # and forward lightweight status logs if you want.
        part_texts = [getattr(p, "text", None) for p in getattr(getattr(event, "content", None), "parts", []) or []]
        for t in part_texts:
            if t: result.append(t)

    return "".join(result)

parent_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="parent",
    instruction="...",
    tools=[FunctionTool(subagent_as_function_tool), FunctionTool(other_tool)],
    planner=BuiltInPlanner(thinking_config=types.ThinkingConfig(include_thoughts=True, thinking_budget=-1)),
)

Doing this you get full visibility into the sub-agent’s live events and tool calls, which you can log/trace.

It would be great if AgentTool supported nested event forwarding, so the AgentTool events could appear in the parent stream.

[sandangel]

#3564

Added progressive tool which should help with this issue as well. Please help vote and push this PR so it can be merged soon.

[klateefa]

@nyl199310 Thank you for reporting this. I understand the request is currently under review. Please keep me posted if anything else is needed. Thanks for your patience.

[sasha-gitg]

When you use AgentTool, the sub-agent's intermediate
events (model responses, function calls, etc.) are
consumed internally and only the final text result is
returned to the parent agent. This is by design —
AgentTool wraps an agent as a tool call, so only the
tool output is surfaced.

If you need real-time streaming of events from child
agents in the SSE stream, consider using sub_agents
(agent transfer) instead of AgentTool. With
sub_agents, the child agent's events are yielded
directly through the event stream and will appear in
your SSE connection