Generative AI Google Brain

agents-agent-engine-tutorial-get-started-with-live-api-on-agent-engine

4 min read

Getting Started with Live API on Agent Engine

Source notebook

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

Deploys bidirectional streaming agents on Vertex AI Agent Engine using Gemini Live API audio and ADK tools.

Summary

The notebook teaches how to build, deploy, test, and clean up Agent Engine agents that support standard query, server-side streaming, and bidirectional streaming. It then demonstrates a WebSocket-based Gemini Live API audio agent and an ADK weather assistant with mock weather tools, in-memory session and memory services, and audio/text response handling.

Key code patterns

Agent Engine operation routing

def register_operations(self):
    return {
        "": ["query"],
        "stream": ["stream_query"],
        "bidi_stream": ["bidi_stream_query"]
    }

Registers which local class methods Agent Engine exposes for query, streaming, and bidirectional endpoints.

Deploy an agent

remote_echo_agent = client.agent_engines.create(
    agent=echo_agent,
    config={
        "display_name": "Echo Agent Tutorial",
        "requirements": ["google-cloud-aiplatform[agent_engines] @ git+https://github.com/googleapis/python-aiplatform.git"],
        "staging_bucket": BUCKET_URI,
        "agent_server_mode": vertexai_types.AgentServerMode.EXPERIMENTAL
    }
)

Shows the core deployment pattern: serialize a Python agent, stage dependencies in Cloud Storage, and create a managed endpoint.

Bidirectional Agent Engine session

async with client.aio.live.agent_engines.connect(
    agent_engine=remote_echo_agent.api_resource.name,
    config={"class_method": "bidi_stream_query"}
) as session:
    await session.send({"input": "Hello from bidi"})
    response = await session.receive()

Uses a persistent async session for two-way communication with the deployed bidi_stream_query method.

Live API WebSocket setup

self.service_url = (
    f"wss://{self.location}-aiplatform.googleapis.com/ws/"
    "google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
)
self.model = f"projects/{self.project}/locations/{self.location}/publishers/google/models/{self.model_id}"
self.config = {"response_modalities": ["AUDIO"]}

Builds the Vertex AI Live API endpoint and configures Gemini to stream audio responses.

ADK tool state

async def get_weather(location: str, tool_context: ToolContext) -> Dict[str, str]:
    if "weather_queries" not in tool_context.state:
        tool_context.state["weather_queries"] = []
    tool_context.state["weather_queries"].append({"location": location})
    return {"location": location, "conditions": conditions}

Demonstrates ADK tool calling with ToolContext state for per-session weather query history.

Wrap ADK agent for Agent Engine

app = AdkApp(
    agent=weather_agent,
    session_service_builder=session_service_builder,
    memory_service_builder=memory_service_builder
)
remote_live_adk_agent = client.agent_engines.create(agent=app, config={...})

Shows that an ADK LlmAgent is deployed by wrapping it in AdkApp with session and memory service builders.

Models & APIs used

  • Models: gemini-2.0-flash-live-preview-04-09
  • APIs / services: Vertex AI, Vertex AI Agent Engine, Live API, Cloud Storage
  • SDKs / libraries: google-cloud-aiplatform, vertexai, google-adk, google-genai, google-auth, websockets, numpy

When to use this

Use this pattern when building managed, real-time conversational agents on Vertex AI that need bidirectional text or audio streaming and optional ADK tool orchestration.

Gotchas & caveats

  • Vertex AI API must be enabled for the Google Cloud project.
  • PROJECT_ID and BUCKET_NAME must be provided or available from environment variables.
  • LOCATION defaults to global and is used to build the Live API host and staging bucket location.
  • A Cloud Storage staging bucket is created with gsutil mb and may fail if it already exists or the user lacks permissions.
  • Agent Engine deployments require all runtime dependencies to be listed in requirements.
  • The notebook uses AgentServerMode.EXPERIMENTAL.
  • The Live API agent requires authentication with google.auth.default and refreshed credentials for WebSocket access.
  • The weather tools return mock data and the notebook notes production use should integrate a real weather API.
  • Cleanup is needed to avoid charges from deployed agents.

Best practices

  • Keep init lightweight and pickle-able for Agent Engine serialization.
  • Put heavy initialization in set_up because Agent Engine calls it when the serverless container starts.
  • Make each stream_query yield a complete serializable response object.
  • Use bidi_stream_query with asyncio.Queue for continuous two-way sessions.
  • Declare agent dependencies in the Agent Engine requirements config.
  • Use Cloud Storage as the staging bucket for Agent Engine deployment.
  • Use ToolContext state to persist tool-related session data in ADK.
  • Wrap ADK agents in AdkApp before deploying them to Agent Engine.
  • Delete deployed agents after the tutorial to avoid charges.

Graph View

Backlinks