feat: #1614 gpt-realtime migration (Realtime API GA)

.github/workflows/tests.yml

@@ -84,7 +84,5 @@ jobs:
  enable-cache: true
  - name: Install dependencies
  run: make sync
- - name: Install Python 3.9 dependencies
- run: UV_PROJECT_ENVIRONMENT=.venv_39 uv sync --all-extras --all-packages --group dev
  - name: Run tests
  run: make old_version_tests

.gitignore

@@ -100,7 +100,8 @@ celerybeat.pid
 *.sage.py
 
 # Environments
-.env
+.python-version
+.env*
 .venv
 env/
 venv/

Makefile

@@ -39,7 +39,8 @@ snapshots-create:
 uv run pytest --inline-snapshot=create 
 
 .PHONY: old_version_tests
-old_version_tests: 
+old_version_tests:
+UV_PROJECT_ENVIRONMENT=.venv_39 uv sync --python 3.9 --all-extras --all-packages --group dev
 UV_PROJECT_ENVIRONMENT=.venv_39 uv run --python 3.9 -m pytest
 
 .PHONY: build-docs

examples/realtime/app/README.md

@@ -29,14 +29,19 @@ To use the same UI with your own agents, edit `agent.py` and ensure get_starting
 1. Click **Connect** to establish a realtime session
 2. Audio capture starts automatically - just speak naturally
 3. Click the **Mic On/Off** button to mute/unmute your microphone
-4. Watch the conversation unfold in the left pane
-5. Monitor raw events in the right pane (click to expand/collapse)
-6. Click **Disconnect** when done
+4. To send an image, enter an optional prompt and click **🖼️ Send Image** (select a file)
+5. Watch the conversation unfold in the left pane (image thumbnails are shown)
+6. Monitor raw events in the right pane (click to expand/collapse)
+7. Click **Disconnect** when done
 
 ## Architecture
 
 - **Backend**: FastAPI server with WebSocket connections for real-time communication
 - **Session Management**: Each connection gets a unique session with the OpenAI Realtime API
+- **Image Inputs**: The UI uploads images and the server forwards a
+ `conversation.item.create` event with `input_image` (plus optional `input_text`),
+ followed by `response.create` to start the model response. The messages pane
+ renders image bubbles for `input_image` content.
 - **Audio Processing**: 24kHz mono audio capture and playback
 - **Event Handling**: Full event stream processing with transcript generation
 - **Frontend**: Vanilla JavaScript with clean, responsive CSS

examples/realtime/app/server.py

@@ -12,6 +12,8 @@
 from typing_extensions import assert_never
 
 from agents.realtime import RealtimeRunner, RealtimeSession, RealtimeSessionEvent
+from agents.realtime.config import RealtimeUserInputMessage
+from agents.realtime.model_inputs import RealtimeModelSendRawMessage
 
 # Import TwilioHandler class - handle both module and package use cases
 if TYPE_CHECKING:
@@ -64,6 +66,34 @@ async def send_audio(self, session_id: str, audio_bytes: bytes):
  if session_id in self.active_sessions:
  await self.active_sessions[session_id].send_audio(audio_bytes)
 
+ async def send_client_event(self, session_id: str, event: dict[str, Any]):
+ """Send a raw client event to the underlying realtime model."""
+ session = self.active_sessions.get(session_id)
+ if not session:
+ return
+ await session.model.send_event(
+ RealtimeModelSendRawMessage(
+ message={
+ "type": event["type"],
+ "other_data": {k: v for k, v in event.items() if k != "type"},
+ }
+ )
+ )
+
+ async def send_user_message(self, session_id: str, message: RealtimeUserInputMessage):
+ """Send a structured user message via the higher-level API (supports input_image)."""
+ session = self.active_sessions.get(session_id)
+ if not session:
+ return
+ await session.send_message(message) # delegates to RealtimeModelSendUserInput path
+
+ async def interrupt(self, session_id: str) -> None:
+ """Interrupt current model playback/response for a session."""
+ session = self.active_sessions.get(session_id)
+ if not session:
+ return
+ await session.interrupt()
+
  async def _process_events(self, session_id: str):
  try:
  session = self.active_sessions[session_id]
@@ -101,7 +131,11 @@ async def _serialize_event(self, event: RealtimeSessionEvent) -> dict[str, Any]:
  elif event.type == "history_updated":
  base_event["history"] = [item.model_dump(mode="json") for item in event.history]
  elif event.type == "history_added":
- pass
+ # Provide the added item so the UI can render incrementally.
+ try:
+ base_event["item"] = event.item.model_dump(mode="json")
+ except Exception:
+ base_event["item"] = None
  elif event.type == "guardrail_tripped":
  base_event["guardrail_results"] = [
  {"name": result.guardrail.name} for result in event.guardrail_results
@@ -134,6 +168,7 @@ async def lifespan(app: FastAPI):
 @app.websocket("/ws/{session_id}")
 async def websocket_endpoint(websocket: WebSocket, session_id: str):
  await manager.connect(websocket, session_id)
+ image_buffers: dict[str, dict[str, Any]] = {}
  try:
  while True:
  data = await websocket.receive_text()
@@ -144,6 +179,124 @@ async def websocket_endpoint(websocket: WebSocket, session_id: str):
  int16_data = message["data"]
  audio_bytes = struct.pack(f"{len(int16_data)}h", *int16_data)
  await manager.send_audio(session_id, audio_bytes)
+ elif message["type"] == "image":
+ logger.info("Received image message from client (session %s).", session_id)
+ # Build a conversation.item.create with input_image (and optional input_text)
+ data_url = message.get("data_url")
+ prompt_text = message.get("text") or "Please describe this image."
+ if data_url:
+ logger.info(
+ "Forwarding image (structured message) to Realtime API (len=%d).",
+ len(data_url),
+ )
+ user_msg: RealtimeUserInputMessage = {
+ "type": "message",
+ "role": "user",
+ "content": (
+ [
+ {"type": "input_image", "image_url": data_url, "detail": "high"},
+ {"type": "input_text", "text": prompt_text},
+ ]
+ if prompt_text
+ else [
+ {"type": "input_image", "image_url": data_url, "detail": "high"}
+ ]
+ ),
+ }
+ await manager.send_user_message(session_id, user_msg)
+ # Acknowledge to client UI
+ await websocket.send_text(
+ json.dumps(
+ {
+ "type": "client_info",
+ "info": "image_enqueued",
+ "size": len(data_url),
+ }
+ )
+ )
+ else:
+ await websocket.send_text(
+ json.dumps(
+ {
+ "type": "error",
+ "error": "No data_url for image message.",
+ }
+ )
+ )
+ elif message["type"] == "commit_audio":
+ # Force close the current input audio turn
+ await manager.send_client_event(session_id, {"type": "input_audio_buffer.commit"})
+ elif message["type"] == "image_start":
+ img_id = str(message.get("id"))
+ image_buffers[img_id] = {
+ "text": message.get("text") or "Please describe this image.",
+ "chunks": [],
+ }
+ await websocket.send_text(
+ json.dumps({"type": "client_info", "info": "image_start_ack", "id": img_id})
+ )
+ elif message["type"] == "image_chunk":
+ img_id = str(message.get("id"))
+ chunk = message.get("chunk", "")
+ if img_id in image_buffers:
+ image_buffers[img_id]["chunks"].append(chunk)
+ if len(image_buffers[img_id]["chunks"]) % 10 == 0:
+ await websocket.send_text(
+ json.dumps(
+ {
+ "type": "client_info",
+ "info": "image_chunk_ack",
+ "id": img_id,
+ "count": len(image_buffers[img_id]["chunks"]),
+ }
+ )
+ )
+ elif message["type"] == "image_end":
+ img_id = str(message.get("id"))
+ buf = image_buffers.pop(img_id, None)
+ if buf is None:
+ await websocket.send_text(
+ json.dumps({"type": "error", "error": "Unknown image id for image_end."})
+ )
+ else:
+ data_url = "".join(buf["chunks"]) if buf["chunks"] else None
+ prompt_text = buf["text"]
+ if data_url:
+ logger.info(
+ "Forwarding chunked image (structured message) to Realtime API (len=%d).",
+ len(data_url),
+ )
+ user_msg2: RealtimeUserInputMessage = {
+ "type": "message",
+ "role": "user",
+ "content": (
+ [
+ {"type": "input_image", "image_url": data_url, "detail": "high"},
+ {"type": "input_text", "text": prompt_text},
+ ]
+ if prompt_text
+ else [
+ {"type": "input_image", "image_url": data_url, "detail": "high"}
+ ]
+ ),
+ }
+ await manager.send_user_message(session_id, user_msg2)
+ await websocket.send_text(
+ json.dumps(
+ {
+ "type": "client_info",
+ "info": "image_enqueued",
+ "id": img_id,
+ "size": len(data_url),
+ }
+ )
+ )
+ else:
+ await websocket.send_text(
+ json.dumps({"type": "error", "error": "Empty image."})
+ )
+ elif message["type"] == "interrupt":
+ await manager.interrupt(session_id)
 
  except WebSocketDisconnect:
  await manager.disconnect(session_id)
@@ -160,4 +313,10 @@ async def read_index():
 if __name__ == "__main__":
  import uvicorn
 
- uvicorn.run(app, host="0.0.0.0", port=8000)
+ uvicorn.run(
+ app,
+ host="0.0.0.0",
+ port=8000,
+ # Increased WebSocket frame size to comfortably handle image data URLs.
+ ws_max_size=16 * 1024 * 1024,
+ )