Implementation:Openai Openai python Realtime Connect

Knowledge Sources	openai-python OpenAI Realtime API
Domains	Realtime_Communication, WebSocket
Last Updated	2026-02-15 00:00 GMT

Overview

Concrete tool for establishing WebSocket connections to the OpenAI Realtime API provided by the OpenAI Python SDK.

Description

The AsyncRealtime resource provides a connect() method that returns an AsyncRealtimeConnectionManager async context manager. When entered, it establishes a WebSocket connection and yields an AsyncRealtimeConnection with typed sub-resources for session management, conversation items, responses, and audio buffers.

Usage

Use as an async context manager with async with client.beta.realtime.connect(model=...) as connection:. Requires AsyncOpenAI client.

Code Reference

Source Location

Repository: openai-python
File: src/openai/resources/beta/realtime/realtime.py
Lines: L121-174 (AsyncRealtime class), L229-303 (AsyncRealtimeConnection), L306-410 (AsyncRealtimeConnectionManager)

Signature

class AsyncRealtime(AsyncAPIResource):
    def connect(
        self,
        *,
        model: str,
        extra_query: Query = {},
        extra_headers: Headers = {},
        websocket_connection_options: WebsocketConnectionOptions = {},
    ) -> AsyncRealtimeConnectionManager:
        """
        Opens a WebSocket connection to the Realtime API.

        Args:
            model: Realtime model ID (e.g., "gpt-4o-realtime-preview").
            extra_query: Additional query parameters.
            extra_headers: Additional HTTP headers.
            websocket_connection_options: WebSocket config (timeout, etc.).
        """

class AsyncRealtimeConnection:
    """Active WebSocket connection with sub-resources."""
    session: AsyncRealtimeSessionResource
    conversation: AsyncRealtimeConversationResource
    response: AsyncRealtimeResponseResource
    input_audio_buffer: AsyncRealtimeInputAudioBufferResource

    async def __aiter__(self) -> AsyncIterator[RealtimeServerEvent]: ...
    async def recv(self) -> RealtimeServerEvent: ...
    async def recv_bytes(self) -> bytes: ...
    async def close(self) -> None: ...

Import

from openai import AsyncOpenAI
# Access via client.beta.realtime.connect()

I/O Contract

Inputs

Name	Type	Required	Description
model	str	Yes	Realtime model ID (e.g., "gpt-4o-realtime-preview")
extra_query	Query	No	Additional query parameters
extra_headers	Headers	No	Additional HTTP headers
websocket_connection_options	WebsocketConnectionOptions	No	WebSocket configuration (timeout, additional_headers)

Outputs

Name	Type	Description
connection	AsyncRealtimeConnection	Active connection with .session, .conversation, .response, .input_audio_buffer

Usage Examples

Basic Connection

import asyncio
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()
    async with client.beta.realtime.connect(
        model="gpt-4o-realtime-preview",
    ) as connection:
        # Connection is now active
        await connection.session.update(session={
            "modalities": ["text"],
        })
        # ... interact with the connection
        async for event in connection:
            print(event.type)
            break

asyncio.run(main())

Related Pages

Implements Principle

Principle:Openai_Openai_python_Realtime_WebSocket_Connection

Requires Environment

Uses Heuristic

Heuristic:Openai_Openai_python_Streaming_Resource_Management

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment