Speech to Speech

import asyncio
import json
import os
import websockets

async def voice_agent():
    async with websockets.connect(
        "wss://api.x.ai/v1/realtime?model=grok-voice-latest",
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    ) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "voice": "eve",
                "instructions": "You are a helpful assistant.",
                "turn_detection": {"type": "server_vad"}
            }
        }))
        
        # Send a text message
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {"type": "message", "role": "user", 
                     "content": [{"type": "input_text", "text": "Hello!"}]}
        }))
        await ws.send(json.dumps({"type": "response.create"}))
        
        # Receive audio/text responses
        async for msg in ws:
            event = json.loads(msg)
            print(f"Event: {event['type']}")

asyncio.run(voice_agent())

MODEL = "grok-voice-latest"
url = f"wss://api.x.ai/v1/realtime?model={MODEL}"

# Configure session with a specific voice
session_config = {
    "type": "session.update",
    "session": {
        "voice": "eve",  # any built-in voice or custom voice ID
        "instructions": "You are a helpful assistant.",
        # Audio format settings (these are the defaults if not specified)
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": 24000}},
            "output": {"format": {"type": "audio/pcm", "rate": 24000}}
        }
    }
}

await ws.send(json.dumps(session_config))

# Configure audio format with custom sample rate for input and output
session_config = {
    "type": "session.update",
    "session": {
        "audio": {
            "input": {
                "format": {
                    "type": "audio/pcm",  # or "audio/pcmu" or "audio/pcma"
                    "rate": 16000  # Only applicable for audio/pcm
                }
            },
            "output": {
                "format": {
                    "type": "audio/pcm",  # or "audio/pcmu" or "audio/pcma"
                    "rate": 16000  # Only applicable for audio/pcm
                }
            }
        },
        "instructions": "You are a helpful assistant.",
    }
}

await ws.send(json.dumps(session_config))

session_config = {
    "type": "session.update",
    "session": {
        "audio": {
            "input": {
                "format": {"type": "audio/pcm", "rate": 24000},
                "transport": "binary",
            },
            "output": {
                "format": {"type": "audio/pcm", "rate": 24000},
                "transport": "binary",
            },
        },
    },
}
await ws.send(json.dumps(session_config))

# Send mic audio as raw PCM16 little-endian frames (not base64 JSON)
await ws.send(pcm16_bytes)  # WebSocket binary message

# Receive: binary messages are audio; text messages are JSON events
async for message in ws:
    if isinstance(message, bytes):
        # raw PCM16 (or Opus packets if format is audio/opus)
        play(message)
    else:
        event = json.loads(message)
        # response.done, transcripts, etc.

import base64
import numpy as np

# Configure session with 16kHz sample rate for lower bandwidth (input and output)
session_config = {
    "type": "session.update",
    "session": {
        "instructions": "You are a helpful assistant.",
        "voice": "eve",
        "turn_detection": {
            "type": "server_vad",
        },
        "audio": {
            "input": {
                "format": {
                    "type": "audio/pcm",
                    "rate": 16000  # 16kHz for lower bandwidth usage
                }
            },
            "output": {
                "format": {
                    "type": "audio/pcm",
                    "rate": 16000  # 16kHz for lower bandwidth usage
                }
            }
        }
    }
}
await ws.send(json.dumps(session_config))

# When processing audio, use the same sample rate
SAMPLE_RATE = 16000

# Convert audio data to PCM16 and base64
def audio_to_base64(audio_data: np.ndarray) -> str:
    """Convert float32 audio array to base64 PCM16 string."""
    # Normalize to [-1, 1] and convert to int16
    audio_int16 = (audio_data * 32767).astype(np.int16)
    # Encode to base64
    audio_bytes = audio_int16.tobytes()
    return base64.b64encode(audio_bytes).decode('utf-8')

# Convert base64 PCM16 to audio data
def base64_to_audio(base64_audio: str) -> np.ndarray:
    """Convert base64 PCM16 string to float32 audio array."""
    # Decode base64
    audio_bytes = base64.b64decode(base64_audio)
    # Convert to int16 array
    audio_int16 = np.frombuffer(audio_bytes, dtype=np.int16)
    # Normalize to [-1, 1]
    return audio_int16.astype(np.float32) / 32768.0

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "voice": "eve",
        "instructions": "You are a helpful assistant.",
        "replace": {"Acme Mobile": "Acme Mobull"}
    }
}))

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "audio": {
            "input": {
                "transcription": {
                    "language_hint": "ja"
                }
            }
        }
    }
}))

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "audio": {
            "input": {
                "transcription": {
                    "keyterms": ["xAI", "Grok", "Understand The Universe"]
                }
            }
        }
    }
}))

COLLECTION_ID = "your-collection-id"  # Replace with your collection ID

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "file_search",
                "vector_store_ids": [COLLECTION_ID],
                "max_num_results": 10,
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "web_search",
                "allowed_domains": ["x.ai", "docs.x.ai"],
                "location": {"country": "US", "city": "San Francisco"},
            },
            {
                "type": "x_search",
                "allowed_x_handles": ["xai"],
                "from_date": "2025-01-01",
                "to_date": "2025-06-01",
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "mcp",
                "server_url": "https://mcp.example.com/mcp",
                "server_label": "my-tools",
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "mcp",
                "server_url": "https://mcp.example.com/mcp",
                "server_label": "my-tools",
                "server_description": "Custom business tools for order management",
                "allowed_tools": ["lookup_order", "check_inventory"],
                "authorization": "Bearer your-token-here",
                "headers": {
                    "X-Custom-Header": "value"
                },
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "mcp",
                "server_url": "https://mcp.deepwiki.com/mcp",
                "server_label": "deepwiki",
            },
            {
                "type": "mcp",
                "server_url": "https://your-tools.example.com/mcp",
                "server_label": "custom-tools",
                "allowed_tools": ["search_database", "format_data"],
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "function",
                "name": "generate_random_number",
                "description": "Generate a random number between min and max values",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "min": {
                            "type": "number",
                            "description": "Minimum value (inclusive)",
                        },
                        "max": {
                            "type": "number",
                            "description": "Maximum value (inclusive)",
                        },
                    },
                    "required": ["min", "max"],
                },
            },
        ],
    },
}

session_config = {
    "type": "session.update",
    "session": {
        ...
        "tools": [
            {
                "type": "file_search",
                "vector_store_ids": ["your-collection-id"],
                "max_num_results": 10,
            },
            {
                "type": "web_search",
            },
            {
                "type": "x_search",
            },
            {
                "type": "mcp",
                "server_url": "https://mcp.example.com/mcp",
                "server_label": "my-tools",
            },
            {
                "type": "function",
                "name": "generate_random_number",
                "description": "Generate a random number",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "min": {"type": "number"},
                        "max": {"type": "number"},
                    },
                    "required": ["min", "max"],
                },
            },
        ],
    },
}

import json
import websockets

# Define your function implementations
def get_weather(location: str, units: str = "celsius"):
    """Get current weather for a location"""
    # In production, call a real weather API
    return {
        "location": location,
        "temperature": 22,
        "units": units,
        "condition": "Sunny",
        "humidity": 45
    }

def book_appointment(date: str, time: str, service: str):
    """Book an appointment"""
    # In production, interact with your booking system
    import random
    confirmation = f"CONF{random.randint(1000, 9999)}"
    return {
        "status": "confirmed",
        "confirmation_code": confirmation,
        "date": date,
        "time": time,
        "service": service
    }

# Map function names to implementations
FUNCTION_HANDLERS = {
    "get_weather": get_weather,
    "book_appointment": book_appointment
}

async def handle_function_call(ws, event):
    """Handle function call from the voice agent"""
    function_name = event["name"]
    call_id = event["call_id"]
    arguments = json.loads(event["arguments"])

    print(f"Function called: {function_name} with args: {arguments}")

    # Execute the function
    if function_name in FUNCTION_HANDLERS:
        result = FUNCTION_HANDLERS[function_name](**arguments)

        # Send result back to agent
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            }
        }))

        # Request agent to continue with the result
        await ws.send(json.dumps({
            "type": "response.create"
        }))
    else:
        print(f"Unknown function: {function_name}")

# In your WebSocket message handler
async def on_message(ws, message):
    event = json.loads(message)

    # Listen for function calls
    if event["type"] == "response.function_call_arguments.done":
        await handle_function_call(ws, event)
    elif event["type"] == "response.output_audio.delta":
        # Handle audio response
        pass

await ws.send(json.dumps({
    "type": "conversation.item.create",
    "item": {
        "type": "force_message",
        "role": "assistant",
        "interruptible": False,
        "content": [{"type": "output_text", "text": "This call is being recorded."}]
    }
}))
# Do NOT send response.create — the force_message IS the turn.

await ws.send(json.dumps({
    "type": "response.create",
    "response": {
        "instructions": "Respond in Spanish for this turn only."
    }
}))

import json, websockets

# resume_id is None on a fresh conversation; pass the saved id to resume.
async def connect(resume_id=None):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    if resume_id:
        url += f"&conversation_id={resume_id}"

    async with websockets.connect(url, additional_headers=headers) as ws:
        # Opt in to resumption (required to cache and to replay).
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {"resumption": {"enabled": True}},
        }))

        async for raw in ws:
            event = json.loads(raw)
            if event["type"] == "conversation.created":
                # Server-assigned id. Save it and pass it as
                # ?conversation_id= on your next connect to resume.
                saved_id = event["conversation"]["id"]
            # ... handle the rest of the session

// 1. Immediately request mic access and start capturing
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });

const audioContext = new AudioContext({ sampleRate: 24000 });

const source = audioContext.createMediaStreamSource(stream);
const processor = audioContext.createScriptProcessor(4096, 1, 1); // or AudioWorklet for better perf

source.connect(processor);
processor.connect(audioContext.destination); // optional

// Buffer incoming PCM data immediately
let earlyAudioBuffer = []; // Float32Array[] or Int16Array[]

processor.onaudioprocess = (e) => {
  const input = e.inputBuffer.getChannelData(0);
  earlyAudioBuffer.push(new Float32Array(input)); // or convert to PCM16
};

// 2. In parallel – connect WebSocket (may take time)
const ws = new WebSocket("wss://api.x.ai/v1/realtime?model=grok-voice-latest", [
  `xai-client-secret.${token}`,
]);

ws.onopen = () => {
  // Send session.update configuration
  ws.send(JSON.stringify({ type: "session.update", session: { ... } }));

  // Flush any buffered audio now that we're connected
  if (earlyAudioBuffer.length > 0) {
    flushBufferedAudioToWS(earlyAudioBuffer);
    earlyAudioBuffer = [];
  }
};

ws.on("message", async (message) => {
  const event = JSON.parse(message);

  if (event.type === "response.function_call_arguments.done") {
    // 1. Execute the tool
    const result = await executeFunction(event.name, JSON.parse(event.arguments));

    // 2. Send the function result immediately
    ws.send(JSON.stringify({
      type: "conversation.item.create",
      item: {
        type: "function_call_output",
        call_id: event.call_id,
        output: JSON.stringify(result),
      },
    }));

    // 3. Show a "thinking" indicator in the UI
    showThinkingIndicator();

    // 4. Wait for current audio playback to finish
    await waitForPlaybackComplete();

    // 5. Now request the next response
    ws.send(JSON.stringify({ type: "response.create" }));
    hideThinkingIndicator();
  }
});

import asyncio
from openai import AsyncOpenAI

# Before (OpenAI)
# client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

# After (xAI)
client = AsyncOpenAI(
    api_key=os.environ["XAI_API_KEY"],
    base_url="https://api.x.ai/v1",
)

async def main():
    async with client.realtime.connect(
        model="grok-voice-latest"
    ) as conn:
        await conn.session.update(session={
            "voice": "eve",
            "instructions": "You are a helpful assistant.",
            "turn_detection": {"type": "server_vad"},
        })
        # ... rest of your application code

asyncio.run(main())

import os
import websockets

# Before (OpenAI)
# url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
# headers = {"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"}

# After (xAI)
url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
headers = {"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}

async with websockets.connect(url, additional_headers=headers) as ws:
    # Your existing event handling code works as-is
    pass

# Pass the model in connect()
async with client.realtime.connect(model="grok-voice-latest") as conn:
    ...