Skip to content

Grok Client

The Grok client provides access to xAI's Grok models. It shares OpenAIClient's OpenAI-compatible chat base (no batch API) since Grok uses an OpenAI-compatible API; its own batch methods (get_batch_results, list_batches) have different signatures than OpenAI's.

Configuration

from padwan_ai.grok import GrokClient

client = GrokClient(
    api_key="...",  # or set GROK_API_KEY env var
    model="grok-3",  # default model
)

padwan_ai.grok.supports_audio(model) / supports_vision(model) check model capabilities.

Usage

Basic Chat

from padwan_ai.conversation import Message

async with GrokClient() as client:
    response, usage = await client.complete_chat(
        [Message(role="user", content="Hello!")]
    )
    print(response["content"])

Streaming

from padwan_ai.conversation import Message

async with GrokClient() as client:
    stream = client.stream_chat([Message(role="user", content="Tell me a story")])
    async for chunk in stream:
        print(chunk, end="")

With System Prompt

from padwan_ai import ConversationState

state = ConversationState(system="You are a helpful assistant.")
state.add_user_message("Hello!")

async with GrokClient() as client:
    response, usage = await client.complete_chat(state.messages)
    state.add_assistant_response(response)
    state.accumulate_usage(usage)

Batch API

Grok uses xAI's native batch API, which differs from OpenAI's file-upload approach. Requests are submitted directly via JSON and results are fetched per-batch.

Create a batch

from padwan_ai.grok import GrokClient, GrokBatchRequest

requests = [
    GrokBatchRequest(
        body={"messages": [{"role": "user", "content": "Summarize X"}]},
        custom_id="req-1",
    ),
    GrokBatchRequest(
        body={"messages": [{"role": "user", "content": "Summarize Y"}]},
        custom_id="req-2",
    ),
]

async with GrokClient() as client:
    job = await client.create_batch(requests, name="summaries")
    print(job.batch_id, job.num_requests)

Each request's body is forwarded in full (temperature, tools, etc.); model comes from the client or the create_batch(model=)/add_batch_requests(model=) argument, overriding any model in body.

Use client.add_batch_requests(batch_id, requests) to add more requests to an existing batch.

Check status and fetch results

async with GrokClient() as client:
    job = await client.get_batch("batch-abc")

    if job.succeeded:
        results, next_token = await client.get_batch_results(job.batch_id)
        for r in results:
            print(r.custom_id, r.content)
        # get_batch_results pages (default limit=100); pass next_token as
        # pagination_token= to fetch the rest

List and cancel batches

async with GrokClient() as client:
    jobs, next_token = await client.list_batches(limit=10)
    job = await client.cancel_batch("batch-abc")

Batch types reference

Type Fields
GrokBatchRequest Single request: body, custom_id
GrokBatchJob Job state: batch_id, name, num_requests, num_pending, num_success, num_error, num_cancelled, create_time, expire_time, is_terminal, succeeded
GrokBatchResult Parsed result: custom_id, content, input_tokens, output_tokens, total_tokens, finish_reason, error_message

Method Outputs

response, usage = await client.complete_chat(messages)

stream = client.stream_chat(messages)
async for chunk in stream:
    ...
usage = stream.usage

Realtime (Voice Agent)

RealtimeClient with a Grok voice model opens a speech-to-speech Voice Agent session. The wire protocol is OpenAI Realtime-compatible, so the connection is the same RealtimeConnection as OpenAI's; only the endpoint, model, and voices differ. Known voices include eve (default), ara, and leo; Grok transcribes natively, so no transcription model is configured.

```python from padwan_ai import RealtimeClient

async with RealtimeClient("grok-voice-latest", instructions="Answer briefly.") as conn: await conn.append_audio(pcm16_chunk) # mono PCM16 @ 24 kHz async for event in conn: if audio := conn.audio_delta_bytes(event): playback.write(audio)