how to get streaming text response.

[parthiaitech]

I'm trying to get the text streaming response. but this gives the audio format streaming. how to get text streaming responses?

import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event
from google.adk.agents import LiveRequestQueue
from dotenv import load_dotenv
load_dotenv()
import os

os.environ["GOOGLE_API_KEY"] = "AIzaSyB3FhuL0UUSXaZeTBAxaEX9fIFM7r7lI6M"

agent = Agent(
name="streaming_agent",
model="gemini-2.0-flash-live-001",
description="A simple streaming agent example",
instruction="Respond to user queries in a streaming manner."
)

live_request_queue = LiveRequestQueue()

async def main():
session_service = InMemorySessionService()
user_id = "user123"

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

session = await session_service.create_session(
    app_name="streaming_agent_app",
    user_id=user_id
)

user_input = types.Content(role="user", parts=[types.Part(text="Hello, Tell me about best beach in america in a five sentence.")])


await session_service.append_event(session, Event(author="user", content=user_input))

async for event in runner.run_live(user_id=user_id, session_id=session.id, live_request_queue=live_request_queue):
    if event.content and event.content.parts:
        print(event.content.parts)
        print(event.content.parts[0].text, end="", flush=True)
print("\nStreaming complete.")

asyncio.run(main())

Output:
[Part(
inline_data=Blob(
data=b'\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00...',
mime_type='audio/pcm;rate=24000'
)
)]
None[Part(
inline_data=Blob(
data=b'\x05\xf7\xa8\xfe\xa4\x05\xe0\x01\xd0\x01K\xfeS\xf4\xe6\nl\x06:\xf6\xd8\xfb\xf2\x04\x1d\x02\xc8\x02\x16\xfe\xda\xf3\xb9\x04\xc4\x0b\xe1\xfe\x8a\xf7\x92\xf9M\x05\xa4\n\xa3\xfb\x9a\xf7\x06\xff\x84\x04O\x03\xfe\xfc\x7f\x004\xf9=\x020\x06\xf0\xfe\xf0\xf8\xfd\x01\x9c\x02\xab\xfd0\x02\xbc\xf3\x83\t<\x05\xf6\xf3r\x06\x94\xf9\xf5\x02q\xfe,\x05|...',
mime_type='audio/pcm;rate=24000'
)
)]
None

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: You are receiving audio output because "live" agents in ADK default to an audio modality. To receive text, you must explicitly configure the agent to output text by passing a RunConfig object to your runner.run_live call.

---

Hello! I can help with your question.

The behavior you are observing is expected. The gemini-2.0-flash-live-001 model is a "live" agent, and these agents are optimized for real-time, spoken conversations. By default, they produce an audio stream (audio/pcm) as output. [2]

To get a streaming text response, you need to configure the run to expect the "TEXT" modality. You can do this by creating a RunConfig object and setting the response_modalities to ["TEXT"]. Then, pass this configuration object to your runner.run_live method. [1]

Here is the corrected version of your code:

import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event
from google.adk.agents import LiveRequestQueue
from google.adk.agents.run_config import RunConfig  # Import RunConfig
from dotenv import load_dotenv
load_dotenv()
import os

os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY" # It is recommended to use environment variables for API keys

agent = Agent(
    name="streaming_agent",
    model="gemini-2.0-flash-live-001",
    description="A simple streaming agent example",
    instruction="Respond to user queries in a streaming manner."
)

live_request_queue = LiveRequestQueue()

async def main():
    session_service = InMemorySessionService()
    user_id = "user123"

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

    session = await session_service.create_session(
        app_name="streaming_agent_app",
        user_id=user_id
    )

    user_input = types.Content(role="user", parts=[types.Part(text="Hello, Tell me about best beach in america in a five sentence.")])

    await session_service.append_event(session, Event(author="user", content=user_input))

    # Configure the run to expect TEXT responses
    run_config = RunConfig(response_modalities=["TEXT"])

    # Pass the run_config to the runner
    async for event in runner.run_live(user_id=user_id, session_id=session.id, live_request_queue=live_request_queue, run_config=run_config):
        if event.content and event.content.parts and event.content.parts[0].text:
            print(event.content.parts[0].text, end="", flush=True)
    print("\nStreaming complete.")


asyncio.run(main())

By making this change, the agent will know to generate text instead of audio, and you will get the streaming text response you're looking for.

---

This response was generated by an automated agent.

[1] https://google.github.io/adk-docs/runtime/runconfig/
[2] https://github.com/google/adk-python/blob/main/src/google/adk/agents/run_config.py