Repository Public
1<br />23> [!NOTE]4> **Note:** Robotics with streaming requires `gemini-robotics-er-2-streaming-preview`. It is not available on ER 1.6 or the standard `gemini-robotics-er-2-preview` model endpoint.56The `gemini-robotics-er-2-streaming-preview` model endpoint exposes a dedicated7streaming endpoint that integrates with the [Live8API](https://ai.google.dev/gemini-api/docs/live-api/get-started-sdk), enabling real-time,9bidirectional interaction between your application and the robot. This makes it10suited for agents that need fast feedback loops and reactive responses to the11environment.12[Try in Google AI Studio](https://aistudio.google.com/prompts/new_chat?model=gemini-robotics-er-2-streaming-preview) [Clone example apps from GitHub](https://github.com/google-gemini/robotics-samples/tree/main/live-api)1314## Use cases1516- **Multi-robot coordination**: Multiple robots that communicate task state and delegate subtasks through a shared session.17- **Continuous monitoring**: Robots that observe a scene and trigger actions when specific events occur, such as a container reaching a fill level.18- **Warehouse and logistics**: Pick-and-pack agents that verify items visually, track packing progress, and recover from errors.1920## Technical specifications2122The following table outlines the technical specifications for the23Live API:2425| Category | Details |26|---|---|27| Input modalities | Audio (raw 16-bit PCM audio, 16kHz, little-endian), images (JPEG \<= 1FPS), text |28| Output modalities | Text |29| Protocol | Stateful WebSocket connection (WSS) |3031## Build an agentic setup3233Every robotics agent built on the Live API follows three steps:34351. **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.362. **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.373. **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.3839> [!NOTE]40> **Note:** Only blocking function calls are supported when using the Live API for robotics. See the [Live API tools guide](https://ai.google.dev/gemini-api/docs/live-api/tools) for details.4142The following sections show how to apply these steps to three common patterns:43a baseline agent loop, proactive scene monitoring with a heartbeat, and routing44speech through TTS as a tool.4546## Orchestrate a robot through function calling4748The following example shows all three steps wired together in a single Python49script.5051Step 1 --- tool definitions --- declares robot capabilities as function52declarations. The `navigate` function uses `"behavior": "BLOCKING"` so the53model waits for the robot to reach the waypoint before calling another tool.54Add more function declarations in the same list to expose additional robot55capabilities.5657Step 2 --- input helpers --- shows three functions that stream different modality58inputs into the session: `send_text` for commands, `send_image` for camera59frames with an optional text prompt, and `send_audio` for raw PCM audio from a60microphone.6162Step 3 --- the receive loop --- runs concurrently and handles two kinds of messages:63`server_content` messages (the model's text output) and `tool_call` messages64(the model requesting a robot action). When a tool call arrives, the loop calls65`execute_tool` --- a stub you replace with your real robot SDK --- then sends back a66`tool_response` so the model can select the next action.6768 import asyncio69 from google import genai70 from google.genai import types7172 MODEL = "gemini-robotics-er-2-streaming-preview"7374 # ── Tool definitions ─────────────────────────────────────────────────────────75 tools = [76 {77 "function_declarations": [78 {79 "name": "navigate",80 "description": "Navigate the robot to a named waypoint.",81 "behavior": "BLOCKING",82 "parameters": {83 "type": "OBJECT",84 "properties": {"name": {"type": "STRING"}},85 "required": ["name"],86 },87 },88 # Add more function definitions here89 ]90 }91 ]9293 # ── Stub tool executor (replace with real robot SDK calls) ───────────────────94 def execute_tool(name: str, args: dict) -> dict:95 print(f" [Tool] {name}({args})")96 return {"status": "success"}9798 # ── Input helpers ────────────────────────────────────────────────────────────99 def send_text(session, text: str):100 """Send a text turn."""101 return session.send_client_content(102 turns=types.Content(role="user", parts=[types.Part(text=text)]),103 turn_complete=True,104 )105106 def send_image(session, image_bytes: bytes, prompt: str = ""):107 """Send a JPEG image with an optional text prompt."""108 parts = [109 types.Part(110 inline_data=types.Blob(data=image_bytes, mime_type="image/jpeg")111 )112 ]113 if prompt:114 parts.append(types.Part(text=prompt))115 return session.send_client_content(116 turns=types.Content(role="user", parts=parts),117 turn_complete=True,118 )119120 def send_audio(session, audio_chunk: bytes):121 """Stream a chunk of raw PCM audio (16-bit, 16 kHz, mono)."""122 return session.send_realtime_input(123 media=types.Blob(data=audio_chunk, mime_type="audio/pcm;rate=16000")124 )125126 # ── Receive loop ─────────────────────────────────────────────────────────────127 async def receive_loop(session):128 """Print model text and handle tool calls until the session ends."""129 async for message in session.receive():130 if message.server_content:131 sc = message.server_content132 if sc.model_turn and sc.model_turn.parts:133 for part in sc.model_turn.parts:134 if part.text:135 print(f"Model: {part.text}", end="", flush=True)136 if sc.turn_complete:137 print("\n[Turn Complete]")138 elif message.tool_call:139 responses = []140 for call in message.tool_call.function_calls:141 print(f"\n[Tool Call] {call.name}({call.args})")142 result = execute_tool(call.name, call.args)143 responses.append(144 types.FunctionResponse(145 name=call.name,146 response=result,147 id=call.id,148 )149 )150 await session.send_tool_response(function_responses=responses)151152 # ── Main ─────────────────────────────────────────────────────────────────────153 async def main():154 client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])155 config = types.LiveConnectConfig(156 response_modalities=["TEXT"],157 tools=tools,158 system_instruction=types.Content(159 parts=[types.Part(text="You are a robot controller. Use tools to execute commands.")]160 ),161 )162 async with client.aio.live.connect(model=MODEL, config=config) as session:163 recv_task = asyncio.create_task(receive_loop(session))164 # Connect robot perception callbacks and user inputs to the helpers above.165 recv_task.cancel()166167 asyncio.run(main())168169The receive loop stays active after each tool response. The model constructs170and revises a long-horizon plan without you encoding the entire action sequence171in advance.172173## Proactive spatial-temporal reasoning174175The Live API streams video in, but video frames alone do176not trigger a new reasoning turn. Video frames must be accompanied by a177text or audio prompt to trigger a model response. See178[Live API capabilities](https://ai.google.dev/gemini-api/docs/live-api/capabilities) for179more details.180181To enable proactive reasoning, implement a **heartbeat**: periodically send the182latest camera frame followed by a short text prompt that forces the model to183inspect the scene and make an explicit decision. Video input is rate-limited to184one frame per second.185186### Implement the heartbeat187188The heartbeat coroutine runs as a separate `asyncio` task in the same session.189It opportunistically targets a 1 Hz cadence (matching the video input rate limit)190while waiting for each turn to complete (`er_turn_done`) to avoid interrupting191in-flight reasoning:192193 async def heartbeat(session, camera, er_turn_done: asyncio.Event):194 TARGET_INTERVAL_SEC = 1.0195196 while True:197 start_time = asyncio.get_running_loop().time()198199 frame = await camera.latest_jpeg()200 await session.send_realtime_input(201 video=types.Blob(data=frame, mime_type="image/jpeg")202 )203 await session.send_realtime_input(204 text=(205 "[HEARTBEAT] If no task is active, call 'ack' and wait for user"206 " input. If a task is active: observe the scene. If the current"207 " step is progressing correctly, call 'ack'. If the current step"208 " is complete, call 'run_instruction' with the next step. If the"209 " overall goal is achieved, call 'reset' and inform the user."210 )211 )212213 # Wait for the model to finish responding before sending the next heartbeat214 await er_turn_done.wait()215 er_turn_done.clear()216217 # Sleep only the remaining time to maintain ~1 Hz cadence218 elapsed = asyncio.get_running_loop().time() - start_time219 remaining = TARGET_INTERVAL_SEC - elapsed220 if remaining > 0:221 await asyncio.sleep(remaining)222223> [!CAUTION]224> **Caution:** Sending a new prompt while the model is generating acts as an interruption (barge-in), which cancels in-flight reasoning and tool calls. If you fire heartbeats on a blind timer without waiting for the previous turn to complete, you may trap the model in a cancellation loop, making it appear unresponsive. See the [Live API guide on interruptions](https://ai.google.dev/gemini-api/docs/live-api/capabilities#interruptions) for more details.225226### Update the receive loop227228To signal when the model has completed its turn, update your `receive_loop`229to set `er_turn_done`:230231 # In receive_loop: signal when the model finishes its turn232 if sc.turn_complete:233 er_turn_done.set()234235## Audio output through external TTS236237Gemini Robotics ER 2 returns text. Your application routes completed responses238to a separate TTS provider (such as239[Gemini TTS](https://ai.google.dev/gemini-api/docs/speech-generation)) via an injected callback.240This keeps speech latency, voice selection, and interruption behavior under your241control, and lets you swap TTS backends without changing agent logic.242243You can also declare TTS as a tool so the model treats "say something" the same244as "move the arm." Add the following function declaration to your `tools` list245from the first section:246247 TOOLS = [248 {249 "name": "send_message",250 "description": (251 "Speak a message aloud via TTS, then deliver it to the"252 " specified target. Use target='user' to speak directly"253 " to the user, or a peer agent name (e.g., 'duo') to"254 " communicate with another robot."255 ),256 "parameters": {257 "type": "object",258 "properties": {259 "target": {260 "type": "string",261 "description": "Recipient: 'user' or a peer agent name.",262 },263 "message": {264 "type": "string",265 "description": "The message to speak and deliver.",266 },267 },268 "required": ["target", "message"],269 },270 },271 ]272273By wrapping TTS in a function declaration, the model handles speech through the274same tool-call path as any other robot action. Your application fulfills the275call with an injected callback.276277## Examples on GitHub278279For full working examples including the Spot robot snack-fetch demo and Tinybot280pan-tilt hello world, see281[Robotics Live API examples](https://github.com/google-gemini/robotics-samples/tree/main/live-api).282283## What's next284285- [Video understanding](https://ai.google.dev/gemini-api/docs/robotics-video-progress) --- moment finding and progress classification.286- [Task orchestration](https://ai.google.dev/gemini-api/docs/robotics-orchestration) --- long-horizon tasks without streaming.287- [Live API overview](https://ai.google.dev/gemini-api/docs/live-api/get-started-sdk) --- full Live API documentation.