realtime-websocket-conversation 0.1.2

Python library that provides a websocket endpoint handler for FastAPI to stream between Twilio and OpenAI realtime.

realtime-websocket-conversation

A Python package for building real-time, streaming websocket conversations between Twilio and OpenAI using FastAPI. This library provides a websocket handler that streams audio between a Twilio client and an OpenAI model, enabling interactive voice assistants and conversational AI applications.

The goal of this package is really to reduce the boilerplate code needed to set up a real-time conversation with OpenAI's models.

Features

• Real-time audio streaming between Twilio and OpenAI
• FastAPI websocket integration
• Supports OpenAI function calling with simple @tool decorator (modified & borrowed from OpenAI's Agent SDK)
  • In your tool function signatures, you can use state and connection parameters to access mutable state and the OpenAI websocket connection (in case you need to update session settings). If you don't need these, use **kwargs in your function signature, as the decorator will pass them anyway.
• Handles interruptions and audio truncation for natural conversations
• Optional callback functions for user and assistant message transcription events

Installation

You can install the package using uv:

uv add realtime-websocket-conversation

or with pip:

pip install realtime-websocket-conversation

Usage Example

FastAPI Integration

from fastapi import FastAPI, Request, WebSocket
from fastapi.responses import HTMLResponse
from openai import AsyncOpenAI
from realtime_websocket_conversation import handle_websocket_conversation, tool
from twilio.twiml.voice_response import Connect, VoiceResponse

app = FastAPI()
openai_client = AsyncOpenAI()

@app.get("/incoming-call", response_class=HTMLResponse)
async def incoming_call(request: Request):
    response = VoiceResponse()
    response.say("Connecting your call...")
    host = (
        request.headers.get("X-Forwarded-Host")
        or request.headers.get("Host")
        or request.url.hostname
    )
    connect = Connect()
    connect.stream(url=f"wss://{host}/media-stream")
    response.append(connect)
    return HTMLResponse(content=str(response), media_type="application/xml")

@app.websocket("/media-stream")
async def media_stream(websocket: WebSocket):
    await handle_websocket_conversation(
        websocket=websocket,
        openai_client=openai_client,
        model="gpt-4o-realtime-preview",
        voice="alloy",
        system_message="You are a helpful chat assistant.",
        temperature=0.8,
        turn_detection=None,  # Optional turn detection config, defaults to {"type": "server_vad"}, fed directly to session settings
        tools=[],  # Optionally pass tool functions
        state={},  # Optionally pass a mutable state object, to be passed to tool functions
        on_user_message=None,  # Optional callback when user message transcription completes
        on_assistant_message=None,  # Optional callback when assistant message transcription completes
    )

Point your Twilio webhook to the /incoming-call endpoint, and the websocket connection will be established when a call is received.

Registering a Tool Function

from realtime_websocket_conversation import tool

@tool
def get_time(state=None, connection=None):
    from datetime import datetime
    return {"time": datetime.now().isoformat()}

Add your tool functions to the tools list when calling handle_websocket_conversation, i.e. [get_time].