Skip to main content
EN · FR Start for free
Menu
Lesson 16 of 16 · Collections, RAG & Files

The xAI Python SDK: Asynchronous Operations

Lesson 16 of 16 Updated November 2, 2026 Developer
On this page
  1. Automate Your RAG Workflows in Python
  2. Installation and Configuration
  3. Collection Operations
  4. Add a Document and Wait for Indexing
  5. Search from the SDK
  6. A Complete End-to-End Pipeline
  7. Key Takeaways
  8. Test Your Knowledge

Updated November 2, 2026. Based on xai-sdk 1.21.0 code and the official xAI example: await calls require the asynchronous client AsyncClient (the Client is synchronous), returned objects expose collection_id, collections, matches, and chunk_content, indexing status compares against collections_pb2.DOCUMENT_STATUS_PROCESSED, and upload_document uploads and attaches the file in a single call (the previous example uploaded it twice, once using a non-existent method). Sources: Collections via the API · Collections Search · xai-sdk-python.

Automate Your RAG Workflows in Python

Everything you’ve called via curl throughout this course exists in Python: document upload, collection management, search, generation. The xAI SDK offers two clients, one synchronous (Client) and one asynchronous (AsyncClient), with identical methods. The latter is the right choice for a RAG workflow, which spends most of its time waiting—for an upload to finish, a document to index, a search to return. Asynchronous code handles these waits in parallel rather than in sequence.

Installation and Configuration

pip install xai-sdk

The client requires two keys: the standard API key for search and generation, and the Management API key for collections.

import os
from xai_sdk import AsyncClient

client = AsyncClient(
    api_key=os.getenv("XAI_API_KEY"),
    management_api_key=os.getenv("XAI_MANAGEMENT_API_KEY"),
)

Each request defaults to a 27-minute timeout before expiring, which covers uploading a large PDF. The timeout parameter, in seconds, lets you tighten this limit for an interactive application.

Collection Operations

Three calls cover the full lifecycle. Creation returns the ID you’ll reuse everywhere else:

collection = await client.collections.create(name="document-base-2026")
collection_id = collection.collection_id
print(f"Collection created: {collection_id}")

Listing helps mainly to recover an ID you didn’t persist, or to check the state of an account shared across multiple projects:

response = await client.collections.list()
for col in response.collections:
    print(f"{col.collection_id} — {col.collection_name} ({col.documents_count} documents)")

Deletion is irreversible and removes the documents and their indexes:

await client.collections.delete(collection_id)

Add a Document and Wait for Indexing

A single call, upload_document, uploads the file via the Files API and attaches it to the collection:

from pathlib import Path


async def add_document(client, collection_id, file_path):
    path = Path(file_path)
    document = await client.collections.upload_document(
        collection_id=collection_id,
        name=path.name,
        data=path.read_bytes(),
    )
    return document.file_metadata.file_id

If the file already exists in the Files API, add_existing_document(collection_id, file_id) attaches it without uploading it again.

Adding a document doesn’t make it queryable immediately. Indexing takes variable time depending on file size, and a search launched too early simply returns fewer results than expected—a bug that’s particularly painful to diagnose, since it raises no error. Hence this waiting loop, which polls the status until DOCUMENT_STATUS_PROCESSED:

import asyncio

from xai_sdk.proto import collections_pb2


async def wait_for_indexing(client, file_id, collection_id):
    while True:
        doc = await client.collections.get_document(file_id, collection_id)
        if doc.status == collections_pb2.DOCUMENT_STATUS_PROCESSED:
            print("Document ready")
            return
        if doc.status == collections_pb2.DOCUMENT_STATUS_FAILED:
            raise RuntimeError(f"Indexing failed: {doc.error_message}")
        await asyncio.sleep(3)

Status is an enumeration value, not a string: comparing it to the text "DOCUMENT_STATUS_PROCESSED" would never be true, and the loop would run forever. For a simple case, upload_document(..., wait_for_indexing=True) handles this wait for you.

Search from the SDK

Search returns the same scored chunks as the HTTP endpoint, in the matches list:

results = await client.collections.search(
    query="Order validation procedure",
    collection_ids=[collection_id],
)

for m in results.matches:
    print(f"Score: {m.score:.2f}")
    print(f"Content: {m.chunk_content[:200]}")
    print("---")

A Complete End-to-End Pipeline

The previous blocks assemble into a script that creates the collection, uploads documents, waits for indexing, searches, then generates a response using the collections_search tool:

import asyncio
import os

from xai_sdk import AsyncClient
from xai_sdk.chat import user
from xai_sdk.tools import collections_search


async def rag_pipeline():
    client = AsyncClient(
        api_key=os.getenv("XAI_API_KEY"),
        management_api_key=os.getenv("XAI_MANAGEMENT_API_KEY"),
    )

    # 1. Create the collection
    collection = await client.collections.create(name="project-documentation")
    collection_id = collection.collection_id

    # 2. Add documents in parallel
    files = [
        "docs/architecture.pdf",
        "docs/api-reference.pdf",
        "docs/deployment-guide.md",
    ]
    file_ids = await asyncio.gather(*[
        add_document(client, collection_id, f) for f in files
    ])

    # 3. Wait for all documents to index
    await asyncio.gather(*[
        wait_for_indexing(client, fid, collection_id) for fid in file_ids
    ])

    # 4. Search
    results = await client.collections.search(
        query="How to deploy to production?",
        collection_ids=[collection_id],
    )
    for m in results.matches:
        print(f"[{m.score:.2f}] {m.chunk_content[:150]}")

    # 5. Generate a response based on the collection
    chat = client.chat.create(
        model="grok-4.7",
        tools=[collections_search(collection_ids=[collection_id])],
    )
    chat.append(user("Explain the deployment procedure and cite your sources."))
    response = await chat.sample()
    print(response.content)
    print(response.citations)


asyncio.run(rag_pipeline())

The two asyncio.gather() calls do all the work. The first launches the three uploads together instead of waiting for each to finish; the second monitors the three indexing jobs simultaneously: total time approaches that of the longest document, rather than the sum of all. And since the wait doesn’t block the event loop, the same code fits directly into a FastAPI or aiohttp application, which continues serving other requests while your documents index.

Key Takeaways

  • The xAI Python SDK exists in synchronous (Client) and asynchronous (AsyncClient) versions; await requires the latter
  • Both keys (API and Management) configure in the client
  • upload_document uploads and attaches in one call; monitor indexing via get_document or wait_for_indexing=True
  • asyncio.gather() lets you upload and wait for indexing in parallel
  • Generation leverages the collection with the collections_search tool, complete with response and citations

Test Your Knowledge

Files, collections, search: RAG managed by xAI.

1. What does the Files API allow you to do?

Answer: Upload and manage files (list, view, delete) in supported formats—the raw material for collections and queries with documents.

2. What is a collection?

Answer: A set of documents managed server-side by xAI, with metadata and permissions—the ready-to-use index you query without building your own vector pipeline.

3. Which search modes are available?

Answer: Keyword (exact terms), semantic (meaning), and hybrid (both)—the choice follows the nature of the questions, with hybrid covering most cases.

4. How does RAG integrate with generation?

Answer: Via the collection search tool: collections_search in the xAI SDK, file_search with collection IDs in the Responses API. The model queries your documents during generation and cites its sources—the RAG pattern without infrastructure to maintain.

5. What points require attention for production?

Answer: Confidentiality of uploaded data (permissions, lifecycle), relevance to optimize (metadata, chunking)—and the asynchronous SDK for high-volume processing.

Managed RAG: you bring the documents and questions, xAI handles the index—keep control of confidentiality and relevance.

← Previous lessonData confidentiality in collections
← Back to catalog