File loaders

File loaders read files from a source and return File objects that document extractors can process.

Available file loaders

Loader	Source
Filesystem File Loader	Local filesystem
AWS S3 File Loader	AWS S3 (and S3-compatible: MinIO, Ceph)
Google Cloud Storage Loader	Google Cloud Storage
Azure Blob Storage Loader	Azure Blob Storage
Custom Loaders	Any source

Filesystem file loader

FilesystemFileLoader loads files from the local filesystem. It accepts a root parameter that restricts access. Any path resolving outside the root is rejected, preventing path traversal.

Installation: Core library (no extra required)

Example:

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(root="/data/documents")
file = await loader.load_file(Path("report.pdf"))

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(root="/data/documents")
file = await loader.load_file(Path("report.pdf"))

Limiting file size:

Use max_file_size to reject files that exceed a given size in bytes. The check is performed via a metadata lookup before loading, so oversized files are never read into memory:

from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(
    root="/data/documents",
    max_file_size=50 * 1024 * 1024,  # 50 MiB
)
file = await loader.load_file("report.pdf")

from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(
    root="/data/documents",
    max_file_size=50 * 1024 * 1024,  # 50 MiB
)
file = await loader.load_file("report.pdf")

A FileSizeLimitExceededException is raised when a file exceeds the limit.

Parameters:

Parameter	Type	Default	Description
`root`	`Path \| str`	`"/"`	Root directory. Paths resolving outside this directory are rejected.
`max_file_size`	`int \| None`	`None`	Maximum file size in bytes. Files exceeding this limit are rejected before loading. `None` means no limit.

Tip

Set root to the narrowest directory that holds your documents. The loader rejects any path that resolves outside root before it opens the file, which helps protect against path traversal.

AWS S3 file loader

Load files from AWS S3 buckets (or S3-compatible services like MinIO, Ceph). S3FileLoader is a convenience wrapper around FileLoader backed by S3BlobStorage.

Installation:

uv add "mistralai-search-toolkit-storage-s3"

uv add "mistralai-search-toolkit-storage-s3"

Example:

from mistralai.search.toolkit.plugins.storage.s3 import S3FileLoader

loader = S3FileLoader(
    bucket_name="my-bucket",
    region_name="us-east-1",
    endpoint_url="http://localhost:9000",  # optional, for MinIO / other S3-compatible backends
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.plugins.storage.s3 import S3FileLoader

loader = S3FileLoader(
    bucket_name="my-bucket",
    region_name="us-east-1",
    endpoint_url="http://localhost:9000",  # optional, for MinIO / other S3-compatible backends
)

file = await loader.load_file("reports/example.pdf")

Parameters (S3FileLoader):

Parameter	Type	Default	Description
`bucket_name`	`str`	(required)	S3 bucket name
`region_name`	`str \| None`	`None`	AWS region (falls back to the SDK/env default)
`endpoint_url`	`str \| None`	`None`	Custom endpoint URL (for MinIO, Ceph, or other S3-compatible services)
`aws_access_key_id`	`str \| None`	`None`	Explicit AWS access key (falls back to the credential chain)
`aws_secret_access_key`	`str \| None`	`None`	Explicit AWS secret key
`aws_session_token`	`str \| None`	`None`	Explicit AWS session token
`prefix`	`str`	`""`	Key prefix scoped to this loader
`max_file_size`	`int \| None`	`None`	Reject files larger than this many bytes

Authentication:

Uses the default AWS credentials chain:

IAM role (recommended for production)
Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
AWS credentials file: ~/.aws/credentials

Azure Blob Storage loader

Load files from Azure Blob Storage. AzureBlobFileLoader is a convenience wrapper around FileLoader backed by AzureBlobStorage.

Installation:

uv add "mistralai-search-toolkit-storage-azure"

uv add "mistralai-search-toolkit-storage-azure"

Example:

from mistralai.search.toolkit.plugins.storage.azure import AzureBlobFileLoader

loader = AzureBlobFileLoader(
    container_name="my-container",
    account_url="https://myaccount.blob.core.windows.net",
    use_workload_identity=True,  # or pass azure_connection_string=... for connection-string auth
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.plugins.storage.azure import AzureBlobFileLoader

loader = AzureBlobFileLoader(
    container_name="my-container",
    account_url="https://myaccount.blob.core.windows.net",
    use_workload_identity=True,  # or pass azure_connection_string=... for connection-string auth
)

file = await loader.load_file("reports/example.pdf")

Parameters (AzureBlobFileLoader):

Parameter	Type	Default	Description
`container_name`	`str`	(required)	Azure Blob Storage container name
`azure_connection_string`	`str \| None`	`None`	Connection string for authentication (required unless `use_workload_identity` is `True`)
`account_url`	`str \| None`	`None`	Azure storage account URL, for example `https://myaccount.blob.core.windows.net` (required when `use_workload_identity` is `True`)
`use_workload_identity`	`bool`	`False`	Use workload identity for authentication (recommended for Azure VMs/Functions)
`max_file_size`	`int \| None`	`None`	Reject files larger than this many bytes
`prefix`	`str`	`""`	Blob name prefix scoped to this loader

Authentication:

Workload Identity (recommended for Azure VMs/Functions)
Connection string: Pass it with the azure_connection_string parameter (required unless use_workload_identity=True)
SAS token: Time-limited access (configure via account credentials)

Google Cloud Storage loader

Load files from Google Cloud Storage (GCS). GCSFileLoader is a convenience wrapper around FileLoader backed by GCSBlobStorage.

Installation:

uv add "mistralai-search-toolkit-storage-gcs"

uv add "mistralai-search-toolkit-storage-gcs"

Example:

from mistralai.search.toolkit.plugins.storage.gcs import GCSFileLoader

loader = GCSFileLoader(
    bucket_name="my-bucket",
    service_account_file="/path/to/service-account.json",  # optional; falls back to ADC
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.plugins.storage.gcs import GCSFileLoader

loader = GCSFileLoader(
    bucket_name="my-bucket",
    service_account_file="/path/to/service-account.json",  # optional; falls back to ADC
)

file = await loader.load_file("reports/example.pdf")

Parameters (GCSFileLoader):

Parameter	Type	Default	Description
`bucket_name`	`str`	(required)	GCS bucket name
`service_account_file`	`str \| None`	`None`	Path to service account JSON file. If not provided, uses Application Default Credentials (ADC)
`api_root`	`str \| None`	`None`	Custom API endpoint for a local emulator (for example, fake-gcs-server); falls back to `STORAGE_EMULATOR_HOST`
`max_file_size`	`int \| None`	`None`	Reject files larger than this many bytes
`prefix`	`str`	`""`	Object name prefix scoped to this loader

Authentication:

Application Default Credentials (ADC) (recommended): uses credentials from environment, metadata service, or gcloud CLI
Service account file: Pass via service_account_file parameter or set GOOGLE_APPLICATION_CREDENTIALS environment variable

Batch loading

Load multiple files with concurrency control:

import asyncio
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.s3 import S3FileLoader

async def load_files_batch(
    loader: FileLoader,
    paths: list[str],
    max_concurrent: int = 10,
) -> list:
    """Load multiple files concurrently with semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def load_with_semaphore(path: str):
        async with semaphore:
            try:
                return await loader.load_file(path)
            except Exception as e:
                print(f"Failed to load {path}: {e}")
                return None

    results = await asyncio.gather(
        *[load_with_semaphore(p) for p in paths],
        return_exceptions=False,
    )
    return [f for f in results if f is not None]

# Usage
loader = S3FileLoader(bucket_name="my-bucket")
files = await load_files_batch(
    loader,
    ["doc1.pdf", "doc2.pdf", "doc3.pdf"],
    max_concurrent=5,
)

import asyncio
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.s3 import S3FileLoader

async def load_files_batch(
    loader: FileLoader,
    paths: list[str],
    max_concurrent: int = 10,
) -> list:
    """Load multiple files concurrently with semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def load_with_semaphore(path: str):
        async with semaphore:
            try:
                return await loader.load_file(path)
            except Exception as e:
                print(f"Failed to load {path}: {e}")
                return None

    results = await asyncio.gather(
        *[load_with_semaphore(p) for p in paths],
        return_exceptions=False,
    )
    return [f for f in results if f is not None]

# Usage
loader = S3FileLoader(bucket_name="my-bucket")
files = await load_files_batch(
    loader,
    ["doc1.pdf", "doc2.pdf", "doc3.pdf"],
    max_concurrent=5,
)

Information

Advanced: each cloud loader wraps a generic FileLoader over an ObjectStorage backend. For full control (custom storage settings, a backend not covered by a wrapper), construct it directly: FileLoader(lambda: S3BlobStorage(...)).

Custom loaders

For a source not covered by the built-in loaders, subclass FileLoader and override load_file. Define your own __init__, because the base constructor expects a storage factory. To reuse the size limit and storage plumbing instead, implement ObjectStorage and wrap it: FileLoader(lambda: MyStorage()).

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.ingestion import File

class CustomFileLoader(FileLoader):
    """Load files from a custom source."""

    def __init__(self) -> None:
        pass

    async def load_file(self, path: Path | str) -> File:
        # Implement your custom loading logic
        content = await self._fetch_from_custom_source(str(path))
        filename = Path(str(path)).name

        return File(
            path=str(path),
            name=filename,
            raw=content,
        )

    async def _fetch_from_custom_source(self, path: str) -> bytes:
        # Your implementation here
        ...

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.ingestion import File

class CustomFileLoader(FileLoader):
    """Load files from a custom source."""

    def __init__(self) -> None:
        pass

    async def load_file(self, path: Path | str) -> File:
        # Implement your custom loading logic
        content = await self._fetch_from_custom_source(str(path))
        filename = Path(str(path)).name

        return File(
            path=str(path),
            name=filename,
            raw=content,
        )

    async def _fetch_from_custom_source(self, path: str) -> bytes:
        # Your implementation here
        ...