Mistral Search Index

Use Search Index, the search storage hosted by Mistral, to store and search the chunks of your Search Toolkit pipelines. MistralSearchIndex implements VectorStoreIndex, NavigableIndex, and RetrievableIndex, so it uses the same pipeline interfaces as the Vespa and Postgres backends. You don't provision or operate a database: the index lives in your Mistral workspace.

The client ships with the core package and doesn't need an optional extra:

uv add mistralai-search-toolkit

uv add mistralai-search-toolkit

Three objects make up the client, all in mistralai.search.toolkit.mistral_search_index:

• MistralSearchIndexCollectionSchema: declares the index name, the embedding model, and the custom fields derived from your Document subclass.
• MistralSearchIndexService: manages the index lifecycle (create_index, update_index, delete_index, get_index_info, index_exists) and resolves an existing index to a store with get_search_index.
• MistralSearchIndex: indexes, deletes, searches, retrieves, and navigates documents. Pass it to Pipeline(stores=...) for ingestion, and query it with search and retrieve.

The server is the source of truth: get_search_index reads the schema and embedding model from the index, so a process that only queries doesn't need the schema declaration. Don't construct MistralSearchIndex yourself. Get it from get_search_index instead, so that it matches the index on the server.

MistralSearchIndexService() builds a Mistral client from the MISTRAL_API_KEY environment variable and raises RuntimeError if it isn't set. To call an API endpoint other than https://api.mistral.ai, set MISTRAL_API_URL. To reuse a client that you already configured, pass it in:

from mistralai.client import Mistral
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService

service = MistralSearchIndexService(client=Mistral(api_key="..."))

from mistralai.client import Mistral
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService

service = MistralSearchIndexService(client=Mistral(api_key="..."))

To create an index, you need the workspace create_search_indexes permission. For the permission model, see Create an index.

Derive the schema from your document type with from_document_type. As with the other backends, embedding_model takes an EmbeddingModel or a MistralEmbeddingPreset, described in Embedding model:

from typing import Annotated, Sequence

from mistralai.search.toolkit.document import Document, DocumentChunk
from mistralai.search.toolkit.embedding import MistralEmbeddingPreset
from mistralai.search.toolkit.mistral_search_index.schema import (
    MistralSearchIndexCollectionSchema,
    MistralSearchIndexField,
    MistralSearchIndexFieldType,
)


class ReportChunk(DocumentChunk):
    page: Annotated[int | None, MistralSearchIndexField(indexed=True)] = None


class ReportDocument(Document):
    title: str
    category: Annotated[str, MistralSearchIndexField(type=MistralSearchIndexFieldType.KEYWORD, indexed=True)]
    chunks: Sequence[ReportChunk] = []


SCHEMA = MistralSearchIndexCollectionSchema.from_document_type(
    collection_name="reports",
    document_type=ReportDocument,
    embedding_model=MistralEmbeddingPreset.MISTRAL_EMBED_DIM_1024,
)

from typing import Annotated, Sequence

from mistralai.search.toolkit.document import Document, DocumentChunk
from mistralai.search.toolkit.embedding import MistralEmbeddingPreset
from mistralai.search.toolkit.mistral_search_index.schema import (
    MistralSearchIndexCollectionSchema,
    MistralSearchIndexField,
    MistralSearchIndexFieldType,
)


class ReportChunk(DocumentChunk):
    page: Annotated[int | None, MistralSearchIndexField(indexed=True)] = None


class ReportDocument(Document):
    title: str
    category: Annotated[str, MistralSearchIndexField(type=MistralSearchIndexFieldType.KEYWORD, indexed=True)]
    chunks: Sequence[ReportChunk] = []


SCHEMA = MistralSearchIndexCollectionSchema.from_document_type(
    collection_name="reports",
    document_type=ReportDocument,
    embedding_model=MistralEmbeddingPreset.MISTRAL_EMBED_DIM_1024,
)

Match the embedder to the index

The embedding model is fixed when you create the index. Every embedding that you send must come from the same model, with the same dimensions, both at ingestion and for queries that you embed yourself. If they don't match, the request fails. Pair the preset with the matching MistralEmbedder model:

The MODEL_*_EMBEDDING constants come from mistralai.search.toolkit.embedding. MistralEmbedder() without a model name uses the 128-dimension model, so pass model_name if the index uses another preset.

Custom fields

Every field that your Document or DocumentChunk subclass adds to the base model becomes a typed field on the index. Document fields describe a whole document, and chunk fields describe one passage. The toolkit handles the base fields (source_id, content, chunks, metadata, and the others).

The toolkit infers each field's type from its annotation:

A field is required unless its annotation allows None. MistralSearchIndexField(type=..., indexed=..., required=..., default=..., ignore=...) overrides any inferred value, and ignore=True keeps a field out of the index:

from datetime import datetime
from enum import StrEnum

from mistralai.search.toolkit.mistral_search_index.schema import MistralSearchIndexField, MistralSearchIndexFieldType


class Category(StrEnum):
    BILLING = "billing"
    SHIPPING = "shipping"


class Article(Document):
    title: str                                                   # text, full-text searchable
    category: Category                                           # keyword, exact match
    sku: Annotated[str, MistralSearchIndexField(type=MistralSearchIndexFieldType.KEYWORD)]
    published_at: Annotated[datetime | None, MistralSearchIndexField(indexed=True)] = None
    internal_note: Annotated[str | None, MistralSearchIndexField(ignore=True)] = None

from datetime import datetime
from enum import StrEnum

from mistralai.search.toolkit.mistral_search_index.schema import MistralSearchIndexField, MistralSearchIndexFieldType


class Category(StrEnum):
    BILLING = "billing"
    SHIPPING = "shipping"


class Article(Document):
    title: str                                                   # text, full-text searchable
    category: Category                                           # keyword, exact match
    sku: Annotated[str, MistralSearchIndexField(type=MistralSearchIndexFieldType.KEYWORD)]
    published_at: Annotated[datetime | None, MistralSearchIndexField(indexed=True)] = None
    internal_note: Annotated[str | None, MistralSearchIndexField(ignore=True)] = None

• Use keyword for identifiers such as SKUs, codes, or tags: a str defaults to text, which is tokenized for keyword search.
• Unsupported types fail early. A field whose type has no mapping raises ValueError when you build the schema. To fix it, add a MistralSearchIndexField(type=...) override or set ignore=True.
• Required fields are checked locally. Before each write, the schema checks that every required custom field is set, so a missing field fails before the request is sent.

For field naming rules and system fields, see Custom fields. To filter on custom fields, see Filters.

Create, update, and delete operations run asynchronously on the server. create_index returns while the index is still provisioning, so wait for it to be ready before you write to it:

import asyncio

from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService
from mistralai.search.toolkit.mistral_search_index.models import MistralSearchIndexStatus

service = MistralSearchIndexService()

if not await service.index_exists("reports"):
    info = await service.create_index(SCHEMA)
    while info.status == MistralSearchIndexStatus.PROVISIONING:
        await asyncio.sleep(1)
        info = await service.get_index_info("reports")
    if info.status != MistralSearchIndexStatus.READY:
        raise RuntimeError(f"provisioning failed: {info.status_message}")

store = await service.get_search_index("reports")  # -> MistralSearchIndex

import asyncio

from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService
from mistralai.search.toolkit.mistral_search_index.models import MistralSearchIndexStatus

service = MistralSearchIndexService()

if not await service.index_exists("reports"):
    info = await service.create_index(SCHEMA)
    while info.status == MistralSearchIndexStatus.PROVISIONING:
        await asyncio.sleep(1)
        info = await service.get_index_info("reports")
    if info.status != MistralSearchIndexStatus.READY:
        raise RuntimeError(f"provisioning failed: {info.status_message}")

store = await service.get_search_index("reports")  # -> MistralSearchIndex

update_index(schema) pushes an expanded schema. Updates are additive: you can add optional fields or turn on indexed. To change anything else, recreate the index. delete_index(name) removes the index and all its documents.

Use the store in an ingestion pipeline:

from mistralai.search.toolkit.embedding import MODEL_1024_EMBEDDING, MistralEmbedder
from mistralai.search.toolkit.ingestion.extractors import MistralOCRExtractor
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader
from mistralai.search.toolkit.ingestion.pipelines import Pipeline
from mistralai.search.toolkit.ingestion.text_splitters import MarkdownTextSplitter, MarkdownTextSplitterConfig

pipeline = Pipeline(
    loader=FilesystemFileLoader(),
    extractor=MistralOCRExtractor(),
    text_splitter=MarkdownTextSplitter(MarkdownTextSplitterConfig()),
    processors=[ReportEnricher()],  # your processor that sets title, category, and page
    embedder=MistralEmbedder(model_name=MODEL_1024_EMBEDDING),
    stores=store,
)
await pipeline.run(documents=["docs/report.pdf"])

from mistralai.search.toolkit.embedding import MODEL_1024_EMBEDDING, MistralEmbedder
from mistralai.search.toolkit.ingestion.extractors import MistralOCRExtractor
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader
from mistralai.search.toolkit.ingestion.pipelines import Pipeline
from mistralai.search.toolkit.ingestion.text_splitters import MarkdownTextSplitter, MarkdownTextSplitterConfig

pipeline = Pipeline(
    loader=FilesystemFileLoader(),
    extractor=MistralOCRExtractor(),
    text_splitter=MarkdownTextSplitter(MarkdownTextSplitterConfig()),
    processors=[ReportEnricher()],  # your processor that sets title, category, and page
    embedder=MistralEmbedder(model_name=MODEL_1024_EMBEDDING),
    stores=store,
)
await pipeline.run(documents=["docs/report.pdf"])

The index doesn't compute embeddings at ingestion, so every chunk must carry one. Keep an embedder in the pipeline that uses the model of the index (MISTRAL_EMBED_DIM_1024 in this example). If a Pipeline writes to a vector index without an embedder, it raises an error when you build it. The source_id of a document is its identity: indexing the same source_id again replaces the document and its chunks.

index_document raises DocumentRejectedError, a subclass of IndexingError, when the server rejects the document for a deterministic reason, such as a missing field or text that the index can't store. The message lists each failing field. The error isn't retryable: fix the data before you send the document again. A pipeline reports the document as failed and continues with the rest of the batch. delete_document(doc_id) takes the document ID, not the source ID, and raises DocumentNotFoundError if no document matches.

Vector and hybrid search

search(VectorSearchQuery) builds the retriever for you:

• With only an embedding, it runs a nearest-neighbor retriever.
• With both query and embedding, it fuses a keyword retriever and a nearest-neighbor retriever with reciprocal rank fusion.

from mistralai.search.toolkit.search import VectorSearchQuery

results = await store.search(
    VectorSearchQuery(query="quarterly revenue", embedding=vec, top_k=10)
)

from mistralai.search.toolkit.search import VectorSearchQuery

results = await store.search(
    VectorSearchQuery(query="quarterly revenue", embedding=vec, top_k=10)
)

The query also honors exclude_ids, exclude_source_ids, and approximate_options=ApproximateQueryOptions(max_candidates=...). Keyword matching runs on content only, and every result includes its custom fields.

Retrieve

For a full walkthrough, see Retrieval. To filter on your fields, match keywords on a custom field, or tune the fusion, call retrieve with a retriever from mistralai.search.toolkit.mistral_search_index.retriever:

from mistralai.search.toolkit.mistral_search_index.retriever import (
    In,
    KeywordRetriever,
    NearestNeighbourRetriever,
    RRFRetriever,
)

results = await store.retrieve(
    RRFRetriever(
        top_k=10,
        filter=In(field="category", values=["financial", "legal"]),
        retrievers=[
            KeywordRetriever(query="termination clauses", top_k=10),
            KeywordRetriever(query="termination clauses", field="title", top_k=10),
            NearestNeighbourRetriever(query="termination clauses", top_k=10),
        ],
        weights=[1.0, 2.0, 1.0],
    )
)

from mistralai.search.toolkit.mistral_search_index.retriever import (
    In,
    KeywordRetriever,
    NearestNeighbourRetriever,
    RRFRetriever,
)

results = await store.retrieve(
    RRFRetriever(
        top_k=10,
        filter=In(field="category", values=["financial", "legal"]),
        retrievers=[
            KeywordRetriever(query="termination clauses", top_k=10),
            KeywordRetriever(query="termination clauses", field="title", top_k=10),
            NearestNeighbourRetriever(query="termination clauses", top_k=10),
        ],
        weights=[1.0, 2.0, 1.0],
    )
)

Every retriever takes a filter built from Equal, In, Range, And, Or, and Not. The top-level top_k is capped at 100. For the limits and score semantics, see Query an index.

Navigation

MistralSearchIndex implements NavigableIndex: navigate, read, grep, and get_chunk walk a source document in reading order instead of by relevance. These methods are the primitives behind Agentic Search. For each method and when to use it, see Document navigation.

• No partial updates. MistralSearchIndex doesn't implement PatchableIndex or VersionedIndex. To change a document, index it again.
• Status checks. Document and query calls on an index that isn't ready fail. The service raises IndexNotFoundError when a lifecycle call targets an index that doesn't exist or that you can't access.

• Search Index: the hosted product, its REST API, and its permissions.
• Basic operations: custom fields, system fields, and the REST equivalents of each call on this page.
• Search index: how this backend compares with the others.
• Embedding model: the embedding_model argument to MistralSearchIndexCollectionSchema.
• Document model: the Document and DocumentChunk types that the schema is derived from.