Skip to content
For the complete documentation index, see llms.txt. Markdown versions of documentation pages are available by appending .md to the page URL.
Primary navigation
Theme

Connect

The Realtime API enables you to build low-latency, multi-modal conversational experiences. It currently supports text and audio as both input and output, as well as function calling.

Some notable benefits of the API include:

  • Native speech-to-speech: Skipping an intermediate text format means low latency and nuanced output.
  • Natural, steerable voices: The models have natural inflection and can laugh, whisper, and adhere to tone direction.
  • Simultaneous multimodal output: Text is useful for moderation; faster-than-realtime audio ensures stable playback.

The Realtime API is a stateful, event-based API that communicates over a WebSocket. To start a Realtime conversation, set model in the connection URL. To join an existing call, set call_id instead. Transcription sessions use intent=transcription.

WebSocket wss://api.openai.com/v1/realtime

Connect

import asyncio

from openai import AsyncOpenAI


async def main():
    async with AsyncOpenAI() as client:
        async with client.realtime.connect(model="gpt-realtime") as connection:
            await connection.session.update(
                session={"type": "realtime", "output_modalities": ["text"]}
            )
            await connection.conversation.item.create(
                item={
                    "type": "message",
                    "role": "user",
                    "content": [
                        {"type": "input_text", "text": "Say hello in one sentence."}
                    ],
                }
            )
            await connection.response.create()
            async for event in connection:
                if event.type == "response.output_text.delta":
                    print(event.delta, end="", flush=True)
                elif event.type == "response.done":
                    if event.response.status != "completed":
                        raise RuntimeError(f"Response ended: {event.response.status}")
                    print()
                    return
                elif event.type == "error":
                    raise RuntimeError(event.error.message)
            raise RuntimeError("Connection closed before the response completed.")


asyncio.run(asyncio.wait_for(main(), 60))