Build a custom WebSocket server for a live agent

A browser speaks your application’s WebSocket protocol, while ADK speaks to the Gemini Live service. Build a FastAPI bridge with two concurrent loops: one receives microphone frames from the client, and the other forwards model audio and control events back. When either loop finishes, the handler must stop its sibling and release the admitted session.

The client sends PCM audio as binary frames and receives audio frames plus JSON control events. The saved connections on gemini-2.5-flash-native-audio-preview-12-2025 answered the microphone-rate question, refused an over-capacity caller and released the handler after an idle hang-up. Local tests also check duplicate-session refusal and setup failures.

When you need this. Use a custom server when your docs client needs its own audio playback, captions and connection controls. Two concurrent loops let the caller keep sending input while the agent speaks.

Download the complete lab ZIP, extract it and open a terminal in adk-live-server/. Create the environment below and set GOOGLE_API_KEY in that terminal before running a live command.

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt

1. Build one runner and two connection tasks

create_app builds one Runner at application startup and creates a new LiveRequestQueue for each WebSocket connection. The in-memory session service keeps this single process’s conversation history; use a database-backed session service when state must survive restarts.

"""A FastAPI WebSocket bridge for a live agent: binary audio frames in and out, JSON events, a session cap."""
import asyncio
import anyio
import logging
from pathlib import Path

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from google.adk.agents import Agent, LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from live_harness import MODEL

APP = "voice_docs"
DOCS = (Path(__file__).parent / "docs/live-streaming.md").read_text()
log = logging.getLogger("voice_docs.server")


def build_agent(model=MODEL) -> Agent:
    return Agent(name="docs_helper", model=model, instruction=(
        "You are the docs assistant. Answer only from this page in one or two short spoken sentences.\n\n" + DOCS))


def create_app(agent: Agent, max_sessions: int = 2) -> FastAPI:
    app = FastAPI()
    runner = Runner(app_name=APP, agent=agent, session_service=InMemorySessionService())  # once, at startup
    active: set[tuple[str, str]] = set()
    app.state.active = active  # sessions with a running handler

    @app.websocket("/ws/{user_id}/{session_id}")
    async def live(websocket: WebSocket, user_id: str, session_id: str) -> None:
        await websocket.accept()
        key = (user_id, session_id)
        if key in active:
            await websocket.close(code=1008, reason="session already has a live connection")
            return
        if len(active) >= max_sessions:  # refuse here, before the Live API quota refuses for you
            await websocket.close(code=1013, reason="voice sessions full, try again shortly")
            return
        active.add(key)
        queue = LiveRequestQueue()  # one per connection, never reused
        run_config = RunConfig(response_modalities=["AUDIO"], session_resumption=types.SessionResumptionConfig())

        async def upstream() -> None:
            while True:
                message = await websocket.receive()
                if message["type"] == "websocket.disconnect":
                    raise WebSocketDisconnect(message.get("code", 1000))
                if message.get("bytes"):  # 16 kHz PCM from the microphone
                    queue.send_realtime(types.Blob(mime_type="audio/pcm;rate=16000", data=message["bytes"]))
                elif message.get("text"):
                    queue.send_content(types.Content(role="user", parts=[types.Part(text=message["text"])]))

        async def downstream() -> None:
            async for event in runner.run_live(user_id=user_id, session_id=session_id,
                                               live_request_queue=queue, run_config=run_config):
                parts = event.content.parts if event.content else []
                for part in parts:
                    if part.inline_data:  # 24 kHz PCM as a binary frame: no base64 inflation
                        await websocket.send_bytes(part.inline_data.data)
                if not (parts and all(p.inline_data for p in parts)):
                    await websocket.send_text(event.model_dump_json(
                        exclude_none=True, by_alias=True, exclude={"content": {"parts": {"__all__": {"inline_data"}}}}))

        tasks = []
        try:
            if not await runner.session_service.get_session(app_name=APP, user_id=user_id, session_id=session_id):
                await runner.session_service.create_session(app_name=APP, user_id=user_id, session_id=session_id)
            tasks = [asyncio.create_task(upstream()), asyncio.create_task(downstream())]
            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            for task in pending:
                task.cancel()
            failure = next((t.exception() for t in done if t.exception()
                            and not isinstance(t.exception(), WebSocketDisconnect)), None)
            if failure:  # tell the client why, instead of a silent close
                log.error("live session %s failed: %r", session_id, failure)
                await websocket.close(code=1011, reason=f"{type(failure).__name__}: {failure}"[:120])
        finally:
            queue.close()  # always: an unclosed queue can strand a Live API session against your quota
            for task in tasks:
                if not task.done():
                    task.cancel()
            active.discard(key)
            # ASGI cancellation must not interrupt the cleanup of sibling tasks.
            with anyio.CancelScope(shield=True):
                await asyncio.gather(*tasks, return_exceptions=True)

    return app


app = create_app(build_agent())

asyncio.create_task schedules both loops before the handler waits for either one. This is concurrency through I/O: a task that awaits a socket or ADK event lets the other task run. The caller can interrupt ongoing speech because receiving a new microphone frame does not wait for the previous spoken response to finish.

The upstream task reads client frames and sends text or PCM to the queue. The downstream task consumes run_live, sends audio bytes directly as binary WebSocket frames, and serializes the remaining event fields as JSON with camelCase aliases such as turnComplete (excluding inline audio so the same bytes are not sent twice).

Sending raw binary frames avoids base64 size overhead on audio chunks:

{
  "frames": {
    "binary": 68,
    "text": 23
  },
  "audio_bytes": 174720,
  "base64_bytes": 232960
}

A standard model is configured. When the underlying Live API connection fails at startup, the bridge catches the failure and closes the WebSocket with code 1011. Choose a model that supports bidirectional streaming (run_live). Check: test_a_model_failure_closes_with_1011_and_the_reason.

2. Stop both directions when one ends

A naive bridge that runs upstream() and downstream() under asyncio.gather can hang when an idle client disconnects: upstream() catches WebSocketDisconnect and returns normally, while gather keeps waiting on downstream(), which never sees queue.close().

Instead, server.py waits on both tasks with asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED). As soon as either direction finishes, the handler cancels the pending sibling and checks the finished task for errors. The finally block closes the LiveRequestQueue, cancels unfinished tasks and releases the active-session slot. It then awaits both tasks with gather(..., return_exceptions=True) inside AnyIO’s CancelScope(shield=True), so ASGI cancellation cannot interrupt that wait. Putting session lookup inside the try block ensures that even a session-store setup error releases the admitted slot.

Client hangs up while downstream is idle. Running the two loops under asyncio.gather leaves the handler open when the client disconnects during an idle pause. Wait for FIRST_COMPLETED, close the queue and await the canceled sibling task while suppressing the expected CancelledError. Check: test_an_idle_hang_up_ends_this_bridges_handler, test_an_idle_hang_up_strands_the_gather_bridge.

3. Count active sessions by user and session

Admission happens before the first session-store await. The active set tracks (user_id, session_id) pairs so identical session IDs from different users do not collide. A second active connection for the same (user_id, session_id) pair is refused with WebSocket code 1008, and a connection beyond max_sessions is refused with code 1013 before opening a Gemini Live session. This local demo trusts user_id from the URL and has no authentication. A deployed service must derive it from authenticated request credentials. With multiple workers, enforce session admission in a shared store or upstream gateway; this set limits only one process.

Two connections use the same active session. Tracking only session_id or failing to reject duplicate (user_id, session_id) pairs lets a second connection collide with an active stream. Reject the duplicate pair with code 1008 without evicting the active handler. Check: test_duplicate_active_session_is_refused_without_replacing_the_first.

Session-store setup fails. Admitting a connection before try without wrapping get_session / create_session in the finally cleanup leaks the capacity slot when the session store raises. Keep session setup inside the try block. Check: test_session_setup_failure_releases_its_capacity_slot.

Process capacity is full. When active sessions reach max_sessions, refuse the new WebSocket with code 1013 before calling run_live(). Check: test_a_caller_over_the_cap_is_refused_with_1013.

4. Run the local recording or offline replay

[spoken question] heard='Which sample rate should the microphone audio use?' said='Microphone audio should use a 16 kHz sample rate.'
  frames={'binary': 68, 'text': 23} audio=174720 bytes, as base64 JSON: 232960 bytes
[second caller, cap 1] closed={'code': 1013, 'reason': 'voice sessions full, try again shortly'} first said='Microphone audio should use 16 kHz for the sample rate.'
[standard model behind the bridge] closed={'code': 1011, 'reason': 'APIError: 1008 None. models/gemini-3.8-flash is not found for API version v1alpha, or is not supported for bidiGenerateC'} events=0
[client hangs up, gather bridge] handlers still open after 10 s: ['hang-up']
[client hangs up, this bridge] handlers still open after 10 s: []

record.py makes live model requests and replaces the saved recording. Run it when you want a new recording:

python record.py

Run the existing offline cases separately:

python -m unittest -v test_server.py

test_a_caller_over_the_cap_is_refused_with_1013 (test_server.Bridge.test_a_caller_over_the_cap_is_refused_with_1013) ... ok
test_a_model_failure_closes_with_1011_and_the_reason (test_server.Bridge.test_a_model_failure_closes_with_1011_and_the_reason) ... ok
test_an_idle_hang_up_ends_this_bridges_handler (test_server.Bridge.test_an_idle_hang_up_ends_this_bridges_handler) ... ok
test_an_idle_hang_up_strands_the_gather_bridge (test_server.Bridge.test_an_idle_hang_up_strands_the_gather_bridge) ... ok
test_audio_leaves_as_binary_frames_and_events_as_json (test_server.Bridge.test_audio_leaves_as_binary_frames_and_events_as_json) ... ok
test_base64_json_would_inflate_the_audio_by_a_third (test_server.Bridge.test_base64_json_would_inflate_the_audio_by_a_third) ... ok
test_duplicate_active_session_is_refused_without_replacing_the_first (test_server.Bridge.test_duplicate_active_session_is_refused_without_replacing_the_first) ... ok
test_session_setup_failure_releases_its_capacity_slot (test_server.Bridge.test_session_setup_failure_releases_its_capacity_slot) ... ok

----------------------------------------------------------------------
Ran 8 tests in 0.176s

OK

record.py starts the FastAPI app locally, drives it with the supplied WAV question and saves the recorded traffic to server-run.json. test_server.py runs Starlette’s TestClient WebSocket against create_app using replayed Live API responses and deterministic fault injection for idle disconnects, duplicate sessions, capacity limits and session-store errors.

How it works

A browser or native voice client connects to your FastAPI server over one WebSocket, while Runner.run_live() opens a second WebSocket from your server to the Gemini Live API. LiveRequestQueue decouples the two sockets: upstream() reads binary PCM chunks (websocket.receive()) and pushes types.Blob items into the queue with send_realtime(), while downstream() iterates over runner.run_live() and writes binary audio (send_bytes) and JSON transcript/control events (send_text) back to the caller.

Because both loops block waiting on I/O, clean shutdown requires coordinating their lifetimes. FIRST_COMPLETED lets the handler enter cleanup when either direction ends. The handler closes the input queue and cancels unfinished tasks. Shielding the finally block protects task cleanup from ASGI cancellation; it does not confirm when the remote service has released its quota.

What’s next

• Read live events and manage live sessions: inspect what run_live() persists across WebSocket reconnections.

Check your understanding

References

Official sources checked October 21, 2026, against the installed Google ADK package.

• ADK: custom live server: the two-direction bridge and input-queue lifecycle. The guide shows gather; this lab adds a FIRST_COMPLETED shutdown policy, exercised by its idle-disconnect tests.

• Python 3.12: tasks, cancellation and waiting primitives: task scheduling and the meaning of FIRST_COMPLETED.

• ADK: live sessions

• FastAPI: WebSockets

• AnyIO: cancellation and shutdown shielding