No response is received on when transfer_to_agent transfers to a sub agent.

[kmckenna66]

Describe the Bug:
No response is received on when transfer_to_agent transfers to a sub agent using bidi_stream_query and gemini-live-2.5-flash-native-audio.

Steps to Reproduce:
Please provide a numbered list of steps to reproduce the behavior:

1. Connect to agent with fast api configured for bidi stream.
2. Ask "What's the weather?" to initiate transfer to sub agent.
3. Observe no response after transfer.
4. Ask the same question and observe response from sub agent.

Expected Behavior:
After transfer_to_agent is completed the sub agent should respond to the input from the user.

Observed Behavior:
No response is received until additional input is provided.

Environment Details:

• ADK Library Version (pip show google-adk): 2.0.0b1
• Desktop OS: MacOS
• Python Version (python -V): Python 3.13.3

Model Information:

• Are you using LiteLLM: No
• Which model is being used: gemini-live-2.5-flash-native-audio

---

🟡 Optional Information

Providing this information greatly speeds up the resolution process.

Regression:
Did this work in a previous version of ADK? If so, which one? Yes, 1.27.1

Logs:
Please attach relevant logs. Wrap them in code blocks (```) or attach a
text file.
This is what shows when the agent stops after transfer.

2026-04-28 16:41:28,695 - google_adk.google.adk.models.google_llm - INFO - Trying to connect to live model: gemini-live-2.5-flash-native-audio with api backend: GoogleLLMVariant.VERTEX_AI

Screenshots / Video:
If applicable, add screenshots or screen recordings to help explain
your problem.

Additional Context:
This exact agent setup will give a response after transfer_to_agent when using async_stream_query with gemini-2.5-flash instead of bidi_stream_query with gemini-live-2.5-flash-native-audio.

Minimal Reproduction Code:
Please provide a code snippet or a link to a Gist/repo that isolates the issue.

Sample Agent File:

import google.adk.agents as agents
from google.adk.apps import App

agent_2 = agents.Agent(
 name="sample_agent_2",
 model="gemini-live-2.5-flash-native-audio",
 instruction="You are a helpful assistant that can answer questions about the weather.",
 description="A sample agent that can answer questions about the weather.",
)

agent = agents.Agent(
 name="sample_agent",
 model="gemini-live-2.5-flash-native-audio",
 instruction="You are a helpful assistant that can answer questions and help with tasks.",
 description="A sample agent that can answer questions and help with tasks.",
 sub_agents=[agent_2],
)


app = App(
 root_agent=agent,
 name="sample_agent",
)

Agent engine host:

import asyncio
import logging
import os
from typing import Any, AsyncIterator

import vertexai
from google.adk.artifacts import GcsArtifactService, InMemoryArtifactService
from vertexai.agent_engines.templates.adk import AdkApp
from vertexai.preview.reasoning_engines import AdkApp as PreviewAdkApp

try:
 from dotenv import load_dotenv

 load_dotenv()
except ImportError:
 pass

try:
 from .agent import app as adk_app
except ImportError: # pragma: no cover - `python agent_engine_app.py` from cwd
 from agent import app as adk_app

logger = logging.getLogger(__name__)

class AgentEngineApp(AdkApp):
 def set_up(self) -> None:
 try:
 vertexai.init()
 super().set_up()
 except Exception:
 logger.exception("set_up failed")
 raise

 def register_operations(self) -> dict[str, list[str]]:
 operations = super().register_operations()
 operations["bidi_stream"] = ["bidi_stream_query"]
 return operations

 async def bidi_stream_query(
 self, input_queue: asyncio.Queue[dict[str, Any]]
 ) -> AsyncIterator[dict[str, Any]]:
 try:
 async for response in PreviewAdkApp.bidi_stream_query(self, input_queue):
 yield response
 except asyncio.CancelledError:
 raise
 except Exception:
 logger.exception("bidi_stream_query failed")
 raise

agent_engine = AgentEngineApp(
 app=adk_app,
)

How often has this issue occurred?:

• Always (100%)

[yinghsienwu]

Did you see something like "400 INVALID_ARGUMENT. {'error': {'code': 400, 'message': 'gemini-live-2.5-flash-native-audio is not supported in the generateContent API.', 'status': 'INVALID_ARGUMENT'}}" in the session log?

Nvm this is an expected error when using the Agent Engine Playground UI Console, when calling Bidi model.

[kmckenna66]

I don't see any errors in the logs and the session is able to continue with additional input.

[yinghsienwu]

Locally ADK v2.0.0b1 + transfer agents is working properly.

I deployed the transfer agent script above (ADK v2.0.0b1) to Agent Engine. I can get the first response when asking the first weather question, indeed it responds slower than using a single agent. I think it is reasonable. Note that in the script below each user input added time.sleep(0.1) # Small delay for display . This might be helpful in your FastAPI script.

The script below was updated from a github sample. Assuming you have a deployed Agent Engine resource. It can be run successfully in colab to test text-to-audio.

agent_engine_resource = "projects/PROJECT/locations/us-central1/reasoningEngines/ID"

async def weather_chat_with_audio(client, agent_resource_name: str):
 """Interactive chat with the deployed weather agent."""

 def prepare_live_request(input_text: str):
 """Prepare request for ADK agent."""
 part = types.Part.from_text(text=input_text)
 content = types.Content(parts=[part])
 return LiveRequest(content=content)

 async with client.aio.live.agent_engines.connect(
 agent_engine=agent_resource_name,
 config={"class_method": "bidi_stream_query"}
 ) as connection:

 print("🌤️ Weather Assistant Ready!")
 print("Ask about weather in any city, or type 'exit' to quit.\n")
 print("Example questions:")
 print("- What's the weather in San Francisco?")
 print("- Give me a 5-day forecast for Tokyo")
 print("- Is it raining in London?\n")

 user_id = "weather_user"
 first_req = True

 while True:
 # Get user input
 time.sleep(0.1) # Small delay for display
 input_text = input("You: ")

 if input_text.lower() == "exit":
 break

 # Prepare for audio response
 audio_data = []
 seen_audio = False

 # Send request
 if first_req:
 await connection.send({
 "user_id": user_id,
 "live_request": prepare_live_request(input_text).model_dump()
 })
 first_req = False
 else:
 await connection.send(prepare_live_request(input_text).model_dump())

 # Receive response
 print("Weather Assistant: ", end="")

 while True:
 received = await connection.receive()
 event_dict = received["bidiStreamOutput"]

 # Handle the response format
 if "actions" in event_dict and "requested_tool_confirmations" in event_dict["actions"]:
 del event_dict["actions"]["requested_tool_confirmations"]

 event = Event.model_validate(event_dict)

 # Extract content
 if event.content and event.content.parts:
 for part in event.content.parts:
 if part.text:
 print(part.text, end="", flush=True)
 if part.inline_data and part.inline_data.data:
 # Collect audio chunks
 chunk_data = part.inline_data.data
 audio_data.append(np.frombuffer(chunk_data, dtype=np.int16))
 
 # Use turn_complete to determine when to stop receiving for this turn
 if event.turn_complete:
 break

 # Play audio if available
 if audio_data:
 print("\n[Audio response playing...]")
 concatenated_audio = np.concatenate(audio_data)
 display(Audio(concatenated_audio, rate=24000, autoplay=True))
 print("[Audio response played]")

 print() # New line for next interaction

 print("\n👋 Thanks for using Weather Assistant!")

# Run audio chat
await weather_chat_with_audio(client, agent_engine_resource)

[yinghsienwu]

Assume you don't have a successfully deployed Agent Engine. Here's how to deploy one:

"""Google Search Agent definition for ADK Gemini Live API Toolkit demo."""
import asyncio
import logging
import os
from typing import Any, AsyncIterator

import vertexai
from dotenv import load_dotenv
from google.adk.artifacts import GcsArtifactService, InMemoryArtifactService
from google.cloud import logging as google_cloud_logging
from vertexai.agent_engines.templates.adk import AdkApp
from vertexai.preview.reasoning_engines import AdkApp as PreviewAdkApp

from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.models import Gemini
from google.genai import types
from google.adk.tools import google_search
from vertexai._genai.types import AgentServerMode

import google.auth

_, project_id = google.auth.default()
os.environ["GOOGLE_CLOUD_PROJECT"] = project_id
os.environ["GOOGLE_CLOUD_LOCATION"] = "us-central1" # or your location, not 'global'
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

client = vertexai.Client(project=project_id, location="us-central1") # or your location, not 'global'

logger = logging.getLogger(__name__)

agent_2 = Agent(
 name="sample_agent_2",
 model="gemini-live-2.5-flash-native-audio",
 instruction="You are a helpful assistant that can answer questions about the weather.",
 description="A sample agent that can answer questions about the weather.",
)

root_agent = Agent(
 name="sample_agent",
 model="gemini-live-2.5-flash-native-audio",
 instruction="You are a helpful assistant that can answer questions and help with tasks.",
 description="A sample agent that can answer questions and help with tasks.",
 sub_agents=[agent_2],
)


app = App(
 root_agent=root_agent,
 name="app",
)

class AgentEngineApp(AdkApp):
 def set_up(self) -> None:
 try:
 vertexai.init()
 super().set_up()
 except Exception:
 logger.exception("set_up failed")
 raise

 def register_operations(self) -> dict[str, list[str]]:
 operations = super().register_operations()
 operations["bidi_stream"] = ["bidi_stream_query"]
 return operations

 async def bidi_stream_query(
 self, input_queue: asyncio.Queue[dict[str, Any]]
 ) -> AsyncIterator[dict[str, Any]]:
 try:
 async for response in PreviewAdkApp.bidi_stream_query(self, input_queue):
 yield response
 except asyncio.CancelledError:
 raise
 except Exception:
 logger.exception("bidi_stream_query failed")
 raise

logs_bucket_name = os.environ.get("LOGS_BUCKET_NAME")
agent_engine = AgentEngineApp(
 app=app,
 artifact_service_builder=lambda: (
 GcsArtifactService(bucket_name=logs_bucket_name)
 if logs_bucket_name
 else InMemoryArtifactService()
 ),
)
# Enable telemetry by default for Agent Engine
env_vars = {
 "GOOGLE_CLOUD_AGENT_ENGINE_ENABLE_TELEMETRY": "true",
 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "true",
}


# Deploy to Agent Engine
remote_live_adk_agent = client.agent_engines.create(
 agent=agent_engine,
 config={
 "display_name": "e2e_deploy_bidi_demo_transfer_agent",
 "env_vars": env_vars,
 "description": "Simple agent with real-time audio capabilities",
 "requirements": [
 "google-adk==2.0.0b1",
 "google-genai",
 "google-cloud-aiplatform[agent_engines] @ git+https://github.com/googleapis/python-aiplatform.git",
 "cloudpickle",
 "pydantic"
 ],
 "staging_bucket": f"gs://{BUCKET}", # your bucket
 # Set the experimental mode
 "agent_server_mode": AgentServerMode.EXPERIMENTAL
 },
)

Make sure you use python < 3.14.
$export GOOGLE_GENAI_USE_VERTEXAI=true
And (GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, etc)
$gcloud auth application-default login

In my_bidi directory:
$uv add google-adk==2.0.0b1 --prerelease=allow
$uv run -m app.e2e_bidi_agent_deploy

Then you'll create the reasoningEngine (AgentEngine, aka AgentRuntime) resource. Use the script in previous comment to call your remote bidi agent in colab text-in-audio-out.

[smwitkowski]

Thanks for the sample and deployment steps. We followed up on the client-delay angle. Adding a small delay did not resolve the Ruby handoff hang, so we dug into the ADK live transfer boundary.

We now have a local no-network harness that exercises ADK's actual BaseLlmFlow.run_live transfer path. It stubs model I/O and controls whether the parent InvocationContext has a Live resumption handle before transfer_to_agent.

Findings:

1. If the parent transfers before any Live resumption handle update, the child starts cleanly:

   • child live_session_resumption_handle = None
   • child run_config.session_resumption.handle = None
   • child connect sends no resume handle
   • history is sent

2. If the parent has already received a Live resumption handle before transfer_to_agent, the child inherits the parent handle:

   • child live_session_resumption_handle = <parent handle>
   • child first LiveConnectConfig sends that handle with transparent=True
   • history is not sent

That looks consistent with the observed behavior: the parent transfer function response is emitted, the parent Live connection is closed, then the child appears to connect as if it is transparently resuming the parent Live session instead of starting a new child Live session. The session can then continue after another user input because that starts a fresh turn/session path.

One extra detail: after simulating same-agent reconnect before transfer, the same stale handle can also be present in invocation_context.run_config.session_resumption.handle. Clearing only invocation_context.live_session_resumption_handle fixes the simple inherited-handle case, but clearing both locations is needed for the reconnect-before-transfer case.

We tested an app-level workaround that clears the inherited Live resumption handle before delegated child run_live starts. The safer version clears both:

• invocation_context.live_session_resumption_handle
• invocation_context.run_config.session_resumption.handle

This should be scoped to agent transfer only. We are not suggesting disabling normal same-agent session resumption.

Suggested upstream fix boundary: in BaseLlmFlow.run_live, inside the transfer_to_agent handling block, after the parent llm_connection.close() and before agent_to_run.run_live(invocation_context), reset Live transport/resumption state for the child run, or create a child invocation context that does not carry the parent Live session resumption handle.

Conceptually:

# In BaseLlmFlow.run_live, inside the transfer_to_agent block:
# after closing the parent live connection, before starting the child agent.

await llm_connection.close()

transfer_to_agent = event.actions.transfer_to_agent
if transfer_to_agent:
 agent_to_run = self._get_agent_to_run(invocation_context, transfer_to_agent)

 # Child Live agent should start a new Live transport session.
 # Do not reuse the parent Live socket's resumption handle.
 invocation_context.live_session_resumption_handle = None

 session_resumption = None
 if invocation_context.run_config:
 session_resumption = invocation_context.run_config.session_resumption
 if session_resumption:
 session_resumption.handle = None

 async with Aclosing(agent_to_run.run_live(invocation_context)) as agen:
 async for item in agen:
 yield item

We have a no-network harness that stubs the parent/child LLM connections and exercises ADK's real BaseLlmFlow.run_live transfer block. Happy to share that as a gist if useful; we avoided pasting the full harness here to keep the issue readable.

Question: are Live session resumption handles intended to be scoped to one Live model connection/invocation? If so, the child agent should not inherit the parent handle across transfer_to_agent.

[smwitkowski]

Follow-up: here is the no-network harness referenced above:

https://gist.github.com/smwitkowski/877a54b458b9fe63ef885f8a645b6f0d

It includes a README plus a single Python script. The script can be run without Gemini, Vertex AI, or Agent Engine. It stubs the parent and child LLM connections, then exercises ADK's real BaseLlmFlow.run_live transfer path to show how the child LiveConnectConfig is constructed across transfer_to_agent.

Example run command:

uv run \
 --with google-adk==1.32.0 \
 --with google-genai \
 --with websockets \
 python adk_live_transfer_resumption_repro.py

The key baseline rows are:

transfer_after_handle_update | none | child_connect_handle=PARENT_HANDLE | history_sent=no
transfer_after_same_agent_reconnect | none | child_connect_handle=PARENT_HANDLE | history_sent=no
transfer_after_same_agent_reconnect | context_only | child_run_config_handle=PARENT_HANDLE | child_connect_handle=PARENT_HANDLE
transfer_after_same_agent_reconnect | complete | child_connect_handle=- | history_sent=yes

So the repro is specifically isolating the inherited parent Live resumption handle at the child-transfer boundary, including the extra run_config.session_resumption.handle case after a same-agent reconnect.

[yinghsienwu]

Thanks for the context. I understand the issue now. This issue happens regardless in ADK v1.32 or v2.0.0b1. By default, ADK run_live turns on session resumption, but the invocation_context of the parent agent, will be transferred to child agents. If parent has already received a Live resumption handle before transfer_to_agent, the child inherits the parent handle and will be interrupted. The child will need another question to continue the conversation.

I'm thinking what would be the risk if transfer_to_agent w/o inheriting the parent. It seems reasonable. But the parent session history won't contain children's at the Live API level (if they are decoupled). I'm thinking if there are alternative solution to fix the issue w/o changing the session handle inheritance.

[yinghsienwu]

However, the parent agent already delegates sub topics to children agents, which maintain its own context (would resume its own session until it delegates to grandchildren agents, if any). So it should be fine. The entire conversation (regardless how the agent graph looks like) is still preserved in SessionService. I think it makes sense.

[smwitkowski]

Thanks, Amy. This matches our understanding as well, and I agree with the direction.

The important distinction seems to be between conversation/session history and Live transport state:

• conversation history should continue to be preserved by SessionService across the whole agent graph
• each delegated Live agent should start with its own Live transport/resumption state
• the child should not transparently resume the parent agent’s Live model connection

That makes the behavior predictable: same-agent reconnect can still use Live session resumption, but transfer_to_agent should create a clean Live connection boundary for the child agent.

One implementation detail from the harness: clearing only invocation_context.live_session_resumption_handle fixes the simple inherited-handle case, but after a same-agent reconnect the stale parent handle can also be present in invocation_context.run_config.session_resumption.handle. So the transfer fix likely needs to avoid carrying both values into the child run.

Conceptually, either of these should work:

1. clear the Live resumption fields before starting the child run_live, scoped only to transfer_to_agent
2. create a child invocation/run config that preserves SessionService-backed conversation state but does not inherit the parent Live connection resumption handle

The second option may be cleaner if ADK wants to avoid mutating the parent invocation context directly, but the required behavior is the same: the child Live agent should not send the parent handle in its first LiveConnectConfig.

We’re happy to validate a PR or branch against the no-network harness and the system reproduction once there is a candidate fix.