OpenAI Realtime API Function Calling in Voice Sessions

import asyncio
import base64
import json
import os
import websockets


OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]


# ✅ Tool implementations
async def get_weather(city: str) -> dict:
    """Async tool — call external API, return JSON-serializable result."""
    # ... real API call here ...
    return {"city": city, "temp_c": 22, "condition": "Sunny"}


async def check_availability(check_in: str, check_out: str) -> dict:
    # ... your DB call ...
    return {"available": True, "rate_usd": 189}


TOOLS_IMPL = {
    "get_weather": get_weather,
    "check_availability": check_availability,
}


# ✅ Session config — register tools + disable auto-response so we drive manually
SESSION_CONFIG = {
    "type": "session.update",
    "session": {
        "instructions": (
            "You are a concise voice concierge for a hotel. "
            "Use tools when the user asks about weather or availability. "
            "Speak the tool results naturally, don't recite JSON."
        ),
        "audio": {
            "input":  {"format": "pcm16"},
            "output": {"format": "pcm16", "voice": "cedar"},
        },
        "turn_detection": {
            "type": "server_vad",
            "silence_duration_ms": 200,
            "create_response": True,      # user speech → auto-respond
            "interrupt_response": True,
        },
        "tools": [
            {
                "type": "function",
                "name": "get_weather",
                "description": "Get current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {"city": {"type": "string"}},
                    "required": ["city"],
                },
            },
            {
                "type": "function",
                "name": "check_availability",
                "description": "Check hotel availability for dates",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "check_in":  {"type": "string", "format": "date"},
                        "check_out": {"type": "string", "format": "date"},
                    },
                    "required": ["check_in", "check_out"],
                },
            },
        ],
        "tool_choice": "auto",
    },
}


async def main():
    url = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
    fn_arg_buffers: dict[str, list[str]] = {}   # item_id → JSON arg chunks
    fn_call_meta:   dict[str, dict]      = {}   # item_id → {name, call_id}

    async with websockets.connect(
        url,
        additional_headers={
            "Authorization": f"Bearer {OPENAI_API_KEY}",
            "OpenAI-Beta": "realtime=v1",
        },
        max_size=None,
    ) as ws:

        # 1. Configure the session
        await ws.send(json.dumps(SESSION_CONFIG))

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

            if t == "response.output_item.added":
                item = event["item"]
                if item.get("type") == "function_call":
                    fn_call_meta[item["id"]] = {
                        "name": item["name"],
                        "call_id": item["call_id"],
                    }
                    fn_arg_buffers[item["id"]] = []

            elif t == "response.function_call_arguments.delta":
                # BUFFER, don't parse
                fn_arg_buffers.setdefault(event["item_id"], []).append(
                    event.get("delta", "")
                )

            elif t == "response.function_call_arguments.done":
                # ✅ FULL 4-STEP TOOL HANDLING
                await handle_tool_call(
                    ws=ws,
                    call_id=event["call_id"],
                    name=event["name"],
                    arguments_json=event["arguments"],   # full JSON here
                )

            elif t == "response.output_audio.delta":
                # base64 audio to speaker
                audio_bytes = base64.b64decode(event["delta"])
                await your_audio_sink(audio_bytes)

            elif t == "error":
                print(f"ERROR: {event['error']}")


async def handle_tool_call(ws, call_id: str, name: str, arguments_json: str):
    """The 4-step dance: parse → execute → output → create."""

    # Step 1 — parse
    try:
        args = json.loads(arguments_json)
    except json.JSONDecodeError as e:
        return await send_tool_error(ws, call_id, f"Invalid arguments: {e}")

    # Step 2 — execute
    impl = TOOLS_IMPL.get(name)
    if impl is None:
        return await send_tool_error(ws, call_id, f"Unknown tool: {name}")

    try:
        result = await impl(**args)
        # Compact output — voice models struggle with huge JSON
        output_str = json.dumps(result, ensure_ascii=False)[:800]
    except Exception as e:
        output_str = json.dumps({"error": str(e)})

    # Step 3 — send function_call_output as a conversation item
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": call_id,              # MUST match the model's call_id
            "output": output_str,
        },
    }))

    # Step 4 — trigger response.create so the model speaks the result
    await ws.send(json.dumps({"type": "response.create"}))


async def send_tool_error(ws, call_id: str, err: str):
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({"error": err}),
        },
    }))
    await ws.send(json.dumps({"type": "response.create"}))


if __name__ == "__main__":
    asyncio.run(main())

from datetime import date
import json


# ❌ BAD — voice model may read this literally
def bad_availability(check_in: str, check_out: str) -> dict:
    return {
        "success": True,
        "data": {
            "available_rooms": [
                {"room_id": "STD-001", "type": "standard", "price_per_night_usd": 189.00, "capacity_adults": 2, "amenities_list": ["wifi", "tv", "safe"]},
                {"room_id": "STD-002", "type": "standard", "price_per_night_usd": 189.00, "capacity_adults": 2, "amenities_list": ["wifi", "tv", "safe"]},
            ],
            "total_nights": 3,
            "estimated_total_usd": 567.00,
        },
        "metadata": {"request_id": "req_abc123", "cached": False, "ttl_seconds": 300},
    }


# ✅ GOOD — voice-friendly, short, natural language
def good_availability(check_in: str, check_out: str) -> str:
    # Do the real query
    rooms_available = query_db_for_availability(check_in, check_out)
    nights = (date.fromisoformat(check_out) - date.fromisoformat(check_in)).days

    if not rooms_available:
        return "No rooms are available for those dates."

    price = rooms_available[0].nightly_rate
    total = price * nights
    return (
        f"Yes, standard rooms are available for those {nights} nights "
        f"at ${price} per night — about ${total} total."
    )


# ✅ Structured but minimal — model will paraphrase
def compact_availability(check_in: str, check_out: str) -> dict:
    return {
        "available": True,
        "nights": 3,
        "rate_per_night_usd": 189,
        "total_usd": 567,
    }


# ✅ Truncation helper for long results
MAX_TOOL_OUTPUT_CHARS = 800

def voice_safe(output) -> str:
    """Format tool output for voice — string, truncated, no raw IDs."""
    if isinstance(output, str):
        text = output
    else:
        text = json.dumps(output, ensure_ascii=False)

    if len(text) > MAX_TOOL_OUTPUT_CHARS:
        text = text[:MAX_TOOL_OUTPUT_CHARS] + " [more details available in the app]"

    return text


# ✅ Summary + optional deep-drill pattern for voice
def summarize_and_offer_details(items: list) -> str:
    """Summarize for voice; store details for later drill-down."""
    if not items:
        return "I didn't find any matches."

    if len(items) == 1:
        it = items[0]
        return f"I found one match: {it.name} at ${it.price}."

    return (
        f"I found {len(items)} matches. The top three are: "
        + ", ".join(f"{it.name} at ${it.price}" for it in items[:3])
        + ". Would you like details on any of them?"
    )

# ✅ Manual-response mode — for tool-heavy workflows
MANUAL_RESPONSE_SESSION = {
    "type": "session.update",
    "session": {
        "audio": {"input": {"format": "pcm16"}, "output": {"format": "pcm16"}},
        "turn_detection": {
            "type": "server_vad",
            "silence_duration_ms": 200,
            "create_response": False,        # ← don't auto-respond
            "interrupt_response": True,
        },
        "tools": [...],
    },
}


async def with_manual_responses(ws):
    """You control WHEN the model responds — after user speech OR after tool output."""

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

        if t == "input_audio_buffer.speech_stopped":
            # User done speaking — decide whether to respond directly or wait
            await ws.send(json.dumps({"type": "response.create"}))

        elif t == "response.function_call_arguments.done":
            # Handle the tool then trigger response
            await handle_tool_call(ws, event["call_id"], event["name"], event["arguments"])
            # response.create is inside handle_tool_call


# ✅ Auto-response mode — for chat-heavy workflows with occasional tools
AUTO_RESPONSE_SESSION = {
    "type": "session.update",
    "session": {
        "audio": {"input": {"format": "pcm16"}, "output": {"format": "pcm16"}},
        "turn_detection": {
            "type": "server_vad",
            "silence_duration_ms": 200,
            "create_response": True,         # ← auto-respond to speech
            "interrupt_response": True,
        },
        "tools": [...],
    },
}


# In auto-response mode, tools still need response.create after output,
# because the auto-response for the user's turn has already completed
# (with the tool call decision), and a new response is needed to speak
# the tool result. The interrupt_response=True setting handles the case
# where a user starts talking before the model can respond.


# ✅ Response cancellation — user interrupted mid-response
# Server sends: response.done with status="cancelled" when interrupted
# Your handler:
async def on_response_cancelled(ws, event):
    """User interrupted; discard pending audio buffer on client side."""
    await ws.send(json.dumps({"type": "input_audio_buffer.clear"}))


# ✅ Explicit conversation.item.truncate for partial audio delivery
# When you know the user interrupted mid-word:
await ws.send(json.dumps({
    "type": "conversation.item.truncate",
    "item_id": current_response_item_id,
    "content_index": 0,
    "audio_end_ms": played_ms,         # how much audio actually reached the user
}))


# ✅ Avoid response.create while a response is already in progress
# Server rejects with:
#   error: {"message": "A response is already active. Cancel it before starting a new one."}
# Track state:
class ResponseState:
    def __init__(self):
        self.active = False

    async def start(self, ws):
        if self.active:
            # Cancel first
            await ws.send(json.dumps({"type": "response.cancel"}))
        self.active = True
        await ws.send(json.dumps({"type": "response.create"}))

    def on_done(self):
        self.active = False