Skip to content

Chat SDK

The Chat SDK allows you to have conversations with Grok in a stateless fashion. Stateless in this context means that conversations are not stored on the server and every request to the API contains the entire conversation history. Use this SDK to implement custom applications that are based on Grok.

Getting started

To get started with the Chat SDK, create a new client and access its chat property:

import xai_sdk

client = xai_sdk.Client()
chat = client.chat

Turn-based conversations

To start a new conversation with Grok via the Chat SDK, call the create_conversation() method followed by the Conversation.add_response() method.

simple_chat.py
"""A simple example demonstrating text completion."""

import asyncio
import sys

import xai_sdk


async def main():
    """Runs the example."""
    client = xai_sdk.Client()

    conversation = client.chat.create_conversation()

    print("Enter an empty message to quit.\n")

    while True:
        user_input = input("Human: ")
        print("")

        if not user_input:
            return

        token_stream, _ = conversation.add_response(user_input)
        print("Grok: ", end="")
        async for token in token_stream:
            print(token, end="")
            sys.stdout.flush()
        print("\n")


asyncio.run(main())

About streaming

The add_response function returns a tuple of the form token_stream, final_response where token_stream is an async iterable and final_response is a future that evaluates to the final response once it has fully been generated. Using token_stream is entirely optional, and it's only used when you want to visualize the token-generation process in real time. Note that final_response won't resolve unless token_stream is used. If you are only interested in the final response, you can use the auxiliary method add_response_no_stream instead.

simple_chat_no_stream.py
"""A simple example demonstrating text completion without using streams."""

import asyncio

import xai_sdk


async def main():
    """Runs the example."""
    client = xai_sdk.Client()

    conversation = client.chat.create_conversation()

    print("Enter an empty message to quit.\n")

    while True:
        user_input = input("Human: ")
        print("")

        if not user_input:
            return

        response = await conversation.add_response_no_stream(user_input)
        print(f"Grok: {response.message}\n")


asyncio.run(main())

API reference

xai_sdk.chat.AsyncChat

Provides a simple chat API that can be used for products.

Source code in xai_sdk/chat.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class AsyncChat:
    """Provides a simple chat API that can be used for products."""

    def __init__(self, stub: stateless_chat_pb2_grpc.StatelessChatStub) -> None:
        """Initializes a new instance of the `AsyncChat` class.

        Args:
            stub: Stub used to communicate with the gRPC API.
        """
        self._stub = stub

    def create_conversation(self, fun_mode: bool = False) -> "Conversation":
        """Creates a new empty conversation.

        Args:
            fun_mode: Whether fun mode shall be enabled for this conversation.

        Returns:
            Newly created conversation.
        """
        return Conversation(self._stub, fun_mode)

xai_sdk.chat.AsyncChat.create_conversation(fun_mode=False)

Creates a new empty conversation.

Parameters:

Name Type Description Default
fun_mode bool

Whether fun mode shall be enabled for this conversation.

 False

Returns:

Type Description
Conversation

Newly created conversation.
Source code in xai_sdk/chat.py 
26
27
28
29
30
31
32
33
34
35
def create_conversation(self, fun_mode: bool = False) -> "Conversation":
    """Creates a new empty conversation.

    Args:
        fun_mode: Whether fun mode shall be enabled for this conversation.

    Returns:
        Newly created conversation.
    """
    return Conversation(self._stub, fun_mode)

xai_sdk.chat.Conversation

A conversation held via the stateless Chat API.

Source code in xai_sdk/chat.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class Conversation:
    """A conversation held via the stateless Chat API."""

    def __init__(self, stub: stateless_chat_pb2_grpc.StatelessChatStub, fun_mode: bool):
        """Initializes a new instance of the `Conversation` class.

        Args:
            stub: Stub used to communicate with the gRPC API.
        """
        self._stub = stub
        self._conversation_id = uuid.uuid4().hex

        self._conversation = stateless_chat_pb2.StatelessConversation(
            stateless_conversation_id=self._conversation_id,
            responses=[],
            system_prompt_name="fun" if fun_mode else "",
        )

    @property
    def history(self) -> Sequence[stateless_chat_pb2.StatelessResponse]:
        """Returns the linear conversation history."""
        return self._conversation.responses

    @property
    def fun_mode(self) -> bool:
        """Returns true if the conversation happens in fun mode."""
        return self._conversation.system_prompt_name == "fun"

    async def add_response_no_stream(self, user_message: str) -> stateless_chat_pb2.StatelessResponse:
        """Same as `add_response` but doesn't return a token stream.

        Use this function if you are only interested in the complete response and don't need to
        stream the individual tokens.

        Args:
            user_message: Message the user has entered.

        Returns:
            The newly generated response.
        """
        stream, response = self.add_response(user_message)

        # We have to iterate over the stream to generate the final response.
        async for _ in stream:
            pass

        response = await response
        return response

    def add_response(
        self, user_message: str
    ) -> tuple[AsyncGenerator[str, None], asyncio.Future[stateless_chat_pb2.StatelessResponse]]:
        """Adds a new user response to the conversation and samples a model response in return.

        Args:
            user_message: Message the user has entered.

        Returns:
            A tuple of the form `token_stream, response` where `token_stream` is an async iterable
                that emits the individual string tokens of the newly sampled response and `response`
                is a future that resolves to the Response object created.
        """
        self._conversation.responses.append(
            stateless_chat_pb2.StatelessResponse(
                sender=stateless_chat_pb2.StatelessResponse.Sender.HUMAN,
                message=user_message,
            )
        )

        response_future: asyncio.Future[stateless_chat_pb2.StatelessResponse] = asyncio.Future()

        async def _unroll_tokens():
            """Unrolls the token stream."""
            try:
                response = stateless_chat_pb2.StatelessResponse(
                    sender=stateless_chat_pb2.StatelessResponse.Sender.ASSISTANT,
                    message="",
                    query="",
                )

                async for update in self._stub.AddResponse(self._conversation):
                    if update.message:
                        response.message += update.message
                        yield update.message

                    if update.query:
                        response.query += update.query

                self._conversation.responses.append(response)
                response_future.set_result(response)
            except Exception as e:
                response_future.set_exception(e)

        return _unroll_tokens(), response_future

xai_sdk.chat.Conversation.fun_mode: bool property

Returns true if the conversation happens in fun mode.

xai_sdk.chat.Conversation.history: Sequence[stateless_chat_pb2.StatelessResponse] property

Returns the linear conversation history.

xai_sdk.chat.Conversation.add_response(user_message)

Adds a new user response to the conversation and samples a model response in return.

Parameters:

Name Type Description Default
user_message str

Message the user has entered.

 required

Returns:

Type Description
tuple[AsyncGenerator[str, None], Future[StatelessResponse]]

A tuple of the form token_stream, response where token_stream is an async iterable that emits the individual string tokens of the newly sampled response and response is a future that resolves to the Response object created.
Source code in xai_sdk/chat.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def add_response(
    self, user_message: str
) -> tuple[AsyncGenerator[str, None], asyncio.Future[stateless_chat_pb2.StatelessResponse]]:
    """Adds a new user response to the conversation and samples a model response in return.

    Args:
        user_message: Message the user has entered.

    Returns:
        A tuple of the form `token_stream, response` where `token_stream` is an async iterable
            that emits the individual string tokens of the newly sampled response and `response`
            is a future that resolves to the Response object created.
    """
    self._conversation.responses.append(
        stateless_chat_pb2.StatelessResponse(
            sender=stateless_chat_pb2.StatelessResponse.Sender.HUMAN,
            message=user_message,
        )
    )

    response_future: asyncio.Future[stateless_chat_pb2.StatelessResponse] = asyncio.Future()

    async def _unroll_tokens():
        """Unrolls the token stream."""
        try:
            response = stateless_chat_pb2.StatelessResponse(
                sender=stateless_chat_pb2.StatelessResponse.Sender.ASSISTANT,
                message="",
                query="",
            )

            async for update in self._stub.AddResponse(self._conversation):
                if update.message:
                    response.message += update.message
                    yield update.message

                if update.query:
                    response.query += update.query

            self._conversation.responses.append(response)
            response_future.set_result(response)
        except Exception as e:
            response_future.set_exception(e)

    return _unroll_tokens(), response_future

xai_sdk.chat.Conversation.add_response_no_stream(user_message) async

Same as add_response but doesn't return a token stream.

Use this function if you are only interested in the complete response and don't need to stream the individual tokens.

Parameters:

Name Type Description Default
user_message str

Message the user has entered.

 required

Returns:

Type Description
StatelessResponse

The newly generated response.
Source code in xai_sdk/chat.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def add_response_no_stream(self, user_message: str) -> stateless_chat_pb2.StatelessResponse:
    """Same as `add_response` but doesn't return a token stream.

    Use this function if you are only interested in the complete response and don't need to
    stream the individual tokens.

    Args:
        user_message: Message the user has entered.

    Returns:
        The newly generated response.
    """
    stream, response = self.add_response(user_message)

    # We have to iterate over the stream to generate the final response.
    async for _ in stream:
        pass

    response = await response
    return response