How to Access LiveRequestQueue Outside WebSocket Context

[S4lem]

I’m working with Google ADK using FastAPI, and I'm combining two types of agents in the same application:

• A RootAgent, implemented as a CustomAgent, which orchestrates a workflow of multiple regular agents (run_async path).
• A LiveAgent acting as a sub-agent, but not part of the workflow.

The LiveAgent is triggered inside a WebSocket endpoint following the ADK documentation:

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: int, is_audio: str):
  # ...
  live_request_queue = LiveRequestQueue()
  # ...
  runner.run_live(..., live_request_queue=live_request_queue)

This works correctly, and I am able to talk to the live agent through audio streaming.

The RootAgent workflow can be triggered from an HTTP endpoint /trigger, where I call:

@app.post("/trigger")
async def trigger_workflow(
  # ...
  runner.run_async(...)

During this workflow, some agents produce results that I would like to send to the LiveAgent so that the live agent stays up-to-date and can talk about those results to the end-user.

ADK allows sending messages to a live agent via:

content = types.Content(parts=[types.Part(text="some text")])
live_request_queue.send_content(content)

However, the live_request_queue only exists inside the WebSocket handler, and I cannot access it from the /trigger endpoint or from inside the workflow.

• What is the recommended ADK-compliant way to communicate from a workflow agent (running under run_async) to a LiveAgent running under run_live?
• Is there an official mechanism to allow a regular agent to “inform” a live agent about new results during execution?

[adk-bot]

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

TLDR: The recommended approach is to maintain a shared dictionary at the application level that maps a unique identifier (like a user_id) to its corresponding LiveRequestQueue instance. Your WebSocket endpoint will add the queue to this dictionary upon connection and remove it upon disconnection. Your HTTP endpoint can then access this dictionary to retrieve the queue and send messages to the LiveAgent.

---

Hello! That's an excellent question about how to architect communication between different agent types in ADK. Here’s a breakdown of the recommended approach.

The Challenge: Scoped vs. Shared State

As you've correctly identified, the LiveRequestQueue is created within the scope of your WebSocket handler. It is tied to a specific run_live session and isn't directly accessible from other parts of your application, like your /trigger HTTP endpoint where run_async is called.

ADK doesn't provide a built-in function to directly look up a LiveRequestQueue from a different execution context. Instead, this is a common application architecture challenge that can be solved using standard Python and FastAPI patterns. The solution, as demonstrated in the ADK documentation, is to manage these queues explicitly. [1]

The Solution: A Shared Queue Manager

You can implement a simple in-memory "manager" (using a Python dictionary) to keep track of the active LiveRequestQueue for each user.

Here is a step-by-step guide:

Step 1: Create a Shared Dictionary

In your main application file, create a global dictionary to store the queues.

from fastapi import FastAPI
from adk.live import LiveRequestQueue

app = FastAPI()

# A dictionary to map a user_id to their active LiveRequestQueue
active_live_queues: dict[str, LiveRequestQueue] = {}

Step 2: Manage the Queue in Your WebSocket Endpoint

In your WebSocket handler, add the user's queue to the dictionary when they connect and, most importantly, remove it when they disconnect. Using a try...finally block is a robust way to ensure cleanup happens even if errors occur.

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
  # ...
  live_request_queue = LiveRequestQueue()
  active_live_queues[user_id] = live_request_queue
  
  try:
    # The run_live method will block here until the connection is closed
    await runner.run_live(..., live_request_queue=live_request_queue)
  finally:
    # Once the connection is closed, remove the queue from the dictionary
    del active_live_queues[user_id]
    print(f"Cleaned up queue for user {user_id}")

Step 3: Access the Queue from Your HTTP Endpoint

Now, your /trigger endpoint (or any other agent or tool in your run_async workflow) can access the active_live_queues dictionary to get the queue for a specific user and send content to it.

from adk.api import types

@app.post("/trigger")
async def trigger_workflow(user_id: str, message: str):
  # 1. Trigger your RootAgent's async workflow
  # runner.run_async(...)
  # ...imagine the workflow runs and produces a result...
  
  # 2. Inform the LiveAgent about the result
  if user_id in active_live_queues:
    live_queue = active_live_queues[user_id]
    
    # Create the content to send
    content = types.Content(parts=[types.Part(text=f"An update from the workflow: {message}")])
    
    # Send the content to the LiveAgent
    live_queue.send_content(content)
    return {"status": "Message sent to LiveAgent"}
  else:
    # Handle the case where the user is not connected via WebSocket
    return {"status": f"User {user_id} not connected"}

This pattern ensures that you can safely access the communication channel to the LiveAgent from anywhere in your application, bridging the gap between your run_async workflow and your run_live streaming session.

For more details you can check out the "Web Application" section in the ADK documentation, which describes a similar mechanism for sending events to a live agent. [1]

---

[1] https://google.github.io/adk-docs/streaming/custom-streaming/
[2] https://google.github.io/adk-docs/streaming/dev-guide/part2/