Realtime: make example customizable: (#1332)

Trying to make it easy to customize/use your own agents and reuse the
UI.

Author: rm-openai

examples/realtime/app/README.md

@@ -20,6 +20,10 @@ cd examples/realtime/app && uv run python server.py
 
 Then open your browser to: http://localhost:8000
 
+## Customization
+
+To use the same UI with your own agents, edit `agent.py` and ensure get_starting_agent() returns the right starting agent for your use case.
+
 ## How to Use
 
 1. Click **Connect** to establish a realtime session

examples/realtime/app/agent.py

@@ -0,0 +1,36 @@
+from agents import function_tool
+from agents.realtime import RealtimeAgent
+
+"""
+When running the UI example locally, you can edit this file to change the setup. THe server
+will use the agent returned from get_starting_agent() as the starting agent."""
+
+
+@function_tool
+def get_weather(city: str) -> str:
+    """Get the weather in a city."""
+    return f"The weather in {city} is sunny."
+
+
+@function_tool
+def get_secret_number() -> int:
+    """Returns the secret number, if the user asks for it."""
+    return 71
+
+
+haiku_agent = RealtimeAgent(
+    name="Haiku Agent",
+    instructions="You are a haiku poet. You must respond ONLY in traditional haiku format (5-7-5 syllables). Every response should be a proper haiku about the topic. Do not break character.",
+    tools=[],
+)
+
+assistant_agent = RealtimeAgent(
+    name="Assistant",
+    instructions="If the user wants poetry or haikus, you can hand them off to the haiku agent via the transfer_to_haiku_agent tool.",
+    tools=[get_weather, get_secret_number],
+    handoffs=[haiku_agent],
+)
+
+
+def get_starting_agent() -> RealtimeAgent:
+    return assistant_agent

examples/realtime/app/server.py

@@ -4,43 +4,30 @@
 import logging
 import struct
 from contextlib import asynccontextmanager
-from typing import Any, assert_never
+from typing import TYPE_CHECKING, Any, assert_never
 
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
-from agents import function_tool
-from agents.realtime import RealtimeAgent, RealtimeRunner, RealtimeSession, RealtimeSessionEvent
-
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-
-@function_tool
-def get_weather(city: str) -> str:
-    """Get the weather in a city."""
-    return f"The weather in {city} is sunny."
-
-
-@function_tool
-def get_secret_number() -> int:
-    """Returns the secret number, if the user asks for it."""
-    return 71
+from agents.realtime import RealtimeRunner, RealtimeSession, RealtimeSessionEvent
 
+# Import TwilioHandler class - handle both module and package use cases
+if TYPE_CHECKING:
+    # For type checking, use the relative import
+    from .agent import get_starting_agent
+else:
+    # At runtime, try both import styles
+    try:
+        # Try relative import first (when used as a package)
+        from .agent import get_starting_agent
+    except ImportError:
+        # Fall back to direct import (when run as a script)
+        from agent import get_starting_agent
 
-haiku_agent = RealtimeAgent(
-    name="Haiku Agent",
-    instructions="You are a haiku poet. You must respond ONLY in traditional haiku format (5-7-5 syllables). Every response should be a proper haiku about the topic. Do not break character.",
-    tools=[],
-)
 
-agent = RealtimeAgent(
-    name="Assistant",
-    instructions="If the user wants poetry or haikus, you can hand them off to the haiku agent via the transfer_to_haiku_agent tool.",
-    tools=[get_weather, get_secret_number],
-    handoffs=[haiku_agent],
-)
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 
 class RealtimeWebSocketManager:
@@ -53,6 +40,7 @@ async def connect(self, websocket: WebSocket, session_id: str):
         await websocket.accept()
         self.websockets[session_id] = websocket
 
+        agent = get_starting_agent()
         runner = RealtimeRunner(agent)
         session_context = await runner.run()
         session = await session_context.__aenter__()
@@ -150,6 +138,7 @@ async def websocket_endpoint(websocket: WebSocket, session_id: str):
 
             if message["type"] == "audio":
                 # Convert int16 array to bytes
+                print("Received audio data")
                 int16_data = message["data"]
                 audio_bytes = struct.pack(f"{len(int16_data)}h", *int16_data)
                 await manager.send_audio(session_id, audio_bytes)