Grok Realtime

​

Overview

​

Installation

uv add "pipecat-ai[grok]"

​

Prerequisites

​

xAI Account Setup

1. xAI Account: Sign up at xAI Console
2. API Key: Generate a Grok API key from your account dashboard
3. Model Access: Ensure access to Grok Voice Agent models
4. Usage Limits: Configure appropriate usage limits and billing

​

Required Environment Variables

• XAI_API_KEY: Your xAI API key for authentication

​

Key Features

• Real-time Speech-to-Speech: Direct audio processing with low latency
• Multilingual Support: Support for multiple languages
• Voice Activity Detection: Server-side VAD for automatic speech detection
• Function Calling: Seamless support for external functions and tool integration
• Multiple Voice Options: Various voice personalities available
• WebSocket Support: Real-time bidirectional audio streaming

​

Configuration

​

GrokRealtimeLLMService

​

Settings

​

SessionProperties

​

TurnDetection

​

AudioConfiguration

​

Built-in Tools

​

Usage

​

Basic Setup

import os
from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService

llm = GrokRealtimeLLMService(
    api_key=os.getenv("XAI_API_KEY"),
)

​

With Session Configuration

from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService
from pipecat.services.xai.realtime.events import (
    SessionProperties,
    TurnDetection,
    AudioConfiguration,
    AudioInput,
    AudioOutput,
    PCMAudioFormat,
)

session_properties = SessionProperties(
    instructions="You are a helpful assistant.",
    voice="rex",
    turn_detection=TurnDetection(type="server_vad"),
    audio=AudioConfiguration(
        input=AudioInput(format=PCMAudioFormat(rate=16000)),
        output=AudioOutput(format=PCMAudioFormat(rate=16000)),
    ),
)

llm = GrokRealtimeLLMService(
    api_key=os.getenv("XAI_API_KEY"),
    settings=GrokRealtimeLLMService.Settings(
        session_properties=session_properties,
    ),
)

​

With Built-in Tools

from pipecat.services.xai.realtime.llm import GrokRealtimeLLMService
from pipecat.services.xai.realtime.events import (
    SessionProperties,
    WebSearchTool,
    XSearchTool,
)

llm = GrokRealtimeLLMService(
    api_key=os.getenv("XAI_API_KEY"),
    settings=GrokRealtimeLLMService.Settings(
        session_properties=SessionProperties(
            instructions="You are a helpful assistant with access to web search.",
            voice="eve",
            tools=[
                WebSearchTool(),
                XSearchTool(allowed_x_handles=["@elonmusk"]),
            ],
        ),
    ),
)

​

Updating Settings at Runtime

from pipecat.frames.frames import LLMUpdateSettingsFrame
from pipecat.services.xai.realtime.llm import GrokRealtimeLLMSettings

session_properties = session_properties.model_copy(update={"voice": "eve"})

await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=GrokRealtimeLLMSettings(
            system_instruction="Now speak in Spanish.",
            session_properties=session_properties,
        )
    )
)

​

Methods

​

delete_conversation_item

await service.delete_conversation_item(item_id: str)

• item_id (str): ID of the conversation item to delete

​

force_message

await service.force_message(text: str)

• text (str): Verbatim text to synthesize and play

​

set_audio_input_paused()

def set_audio_input_paused(paused: bool)

​

Notes

• Model versioning: The default model "grok-voice-latest" tracks xAI’s recommended Voice Agent. For stability, pin a specific version in settings (e.g., model="grok-voice-think-fast-1.0").
• Audio format auto-configuration: If audio.input or audio.output is not set in session_properties, the service configures PCM for it using the pipeline’s sample rates, which must be one of the supported PCM rates. When you set audio.input or audio.output (for example, for transcription or speed), set its format too. Input audio is sent without resampling, so a configured input rate must match the pipeline’s audio_in_sample_rate.
• Server-side VAD: Enabled by default. When VAD is enabled, the server handles speech detection and turn management automatically. Set turn_detection to None for manual turn detection.
• Audio flow timing: User audio flows once the session is configured, without waiting for a first context, so the user can speak first without an LLMRunFrame.
• Interruption handling: Interruptions send response.cancel to stop in-flight assistant audio on the wire and conversation.item.truncate to align server-side history with what the user heard. With server VAD enabled, the input buffer is preserved (not cleared) so interrupting user speech remains intact. Manual turn mode clears the input buffer on interruption.
• Voice IDs: Accepts any built-in voice ID from xAI’s catalogue or a custom voice ID from the Custom Voices API. Voice IDs are case-insensitive. Default is "eve".
• G.711 formats: PCMU and PCMA are fixed at 8000 Hz. The service passes audio through without converting it, and pipeline audio frames are 16-bit PCM, so use PCMAudioFormat; for telephony, the transport’s serializer converts to and from G.711.
• System instruction precedence: The system_instruction from service settings takes precedence over an initial system message in the LLM context. A warning is logged when both are set.
• Async tool support: Functions registered with cancel_on_interruption=False are supported. The final tool result is delivered via the formal tool-result channel once the async function completes. Intermediate results (is_final=False) reach the model as conversation messages.

​

Event Handlers

@llm.event_handler("on_conversation_item_created")
async def on_item_created(service, item_id, item):
    print(f"New conversation item: {item_id}")

@llm.event_handler("on_conversation_item_updated")
async def on_item_updated(service, item_id, item):
    print(f"Conversation item updated: {item_id}")

@llm.event_handler("on_conversation_item_deleted")
async def on_item_deleted(service, item_id):
    print(f"Conversation item deleted: {item_id}")

@llm.event_handler("on_idle_timeout")
async def on_idle_timeout(service, item_id):
    print(f"Idle timeout triggered")

@llm.event_handler("on_dtmf_received")
async def on_dtmf(service, button):
    print(f"DTMF digit received: {button}")

@llm.event_handler("on_mcp_event")
async def on_mcp(service, event_type, event):
    print(f"MCP event: {event_type}")