OpenAI Live WebSocket Integration

This guide will walk you through integrating Bandwidth's Programmable Voice API with OpenAI's Live API via Websocket. This integration allows you to leverage OpenAI's GPT-Live-1 model in your programmable call flows.

What you'll need

• A phone number associated to a Voice Configuration Package + a Programmable Voice Application
• Your OpenAI API Key (must have access to the Live API)
• Docker
• A publicly accessible server to host your webhook + websocket application (e.g., using ngrok)
• (Optional) Our sample application to get started: bandwidth-samples/openai-live-websockets-python

Call Flow

Before we dive in, let's walk through what an inbound call flow looks like with this integration.

This flow demonstrates a call that is answered by an AI agent and then transferred to a human agent. Let's break it down:

1. A caller dials your Bandwidth number.
2. Bandwidth sends a webhook to your server indicating an inbound call.
3. Your server responds with BXML containing a <StartStream> verb, instructing Bandwidth to start streaming audio to your Websocket Server.
4. Bandwidth initiates a stream event to your websocket server.
5. Your websocket server calls session.start on the OpenAI Live connection.
6. Bandwidth streams the caller's audio to your websocket server, which forwards it to OpenAI.
7. The caller has a conversation with the AI agent.
8. When the AI agent determines that the call should be transferred to a human agent, it uses OpenAI's tool calling mechanism.
9. Your websocket server receives the transfer instruction and makes a PUT /calls/{callId} request to Bandwidth to transfer the call.
10. Bandwidth transfers the call to the specified human agent.
11. The caller continues the conversation with the human agent.

Let's Build It!

For convenience - we have provided a sample application to get you started. You can find it here: bandwidth-samples/openai-live-websockets-python. The sample application is built using Python and FastAPI, but you can use any language or framework that you prefer, such as NodeJS + Express or Java + Spring.

To run the sample application, simply clone the repository and follow the instructions in the README.

The following sections will walk you through the sample application code to help you understand how it works.

Setup our Environment

Lets first clone our sample application:

git clone https://github.com/Bandwidth-Samples/openai-live-websockets-python
cd openai-live-websockets-python

The application provides a docker compose file to help you get started quickly, but you can also run the application via your local Python environment if you prefer.

First - ensure you have a .env file in the root of the project with the following variables:

export BW_ACCOUNT_ID="your_bw_account_id_here"
export BW_CLIENT_ID="your_bw_client_id_here"
export BW_CLIENT_SECRET="your_bw_client_secret_here"
export OPENAI_API_KEY="your_openai_api_key_here"
export TRANSFER_TO="+19195554321"
export BASE_URL="https://someNgrokId.ngrok-free.app"
export LOG_LEVEL="INFO"
export LOCAL_PORT=3000

Using Docker

docker compose up --build

Using Local Python Environment

python -m venv .venv
source .venv/bin/activate
cd app
pip install -r requirements.txt
python main.py

A successful startup should log the following:

INFO:     Will watch for changes in these directories: ['/app']
INFO:     Uvicorn running on http://0.0.0.0:3000 (Press CTRL+C to quit)
INFO:     Started reloader process [1] using WatchFiles
INFO:     Started server process [8]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

The application runs on port 3000 by default, but can be overridden by setting the LOCAL_PORT environment variable.

Creating our FastAPI Server

The sample application uses FastAPI to create a simple web server that can handle incoming HTTP requests from Bandwidth.

The sample application also provides a models directory that contains Pydantic models for the various Bandwidth webhook events. We wont define what those models look like here, but you can find them in the models directory of the sample application.

# main.py

# ...imports...

# Set our Environment Variables
try:
    BW_ACCOUNT = os.environ["BW_ACCOUNT_ID"]
    BW_CLIENT_ID = os.environ["BW_CLIENT_ID"]
    BW_CLIENT_SECRET = os.environ["BW_CLIENT_SECRET"]
    OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
    TRANSFER_TO = os.environ["TRANSFER_TO"]
    BASE_URL = os.environ["BASE_URL"]
    LOG_LEVEL = os.environ["LOG_LEVEL"].upper()
    LOCAL_PORT = int(os.environ.get("LOCAL_PORT", 3000))
except KeyError:
    print("environment variables not set")
    exit(1)

# OpenAI Live Client
openai_client = AsyncOpenAI(api_key=OPENAI_API_KEY)

app = FastAPI()

# Active OpenAI Live connections keyed by call_id
call_sessions: dict[str, AsyncLiveConnection] = {}


# Health Check
@app.get("/health", status_code=http.HTTPStatus.NO_CONTENT)
def health():
    return


# Handle Inbound Call Event from Bandwidth
@app.post("/webhooks/bandwidth/voice/initiate", status_code=http.HTTPStatus.OK)
def handle_initiate_event(callback: InitiateCallback) -> Response:
    return Response()


# Handle Inbound WebSocket Connection from Bandwidth
@app.websocket("/ws")
async def handle_inbound_websocket(bandwidth_websocket: WebSocket, call_id: str = None):
    return


def start_server(port: int) -> None:
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=port,
        log_level="info",
        reload=True,
    )


if __name__ == "__main__":
    start_server(LOCAL_PORT)

The above code snippet creates a simple FastAPI server with three endpoints:

1. A health check endpoint at /health that returns a 204 No Content status code.
2. A POST endpoint at /webhooks/bandwidth/voice/initiate that handles inbound call events from Bandwidth.
3. A WebSocket endpoint at /ws that handles an inbound WebSocket connection from Bandwidth for the Bi-Directional Audio Stream.

Handle Inbound Call Event

When a call is received on your Bandwidth number, Bandwidth will send a webhook to your server at the /webhooks/bandwidth/voice/initiate endpoint.

@app.post("/webhooks/bandwidth/voice/initiate", status_code=http.HTTPStatus.OK)
def handle_initiate_event(callback: InitiateCallback) -> Response:
    call_id = callback.call_id
    websocket_url = f"wss://{BASE_URL.replace('https://', '').replace('http://', '')}/ws"
    start_stream = StartStream(
        destination=f"{websocket_url}?call_id={call_id}",
        mode="bidirectional",
        name=call_id,
        destination_username="foo",
        destination_password="bar"
    )
    stop_stream = StopStream(name=call_id, wait="true")
    bxml_response = Bxml(nested_verbs=[start_stream, stop_stream])

    return Response(status_code=http.HTTPStatus.OK, content=bxml_response.to_bxml(), media_type="application/xml")

The above code snippet does the following:

1. Parses the incoming webhook payload into an InitiateCallback Pydantic model.
2. Extracts the callId from the webhook payload.
3. Constructs the WebSocket URL that Bandwidth will use to stream audio to your server.
4. Creates a <StartStream> verb with the WebSocket URL and other necessary parameters.
5. Creates a <StopStream> verb to stop the stream when the call ends.
6. Constructs a BXML response containing the <StartStream> and <StopStream> verbs.
7. Returns the BXML response to Bandwidth.

tip

Consider adding <StartRecording transcribe="true" /> before the <StartStream> verb to record and transcribe the call.

Handle Inbound WebSocket Connection

When Bandwidth receives the BXML response with the <StartStream> verb, it will initiate a WebSocket connection to your server at the /ws endpoint.

The sample application uses the OpenAI Python SDK (openai>=3.16.0) to connect to the Live API. The SDK manages the underlying WebSocket and exposes a typed AsyncLiveConnection object.

OPENAI_LIVE_MODEL = "gpt-live-1"

@app.websocket("/ws")
async def handle_inbound_websocket(bandwidth_websocket: WebSocket, call_id: str = None):
    await bandwidth_websocket.accept()

    if not call_id:
        await bandwidth_websocket.close(code=1008, reason="Missing call_id parameter")
        return

    async with openai_client.live.connect(max_retries=0) as connection:
        call_sessions[call_id] = connection
        try:
            await initialize_openai_session(connection)
            bw_task = asyncio.create_task(receive_from_bandwidth_ws(bandwidth_websocket, connection))
            oai_task = asyncio.create_task(receive_from_openai_ws(connection, bandwidth_websocket, call_id))
            done, pending = await asyncio.wait(
                [bw_task, oai_task], return_when=asyncio.FIRST_COMPLETED
            )
            for task in pending:
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
            for task in done:
                task.result()
        finally:
            call_sessions.pop(call_id, None)

The above code snippet does the following:

1. Accepts the incoming WebSocket connection from Bandwidth.
2. Extracts the call_id query parameter from the WebSocket URL.
3. Opens an OpenAI Live connection via the SDK (openai_client.live.connect).
4. Registers the connection in call_sessions so the hold endpoint can look it up.
5. Initializes the OpenAI session (see below).
6. Runs both receive loops as concurrent tasks, using asyncio.wait(FIRST_COMPLETED) so that when either side closes, the other is cleanly cancelled.

note

We use a call_id query parameter to correlate the WebSocket connection to the Bandwidth call. This call_id value is also present in the start websocket message from Bandwidth, but we need to provide it to the OpenAI connection for tool calls.

For demo purposes - the query parameter was the simplest way. Your implementation may vary, or not require the call id at all.

note

asyncio.wait(FIRST_COMPLETED) is important here. Using asyncio.gather instead would let both tasks keep running after one exits, which causes a "Cannot call send once a close message has been sent" error when the completed task's side has already closed.

Initialize the OpenAI Session

Unlike the Realtime API, GPT-Live requires an explicit session.start call immediately after connecting. This is where you configure the model, audio format, voice, and delegation settings.

The agent prompt lives in app/sample-prompt.md and is loaded at startup — keeping instructions in a separate file makes iteration easier without touching application code.

OPENAI_LIVE_MODEL = "gpt-live-1"
OPENAI_RESPONSES_MODEL = "gpt-5.6-terra"
AGENT_VOICE = "alloy"

with open("sample-prompt.md", "r") as file:
    AGENT_PROMPT = file.read()

TOOLS = [
    {
        "type": "function",
        "name": "transfer_call",
        "description": "Transfer the call to a live human agent. ONLY call this when the caller uses explicit phrases like 'speak to a human', 'transfer me', 'get me an agent', 'talk to a person', or 'get me a manager'. Never call this for questions, information requests, or searches — use web_search for those.",
        "parameters": {"type": "object", "properties": {}}
    },
    {
        "type": "web_search",
    }
]

async def initialize_openai_session(connection: AsyncLiveConnection):
    await connection.session.start(session={
        "model": OPENAI_LIVE_MODEL,
        "instructions": AGENT_PROMPT,
        "audio": {
            "format": {"type": "audio/pcmu", "rate": 8000},
            "output": {"voice": AGENT_VOICE}
        },
        "delegation": {
            "type": "responses",
            "responses": {
                "model": OPENAI_RESPONSES_MODEL,
                "instructions": AGENT_PROMPT,
                "tools": TOOLS,
                "tool_choice": "auto"
            }
        }
    })
    async for event in connection:
        if event.type == "session.started":
            break
        if event.type == "error":
            raise RuntimeError(f"OpenAI session start failed: {getattr(event, 'error', event)}")

    await connection.response.create()

info

GPT-Live uses G.711 μ-law (audio/pcmu) at 8000Hz for telephony integrations. This matches the audio format that Bandwidth streams over WebSocket, so no transcoding is required.

Delegation (Responses API)

The delegation block is what gives the Live session access to tools and a more capable reasoning model. GPT-Live-1 handles real-time audio; the delegation backend (OPENAI_RESPONSES_MODEL) handles tool selection and text generation. Tool calls and their results are surfaced back through the Live session as response.event messages.

After session.start, the app waits for session.started before calling connection.response.create() — this triggers the AI's opening greeting. Calling response.create() before session.started will fail.

Hosted Tools

The web_search tool is a hosted tool managed server-side by OpenAI — no handler is needed in your application code. When the model decides to search the web, OpenAI executes the search and returns results automatically. Only function tools (like transfer_call) require a handler on your side.

Broker the WebSocket Connections

The receive_from_bandwidth_ws and receive_from_openai_ws functions are responsible for brokering the audio and messages between Bandwidth and OpenAI.

async def receive_from_bandwidth_ws(bandwidth_websocket: WebSocket, connection: AsyncLiveConnection):
    try:
        async for message in bandwidth_websocket.iter_json():
            event = BandwidthStreamEvent.model_validate(message)
            match event.event_type:
                case StreamEventType.STREAM_STARTED:
                    logger.info(f"Stream started for call ID: {event.metadata.call_id}")
                case StreamEventType.MEDIA:
                    await connection.session.input_audio.append(audio=event.payload)
                case StreamEventType.STREAM_STOPPED:
                    return
                case _:
                    logger.warning(f"Unhandled event type: {event.event_type}")
    except Exception:
        pass
    finally:
        try:
            await bandwidth_websocket.close()
        except Exception:
            pass


async def receive_from_openai_ws(connection: AsyncLiveConnection, bandwidth_websocket: WebSocket, call_id: str):
    buf: dict[str, str] = {"input": "", "output": ""}
    last_speaker: str | None = None

    def flush(speaker: str) -> None:
        text = buf[speaker].strip()
        if text:
            logger.info(f"[{'Caller' if speaker == 'input' else 'Agent'}] {text}")
            buf[speaker] = ""

    try:
        async for event in connection:
            match event.type:
                case "session.output_audio.delta":
                    delta = getattr(event, "delta", None)
                    if delta:
                        audio_payload = base64.b64encode(base64.b64decode(delta)).decode("utf-8")
                        media = StreamMedia(content_type="audio/pcmu", payload=audio_payload)
                        play_audio_event = BandwidthStreamEvent(
                            event_type=StreamEventType.PLAY_AUDIO, media=media
                        )
                        await bandwidth_websocket.send_text(
                            play_audio_event.model_dump_json(by_alias=True, exclude_none=True)
                        )
                case "session.input_transcript.delta":
                    delta = getattr(event, "delta", "")
                    if last_speaker == "output":
                        flush("output")
                    buf["input"] += delta
                    last_speaker = "input"
                case "session.output_transcript.delta":
                    delta = getattr(event, "delta", "")
                    if last_speaker == "input":
                        flush("input")
                    buf["output"] += delta
                    last_speaker = "output"
                    if buf["output"].rstrip().endswith((".", "!", "?", "…")):
                        flush("output")
                case "response.event":
                    await handle_response_event(event, connection, call_id)
                case "session.closed":
                    if last_speaker:
                        flush(last_speaker)
                    return
                case "error":
                    logger.error(f"OpenAI Error: {getattr(event, 'error', event)}")
                case _:
                    logger.debug(f"Unhandled OpenAI event: {event.type}")
    except Exception as e:
        if str(e):
            logger.error(f"OpenAI connection error: {e}")

Bandwidth Messages

The receive_from_bandwidth_ws function listens for messages from Bandwidth and handles them based on the event type:

• For start events, it logs the start of the stream.
• For media events, it forwards audio to OpenAI via connection.session.input_audio.append(audio=...).
• For stop events, it returns — the asyncio.wait teardown in the WebSocket handler cancels the sibling task and closes the OpenAI connection cleanly.
• For unhandled event types, it logs a warning.

OpenAI Messages

The receive_from_openai_ws function iterates over the SDK connection directly — events arrive as typed objects with attribute access (event.type, event.delta), not raw JSON dicts.

• For session.output_audio.delta events, it encodes the audio delta and sends it to Bandwidth as a PLAY_AUDIO event.
• For session.input_transcript.delta events, it buffers the caller's spoken words. When the speaker switches, the previous buffer is flushed as a complete utterance.
• For session.output_transcript.delta events, it buffers the AI's generated speech and flushes at sentence boundaries (., !, ?, …).
• For response.event events, it delegates to handle_response_event to process nested Responses API events (including tool calls).
• For session.closed events, it flushes any remaining transcript buffer and returns.
• For error events, it logs the error.
• For unhandled event types, it logs a debug message.

Tools

Tool calls arrive nested inside response.event messages from the Responses delegation backend. The handle_response_event function extracts them; handle_tool_call executes them and always sends a result back so the model can continue.

async def handle_response_event(event, connection: AsyncLiveConnection, call_id: str):
    backend_event: dict = event.event
    event_type = backend_event.get("type")
    if event_type == "response.output_item.done":
        item = backend_event.get("item", {})
        if item.get("type") == "function_call":
            await handle_tool_call(item, connection, call_id)


async def handle_tool_call(item: dict, connection: AsyncLiveConnection, call_id: str):
    function_name = item.get("name")
    tool_call_id = item.get("id")

    result: str
    match function_name:
        case "transfer_call":
            transfer_bxml = Bxml([Transfer([PhoneNumber(TRANSFER_TO)])])
            try:
                bandwidth_voice_api_instance.update_call_bxml(BW_ACCOUNT, call_id, transfer_bxml.to_bxml())
                result = "success"
            except Exception as e:
                logger.error(f"Error transferring call: {e}")
                result = f"error: {e}"
        case _:
            logger.warning(f"Unhandled function call: {function_name}")
            result = f"error: unknown function {function_name}"

    await connection.response.item.create(item={
        "type": "function_call_output",
        "call_id": tool_call_id,
        "output": result,
    })
    await connection.response.create()

After executing the tool, the result is sent back via connection.response.item.create followed by connection.response.create(). This is required — without it, the model waits indefinitely for the tool output and never responds.

Tools are a powerful way to extend the capabilities of your AI agent. You can define as many function tools as you need and handle them in handle_tool_call. For hosted tools like web_search, OpenAI handles execution server-side — no handler is needed.

Call Hold (Mute / Unmute)

GPT-Live supports pausing and resuming audio input mid-call via the SDK's session.input_audio.mute() and session.input_audio.unmute() methods. This lets you implement hold music, agent coaching whisper, or any scenario where the caller's audio should stop reaching the model without disconnecting the call.

The sample application exposes a /webhooks/bandwidth/voice/hold endpoint that accepts a JSON body with a call_id and a hold boolean:

class HoldRequest(BaseModel):
    call_id: str
    hold: bool

# Active OpenAI Live connections keyed by call_id
call_sessions: dict[str, AsyncLiveConnection] = {}

@app.post("/webhooks/bandwidth/voice/hold", status_code=http.HTTPStatus.NO_CONTENT)
async def handle_hold_event(request: HoldRequest) -> None:
    connection = call_sessions.get(request.call_id)
    if not connection:
        raise HTTPException(status_code=404, detail=f"No active session for call_id: {request.call_id}")

    if request.hold:
        await connection.session.input_audio.mute()
    else:
        await connection.session.input_audio.unmute()

The WebSocket handler registers and deregisters the OpenAI connection so the hold endpoint can look it up:

@app.websocket("/ws")
async def handle_inbound_websocket(bandwidth_websocket: WebSocket, call_id: str = None):
    await bandwidth_websocket.accept()
    async with openai_client.live.connect(max_retries=0) as connection:
        call_sessions[call_id] = connection
        try:
            await initialize_openai_session(connection)
            # ... run receive tasks ...
        finally:
            call_sessions.pop(call_id, None)

To place a call on hold, send a POST request to the hold endpoint:

curl -X POST http://localhost:3000/webhooks/bandwidth/voice/hold \
  -H "Content-Type: application/json" \
  -d '{"call_id": "<your-call-id>", "hold": true}'

To remove the hold, send the same request with "hold": false.

note

The call_sessions dict is in-memory and per-process. If you run multiple application instances behind a load balancer, ensure the hold request routes to the same instance that owns the WebSocket connection, or use a shared store such as Redis.

Live Transcription

GPT-Live streams transcripts for both sides of the conversation in real time:

• session.input_transcript.delta — the caller's spoken words as recognized by the model.
• session.output_transcript.delta — the AI's generated speech, as text.

These arrive as incremental deltas. The sample application buffers them per speaker and flushes at sentence boundaries or when the speaker switches, so each logged line is a complete utterance rather than a word fragment:

case "session.input_transcript.delta":
    delta = getattr(event, "delta", "")
    if last_speaker == "output":
        flush("output")
    buf["input"] += delta
    last_speaker = "input"
case "session.output_transcript.delta":
    delta = getattr(event, "delta", "")
    if last_speaker == "input":
        flush("input")
    buf["output"] += delta
    last_speaker = "output"
    if buf["output"].rstrip().endswith((".", "!", "?", "…")):
        flush("output")

You can extend this to write transcripts to a database, emit them to a WebSocket client, or feed them into a downstream analytics pipeline.

Connect to your Public Server

Now that we have our application running locally, we need to expose it to the internet so that Bandwidth can send webhook events and stream audio to it. You can use a tool like ngrok to create a secure tunnel to your local server.

In a new terminal window, run the following command:

ngrok http 3000

This will give you a public URL that you can use to configure your Bandwidth Voice Application.

info

The URL generated by NGROK is what you will use as your BASE_URL environment variable. Ngrok must be started before running docker compose up if you are running the sample application.

Configure your Programmable Voice Application

This guide assumes that you have created a Programmable Voice Application and associated it with a phone number. If you haven't done this yet, please refer to the account setup guide to learn more.

Once your application is created - you will set the Inbound Call webhook URL to point to your public server's /webhooks/bandwidth/voice/initiate endpoint and the Status Callback URL to your /webhooks/bandwidth/voice/status endpoint.

http://someNgrokId.ngrok-free.app/webhooks/bandwidth/voice/initiate
http://someNgrokId.ngrok-free.app/webhooks/bandwidth/voice/status

Test the Integration

Now that everything is set up, you can test the integration by calling your Bandwidth phone number. You should be connected to the AI agent, and you can have a conversation with it. When you say "transfer me to a human agent", the call should be transferred to the number specified in the TRANSFER_TO environment variable.