Generative AI Google Brain

agents-agent-engine-tutorial-bidi-stream-v2

3 min read

Getting Started with Bidirectional Streaming v2 on Agent Runtime

Source notebook

Repo path: agents/agent_engine/tutorial_bidi_stream_v2.ipynb · Open on GitHub · advanced

Builds and deploys Agent Runtime bidirectional streaming agents, including a Live API audio agent.

Summary

This notebook teaches how to build, deploy, and call bidirectional streaming agents on Agent Runtime using bring-your-own-Dockerfile deployments. It first implements a FastAPI WebSocket echo agent, then deploys a Google ADK Live API agent that answers capital-city questions with tool use. The workflow covers project setup, authentication, Vertex AI client configuration, Docker source packaging, Reasoning Engine WebSocket calls, session creation, audio response handling, and cleanup.

Key code patterns

Initialize Agent Platform client

import vertexai
 
LOCATION = "us-central1"
ENDPOINT = f"https://{LOCATION}-aiplatform.googleapis.com"
 
client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
)

Configures the Vertex AI client and regional endpoint used for Agent Runtime deployments and calls.

Deploy from Dockerfile

remote_agent = client.agent_engines.create(
    config={
        "source_packages": ["agent_dir", "Dockerfile", "main.py", "requirements.txt"],
        "image_spec": {},
        "agent_framework": "google-adk",
        "env_vars": {"GOOGLE_GENAI_USE_VERTEXAI": "1"},
        "max_instances": 5,
    },
)

Uses Agent Runtime bring-your-own-Dockerfile deployment and sets max_instances to stay within quota.

Custom bidirectional WebSocket

@app.websocket("/bidi_echo")
async def bidi_handler(websocket: WebSocket):
    await websocket.accept()
    initial_data = await websocket.receive_json(mode="binary")
    queue = asyncio.Queue()
    if "input" in initial_data:
        queue.put_nowait(initial_data["input"])
    await asyncio.gather(receive_messages(), send_messages())

Shows how a deployed agent can expose an arbitrary WebSocket protocol for bidirectional streaming.

ADK Live API agent with tool

root_agent = Agent(
    model="gemini-live-2.5-flash-native-audio",
    name="capital_agent",
    instruction="Use get_capital_city to answer capital questions.",
    tools=[get_capital_city],
)

Defines a Live API ADK agent that uses a Python function tool during real-time audio interaction.

Receive audio over WebSocket

if "inlineData" in part:
    mime_type = part["inlineData"].get("mimeType")
    if mime_type == "audio/pcm":
        b64_data = part["inlineData"].get("data")
        decoded_chunk = base64.urlsafe_b64decode(b64_data)
        audio_buffer += decoded_chunk

Parses model response parts, decodes audio/pcm inline data, and accumulates it for WAV playback.

Models & APIs used

  • Models: gemini-live-2.5-flash-native-audio
  • APIs / services: Vertex AI, Agent Runtime, Agent Platform, Live API
  • SDKs / libraries: google-cloud-aiplatform, vertexai, google-adk, FastAPI, uvicorn, websockets, google-auth, requests

When to use this

Use this pattern when deploying custom Agent Runtime services that need real-time bidirectional WebSocket interaction or Live API audio conversations.

Gotchas & caveats

  • Requires a Google Cloud project ID and authenticated Google user credentials.
  • Notebook uses location us-central1 and builds endpoints from that region.
  • Requires google-cloud-aiplatform[agent_engines,adk]>=1.144.
  • The Dockerfile uses python:3.13-alpine and expects Cloud Run to provide the PORT environment variable.
  • The echo WebSocket path must match AGENT_PATH, such as bidi_echo.
  • The live agent requires creating a session before calling run_live.
  • The notebook explicitly sets max_instances to 5 to stay within quota.
  • Live audio handling expects inlineData with mimeType audio/pcm and base64 data.

Best practices

  • Use bring-your-own-Dockerfile deployment with source_packages for custom Agent Runtime servers.
  • Set GOOGLE_GENAI_USE_VERTEXAI to 1 for the deployed ADK agent environment.
  • Use resource_limits and max_instances in Agent Runtime deployment config.
  • Refresh google.auth credentials before making HTTP or WebSocket calls.
  • Separate receive and send loops for bidirectional WebSocket handling.
  • Delete deployed agents with force=True during cleanup.

Graph View

Backlinks