Home
Studio

Quickstart: create and query an index

In this quickstart you create a search index, ingest a few documents into it, and run a first filtered hybrid query — then the index is ready to use from an agent or a workflow. You need a Mistral API key, Python 3.12 or later, uv, and a workspace with search indexes enabled.

Get an API key

Get an API key

Generate an API key in the Mistral Console and export it — API keys are only accessible once:

export MISTRAL_API_KEY=your-api-key
Note

Creating an index needs the workspace create_search_indexes permission. Without it, creation returns 403. The creator becomes the index owner and can see, query, ingest, modify, and delete the index afterwards; a denial on an existing index is reported as 404, never 403.

Install the Search Toolkit, which is the client for search indexes:

uv init search-index-quickstart && cd search-index-quickstart
uv add "mistralai-search-toolkit>=0.1.0"

Create, ingest, and query

Create, ingest, and query

Build quickstart.py step by step: declare the schema as Document / DocumentChunk subclasses, create the index, ingest three articles, run two hybrid queries, and clean up when you are done.

Imports and sample data. MistralSearchIndexService manages the index lifecycle, and the retriever imports are for the hybrid query at the end. Index names are unique per workspace, so a shared workspace needs a unique name — the random suffix keeps this quickstart re-runnable. Each sample article carries a source id, a title, a product, and a text.

import asyncio
import secrets
from typing import Annotated

from mistralai.search.toolkit.document import Document, DocumentChunk, SourceId
from mistralai.search.toolkit.embedding import MistralEmbedder, MistralEmbeddingPreset
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndex, MistralSearchIndexService
from mistralai.search.toolkit.mistral_search_index.models import MistralSearchIndexStatus
from mistralai.search.toolkit.mistral_search_index.retriever import (
    Equal,
    KeywordRetriever,
    NearestNeighbourRetriever,
    RRFRetriever,
)
from mistralai.search.toolkit.mistral_search_index.schema import (
    MistralSearchIndexCollectionSchema,
    MistralSearchIndexField,
    MistralSearchIndexFieldType,
)
from mistralai.search.toolkit.search import SearchResult

# Index names are unique per workspace, so a shared workspace needs a unique name.
INDEX = f"support_articles_{secrets.token_hex(4)}"
EMBEDDING = MistralEmbeddingPreset.MISTRAL_EMBED_DIM_1024

ARTICLES = [
    # (source_id, title, product, text)
    ("s3://docs/api-keys.md", "Managing API keys", "console",
     "Rotate an API key from the console under Workspace > API keys."),
    ("s3://docs/billing.md", "Invoices and billing", "billing",
     "Download monthly invoices and update your card on the billing page."),
    ("s3://docs/members.md", "Inviting members", "console",
     "Invite teammates to your workspace from the Members tab and assign roles."),
]

Declare the schema. Every field added on top of the base models becomes an index field; the wire type is inferred from the Python annotation, and MistralSearchIndexField overrides it. page_number is an indexed INT, and product is a KEYWORD, so queries can filter on it exactly.

# Every field added on top of the base models becomes an index field. The wire
# type is inferred from the annotation; MistralSearchIndexField overrides it.
class ArticleChunk(DocumentChunk):
    page_number: Annotated[int, MistralSearchIndexField(type=MistralSearchIndexFieldType.INT, indexed=True)]


class Article(Document):
    title: str
    product: Annotated[str, MistralSearchIndexField(type=MistralSearchIndexFieldType.KEYWORD)]
    chunks: list[ArticleChunk]

Create the index. from_document_type derives the collection schema from the Article model. Creation reserves the name and provisions the backing table asynchronously, so poll until the index is READY. The embedding model is immutable — it decides the physical column type and the vector operator class, so it cannot be changed after creation.

async def create_index(service: MistralSearchIndexService) -> None:
    schema = MistralSearchIndexCollectionSchema.from_document_type(
        collection_name=INDEX,
        document_type=Article,
        embedding_model=EMBEDDING,
    )
    info = await service.create_index(schema)
    while info.status == MistralSearchIndexStatus.PROVISIONING:
        await asyncio.sleep(1)
        info = await service.get_index_info(INDEX)
    if info.status != MistralSearchIndexStatus.READY:
        raise RuntimeError(f"provisioning failed: {info.status_message}")

Ingest the articles. Embed the texts in one call, then write each article as one chunk that carries its embedding — ingestion never embeds. Re-ingesting the same source_id replaces that document: its existing chunks are deleted and the new ones inserted, in one transaction.

async def ingest(index: MistralSearchIndex, embedder: MistralEmbedder) -> None:
    texts = [text for *_, text in ARTICLES]
    embeddings = (await embedder.embed(texts)).embeddings
    for (source_id, title, product, text), embedding in zip(ARTICLES, embeddings, strict=True):
        await index.index_document(
            Article(
                source_id=SourceId(source_id),
                content=text,
                title=title,
                product=product,
                chunks=[
                    ArticleChunk(
                        source_id=SourceId(source_id),
                        locator=f"char:0-{len(text)}",
                        start_offset=0,
                        end_offset=len(text),
                        content=text,
                        embedding=embedding,
                        page_number=1,
                    )
                ],
            )
        )

Run a hybrid query. The keyword and vector rankings are fused with reciprocal rank fusion. For a Mistral embedding model the server embeds the query itself, and the Equal filter on product is pushed down to both arms.

async def search(index: MistralSearchIndex, query: str, product: str | None = None) -> list[SearchResult]:
    return await index.retrieve(
        RRFRetriever(
            top_k=5,
            filter=Equal(field="product", value=product) if product else None,
            retrievers=[
                KeywordRetriever(query=query, top_k=5),
                NearestNeighbourRetriever(query=query, top_k=5),
            ],
        )
    )

Tie it together. Create the index, get a handle, ingest with an embedder on the chosen preset, print the hits of an unfiltered query and of a query filtered to console articles, and delete the index in finally — deleting is asynchronous too: the index goes to deleting, then disappears.

async def main() -> None:
    service = MistralSearchIndexService()
    await create_index(service)
    try:
        index = await service.get_search_index(INDEX)
        await ingest(index, MistralEmbedder(model_name=EMBEDDING.full_model_name))

        for hit in await search(index, "how do I rotate an api key"):
            print(f"{hit.score:.4f}  {hit.chunk.source_id}")
        for hit in await search(index, "add a colleague", product="console"):
            print(f"{hit.score:.4f}  {hit.chunk.source_id}")
    finally:
        # Deleting is asynchronous too: the index goes to deleting, then disappears.
        await service.delete_index(INDEX)


if __name__ == "__main__":
    asyncio.run(main())

Run it:

uv run quickstart.py

You should see

You should see

The index provisions and reaches READY status, then the script prints the search hits as score source_id lines. The first query returns articles matching the question across all products; the second query filters on product="console", so only console articles can appear.

Going further

Going further

Search index
Search Toolkit
Libraries
Basic operations