Build with the OpenAI Realtime API

export OPENAI_API_KEY="sk-..."

# Python
pip install websockets numpy

# Node.js
npm install ws

import asyncio
import json
import websockets
import os

API_KEY = os.environ["OPENAI_API_KEY"]
URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"

async def connect():
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }
    async with websockets.connect(URL, additional_headers=headers) as ws:
        # The first event is always session.created
        event = json.loads(await ws.recv())
        print(f"✅ Connected! Session: {event['session']['id']}")

        # Configure the session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "instructions": "You are a friendly assistant.",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 600
                }
            }
        }))

        updated = json.loads(await ws.recv())
        print(f"✅ Session configured: {updated['type']}")

asyncio.run(connect())

import WebSocket from "ws";

const ws = new WebSocket(
  "wss://api.openai.com/v1/realtime?model=gpt-realtime",
  {
    headers: {
      Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
      "OpenAI-Beta": "realtime=v1",
    },
  }
);

ws.on("open", () => {
  ws.send(JSON.stringify({
    type: "session.update",
    session: {
      instructions: "You are a friendly assistant.",
      voice: "alloy",
      input_audio_format: "pcm16",
      output_audio_format: "pcm16",
      turn_detection: { type: "server_vad" },
    },
  }));
});

ws.on("message", (raw) => {
  const event = JSON.parse(raw.toString());
  console.log(`📨 ${event.type}`);
});

ws.on("error", (err) => console.error("Error:", err));

import asyncio, json, base64, os
import websockets
import numpy as np

# pip install sounddevice
import sounddevice as sd

API_KEY = os.environ["OPENAI_API_KEY"]
URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
SAMPLE_RATE = 24000
CHUNK_DURATION_MS = 100
CHUNK_SAMPLES = int(SAMPLE_RATE * CHUNK_DURATION_MS / 1000)

async def voice_chat():
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(URL, additional_headers=headers) as ws:
        await ws.recv()  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "instructions": "You are a concise, friendly voice assistant.",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 700
                }
            }
        }))
        await ws.recv()  # session.updated
        print("🎤 Listening... (Ctrl+C to stop)\n")

        # --- Mic capture task ---
        audio_queue = asyncio.Queue()

        def mic_callback(indata, frames, time_info, status):
            audio_queue.put_nowait(indata.copy())

        stream = sd.InputStream(
            samplerate=SAMPLE_RATE,
            channels=1,
            dtype="int16",
            blocksize=CHUNK_SAMPLES,
            callback=mic_callback,
        )
        stream.start()

        async def send_audio():
            while True:
                chunk = await audio_queue.get()
                encoded = base64.b64encode(chunk.tobytes()).decode()
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": encoded
                }))

        # --- Receive task ---
        async def receive_events():
            playback_buffer = bytearray()
            async for msg in ws:
                event = json.loads(msg)
                t = event["type"]

                if t == "response.audio.delta":
                    audio = base64.b64decode(event["delta"])
                    playback_buffer.extend(audio)

                elif t == "response.audio_transcript.delta":
                    print(event["delta"], end="", flush=True)

                elif t == "response.done":
                    print()  # newline
                    if playback_buffer:
                        samples = np.frombuffer(
                            playback_buffer, dtype=np.int16
                        )
                        sd.play(samples, SAMPLE_RATE)
                        sd.wait()
                        playback_buffer.clear()

                elif t == "error":
                    print(f"\n❌ Error: {event['error']}")

        await asyncio.gather(send_audio(), receive_events())

asyncio.run(voice_chat())

// browser-voice-chat.js
// Requires a server endpoint POST /api/token that returns
// { client_secret: { value: "ek_..." } }

async function startVoiceChat() {
  // 1. Get ephemeral key
  const res = await fetch("/api/token", { method: "POST" });
  const { client_secret } = await res.json();

  // 2. Set up WebRTC
  const pc = new RTCPeerConnection();

  // Audio playback
  const audio = document.createElement("audio");
  audio.autoplay = true;
  pc.ontrack = (e) => (audio.srcObject = e.streams[0]);

  // Mic capture
  const mic = await navigator.mediaDevices.getUserMedia({ audio: true });
  pc.addTrack(mic.getAudioTracks()[0], mic);

  // Data channel for events
  const dc = pc.createDataChannel("oai-events");
  dc.onopen = () => {
    dc.send(JSON.stringify({
      type: "session.update",
      session: {
        instructions: "You are a helpful assistant.",
        voice: "alloy",
        turn_detection: { type: "server_vad" },
      },
    }));
    document.getElementById("status").textContent = "🎤 Listening...";
  };

  dc.onmessage = (e) => {
    const evt = JSON.parse(e.data);
    if (evt.type === "response.audio_transcript.delta") {
      document.getElementById("transcript").textContent += evt.delta;
    } else if (evt.type === "response.done") {
      document.getElementById("transcript").textContent += "\n";
    }
  };

  // 3. SDP exchange
  const offer = await pc.createOffer();
  await pc.setLocalDescription(offer);

  const sdpRes = await fetch(
    "https://api.openai.com/v1/realtime?model=gpt-realtime",
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${client_secret.value}`,
        "Content-Type": "application/sdp",
      },
      body: offer.sdp,
    }
  );
  await pc.setRemoteDescription({
    type: "answer",
    sdp: await sdpRes.text(),
  });
}

startVoiceChat();

tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get the current weather for a location.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name, e.g. 'Tokyo'"
                }
            },
            "required": ["city"]
        }
    },
    {
        "type": "function",
        "name": "search_web",
        "description": "Search the web for current information.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query"
                }
            },
            "required": ["query"]
        }
    }
]

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "instructions": "You are a helpful assistant with access to weather and search tools.",
        "tools": tools,
        "tool_choice": "auto"
    }
}))

import httpx  # pip install httpx

async def execute_tool(name: str, args: dict) -> str:
    if name == "get_weather":
        # Replace with a real weather API
        return json.dumps({
            "city": args["city"],
            "temp": "22°C",
            "condition": "Partly cloudy"
        })
    elif name == "search_web":
        return json.dumps({
            "query": args["query"],
            "results": ["Result 1", "Result 2"]
        })
    return json.dumps({"error": "Unknown tool"})

async def receive_with_tools(ws):
    async for msg in ws:
        event = json.loads(msg)
        t = event["type"]

        if t == "response.function_call_arguments.done":
            name = event["name"]
            call_id = event["call_id"]
            args = json.loads(event["arguments"])

            print(f"🔧 Calling {name}({args})")
            result = await execute_tool(name, args)

            # Return the result
            await ws.send(json.dumps({
                "type": "conversation.item.create",
                "item": {
                    "type": "function_call_output",
                    "call_id": call_id,
                    "output": result
                }
            }))

            # Tell the model to continue
            await ws.send(json.dumps({
                "type": "response.create"
            }))

        elif t == "response.audio_transcript.delta":
            print(event["delta"], end="", flush=True)

        elif t == "response.done":
            print()

import asyncio, json, base64, os
import websockets

API_KEY = os.environ["OPENAI_API_KEY"]
TRANSLATE_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime-translate"

async def live_translate(target_language="Spanish"):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(
        TRANSLATE_URL, additional_headers=headers
    ) as ws:
        await ws.recv()  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "instructions": f"Translate all speech to {target_language}.",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",
                    "silence_duration_ms": 300
                }
            }
        }))
        await ws.recv()  # session.updated
        print(f"🌍 Translating to {target_language}...")

        # In production: stream mic audio in, play translated audio out
        # Same send/receive pattern as Module 2

        async for msg in ws:
            event = json.loads(msg)
            if event["type"] == "response.audio_transcript.delta":
                print(event["delta"], end="", flush=True)
            elif event["type"] == "response.done":
                print()

asyncio.run(live_translate("Japanese"))

import asyncio, json, base64, os, time
import websockets

API_KEY = os.environ["OPENAI_API_KEY"]
WHISPER_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime-whisper"

async def live_transcribe():
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(
        WHISPER_URL, additional_headers=headers
    ) as ws:
        await ws.recv()  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "input_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",
                    "silence_duration_ms": 500
                }
            }
        }))
        await ws.recv()  # session.updated
        print("📝 Transcribing... speak now.\n")

        # Mic capture (same pattern as Module 2)
        # ... send audio chunks ...

        transcript_lines = []
        current_line = ""
        start_time = time.time()

        async for msg in ws:
            event = json.loads(msg)
            t = event["type"]

            if t == "response.audio_transcript.delta":
                current_line += event["delta"]
                elapsed = time.time() - start_time
                # Overwrite the current line
                print(f"\r[{elapsed:6.1f}s] {current_line}", end="", flush=True)

            elif t == "response.done":
                if current_line.strip():
                    transcript_lines.append(current_line.strip())
                    print()  # finalize line
                current_line = ""

        # Save transcript
        with open("transcript.txt", "w") as f:
            f.write("\n".join(transcript_lines))

asyncio.run(live_transcribe())

// Uses WebRTC for low-latency browser transcription
async function startCaptioning() {
  const res = await fetch("/api/whisper-token", { method: "POST" });
  const { client_secret } = await res.json();

  const pc = new RTCPeerConnection();
  const mic = await navigator.mediaDevices.getUserMedia({ audio: true });
  pc.addTrack(mic.getAudioTracks()[0], mic);

  const dc = pc.createDataChannel("oai-events");
  const captionEl = document.getElementById("captions");

  dc.onopen = () => {
    dc.send(JSON.stringify({
      type: "session.update",
      session: {
        input_audio_format: "pcm16",
        turn_detection: { type: "server_vad" },
      },
    }));
  };

  let currentLine = "";
  dc.onmessage = (e) => {
    const evt = JSON.parse(e.data);
    if (evt.type === "response.audio_transcript.delta") {
      currentLine += evt.delta;
      captionEl.textContent = currentLine;
    } else if (evt.type === "response.done") {
      const p = document.createElement("p");
      p.textContent = currentLine;
      captionEl.parentElement.insertBefore(p, captionEl);
      currentLine = "";
      captionEl.textContent = "";
    }
  };

  // SDP exchange (same as Module 2 WebRTC pattern)
  const offer = await pc.createOffer();
  await pc.setLocalDescription(offer);
  const sdp = await fetch(
    "https://api.openai.com/v1/realtime?model=gpt-realtime-whisper",
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${client_secret.value}`,
        "Content-Type": "application/sdp",
      },
      body: offer.sdp,
    }
  );
  await pc.setRemoteDescription({ type: "answer", sdp: await sdp.text() });
}

pip install 'openai-agents[voice]'

from agents import Agent, function_tool
from agents.realtime import RealtimeAgent, RealtimeSession

# --- Tools ---
@function_tool
def check_order_status(order_id: str) -> str:
    """Look up the current status of a customer order."""
    # Replace with real database lookup
    statuses = {
        "ORD-001": "Shipped — arriving May 10",
        "ORD-002": "Processing — expected ship date May 12",
    }
    return statuses.get(order_id, f"Order {order_id} not found.")

@function_tool
def search_faq(question: str) -> str:
    """Search the FAQ knowledge base."""
    faqs = {
        "return": "Returns accepted within 30 days with receipt.",
        "shipping": "Free shipping on orders over $50.",
        "hours": "Support available Mon-Fri, 9 AM - 6 PM EST.",
    }
    for key, answer in faqs.items():
        if key in question.lower():
            return answer
    return "I couldn't find an answer. Let me connect you with a specialist."

@function_tool
def escalate_to_human(reason: str) -> str:
    """Escalate the conversation to a human agent."""
    return f"Escalation created: {reason}. A human agent will join shortly."

# --- Specialized Agents ---
order_agent = RealtimeAgent(
    name="Order Specialist",
    instructions=(
        "You handle order-related queries. Use check_order_status "
        "to look up orders. Be precise and helpful."
    ),
    tools=[check_order_status],
    model="gpt-realtime",
    voice="coral",
)

faq_agent = RealtimeAgent(
    name="FAQ Specialist",
    instructions=(
        "You answer general questions using the FAQ knowledge base. "
        "If you can't find an answer, escalate to a human."
    ),
    tools=[search_faq, escalate_to_human],
    model="gpt-realtime",
    voice="coral",
)

# --- Triage Agent (entry point) ---
triage_agent = RealtimeAgent(
    name="Support Triage",
    instructions=(
        "You are the front-line customer support agent. Greet the "
        "customer warmly. Determine if they need help with an order "
        "(hand off to Order Specialist) or have a general question "
        "(hand off to FAQ Specialist). If unsure, ask clarifying questions."
    ),
    handoffs=[order_agent, faq_agent],
    model="gpt-realtime",
    voice="alloy",
)

import asyncio

async def main():
    async with RealtimeSession(triage_agent) as session:
        print("🎧 Customer support agent is live!")
        print("   Speak to start a conversation.\n")
        await session.run()

asyncio.run(main())

from agents import InputGuardrail, GuardrailFunctionOutput

BLOCKED_TOPICS = ["competitor", "lawsuit", "internal"]

async def content_filter(ctx, agent, user_input):
    """Block off-topic or sensitive requests."""
    is_blocked = any(
        word in user_input.lower() for word in BLOCKED_TOPICS
    )
    return GuardrailFunctionOutput(
        output_info={"filtered": is_blocked},
        tripwire_triggered=is_blocked,
    )

# Add to any agent
safe_triage = RealtimeAgent(
    name="Safe Triage",
    instructions="You are a helpful support agent.",
    handoffs=[order_agent, faq_agent],
    input_guardrails=[
        InputGuardrail(guardrail_function=content_filter)
    ],
    model="gpt-realtime",
    voice="alloy",
)

import asyncio
from websockets.exceptions import ConnectionClosed

MAX_RETRIES = 5

async def resilient_agent():
    retries = 0
    while retries < MAX_RETRIES:
        try:
            async with RealtimeSession(triage_agent) as session:
                retries = 0
                await session.run()
        except ConnectionClosed as e:
            retries += 1
            wait = 2 ** retries
            print(f"⚠️  Disconnected ({e.code}). Retry {retries}/{MAX_RETRIES} in {wait}s...")
            await asyncio.sleep(wait)
        except KeyboardInterrupt:
            print("\n👋 Shutting down.")
            break
    else:
        print("❌ Max retries reached.")

asyncio.run(resilient_agent())