Robotics with streaming

The gemini-robotics-er-2-streaming-preview model endpoint exposes a dedicated streaming endpoint that integrates with the Live API, enabling real-time, bidirectional interaction between your application and the robot. This makes it suited for agents that need fast feedback loops and reactive responses to the environment.

Use cases

• Multi-robot coordination: Multiple robots that communicate task state and delegate subtasks through a shared session.
• Continuous monitoring: Robots that observe a scene and trigger actions when specific events occur, such as a container reaching a fill level.
• Warehouse and logistics: Pick-and-pack agents that verify items visually, track packing progress, and recover from errors.

Technical specifications

The following table outlines the technical specifications for the Live API:

Build an agentic setup

Every robotics agent built on the Live API follows three steps:

1. Declare robot capabilities as tools. Each action the robot can perform — navigate, grasp, speak — becomes a function declaration with a name, description, and parameter schema. Physical actions must use "behavior": "BLOCKING" so the model waits for the robot to finish before choosing the next step.
2. Stream multimodal input into a persistent session. Open a live.connect session and keep it open for the life of the task. Send video frames, audio, or text as they arrive from your robot's sensors.
3. Handle tool calls in a receive loop. Each time the model selects an action, it sends a tool_call message. Your receive loop executes the function against your robot SDK and sends back a tool_response. The session stays open, and the model picks the next action based on the result.

The following sections show how to apply these steps to three common patterns: a baseline agent loop, proactive scene monitoring with a heartbeat, and routing speech through TTS as a tool.

Orchestrate a robot through function calling

The following example shows all three steps wired together in a single Python script.

Step 1 — tool definitions — declares robot capabilities as function declarations. The navigate function uses "behavior": "BLOCKING" so the model waits for the robot to reach the waypoint before calling another tool. Add more function declarations in the same list to expose additional robot capabilities.

Step 2 — input helpers — shows three functions that stream different modality inputs into the session: send_text for commands, send_image for camera frames with an optional text prompt, and send_audio for raw PCM audio from a microphone.

Step 3 — the receive loop — runs concurrently and handles two kinds of messages: server_content messages (the model's text output) and tool_call messages (the model requesting a robot action). When a tool call arrives, the loop calls execute_tool — a stub you replace with your real robot SDK — then sends back a tool_response so the model can select the next action.

import asyncio
from google import genai
from google.genai import types

MODEL = "gemini-robotics-er-2-streaming-preview"

# ── Tool definitions ─────────────────────────────────────────────────────────
tools = [
   {
       "function_declarations": [
           {
               "name": "navigate",
               "description": "Navigate the robot to a named waypoint.",
               "behavior": "BLOCKING",
               "parameters": {
                   "type": "OBJECT",
                   "properties": {"name": {"type": "STRING"}},
                   "required": ["name"],
               },
           },
           # Add more function definitions here
       ]
   }
]

# ── Stub tool executor (replace with real robot SDK calls) ───────────────────
def execute_tool(name: str, args: dict) -> dict:
   print(f"  [Tool] {name}({args})")
   return {"status": "success"}

# ── Input helpers ────────────────────────────────────────────────────────────
def send_text(session, text: str):
   """Send a text turn."""
   return session.send_client_content(
       turns=types.Content(role="user", parts=[types.Part(text=text)]),
       turn_complete=True,
   )

def send_image(session, image_bytes: bytes, prompt: str = ""):
   """Send a JPEG image with an optional text prompt."""
   parts = [
       types.Part(
           inline_data=types.Blob(data=image_bytes, mime_type="image/jpeg")
       )
   ]
   if prompt:
       parts.append(types.Part(text=prompt))
   return session.send_client_content(
       turns=types.Content(role="user", parts=parts),
       turn_complete=True,
   )

def send_audio(session, audio_chunk: bytes):
   """Stream a chunk of raw PCM audio (16-bit, 16 kHz, mono)."""
   return session.send_realtime_input(
       media=types.Blob(data=audio_chunk, mime_type="audio/pcm;rate=16000")
   )

# ── Receive loop ─────────────────────────────────────────────────────────────
async def receive_loop(session):
   """Print model text and handle tool calls until the session ends."""
   async for message in session.receive():
       if message.server_content:
           sc = message.server_content
           if sc.model_turn and sc.model_turn.parts:
               for part in sc.model_turn.parts:
                   if part.text:
                       print(f"Model: