Create a Batch Job, Track Status, and Download Results

Updated November 3, 2026: Addendum for November 3: files.download returns an HTTP response (httpx2.Response); read its content with .read() before calling .decode(). Upload expects file={"file_name": …, "content": …}. The API key is no longer hardcoded.

From JSONL File to Results: The Complete Lifecycle

Your JSONL file is prepared and uploaded, and you have a file_id. The next step is transforming that dormant file into usable results: create the batch job, monitor its progress over the minutes or hours the processing takes, then retrieve and reconcile the responses. This lesson walks through the end-to-end Python code in the exact order you will write it.

Create a Batch Job

Creation requires a single call. You specify the source file, the model, and the target endpoint:

from mistralai.client import Mistral
import time

client = Mistral(api_key="your-api-key")

# Create the batch job
job = client.batch.jobs.create(
    input_files=[batch_file.id],   # ID of the uploaded file
    model="mistral-small-latest",
    endpoint="/v1/chat/completions",
    metadata={"project": "document-analysis", "version": "1.0"}
)

print(f"Job created: {job.id}")
print(f"Initial status: {job.status}")

Each parameter deserves a note:

• input_files — A list of uploaded file IDs. You can pass multiple files to a single job.
• model — The Mistral model to use. Only one model per batch job.
• endpoint — The target API endpoint (must match the body format in your JSONL).
• metadata — An optional dictionary of custom metadata to organize and retrieve your jobs.

The single-model-per-job constraint structures how you split your work: if you want to compare two models on the same corpus, submit two distinct jobs pointing to the same file. As for metadata, it may seem incidental until you have thirty jobs in your history; the project/version pair shown above then lets you find the batch that produced questionable data in three seconds flat.

Batch Job Statuses

A batch job moves through several states during its lifecycle:

Four of these seven states are terminal—SUCCESS, FAILED, TIMEOUT_EXCEEDED, and CANCELLED—and this distinction drives the monitoring loop: as long as the status is not in this set, the job is still alive and you must wait. CANCELLATION_REQUESTED, in particular, is not a stopped state but a transitional one; code that treats it as terminal would exit the loop too early.

Track Progress

The polling pattern below encapsulates this logic exactly, with a maximum duration guardrail to prevent a forgotten script from running indefinitely:

def wait_for_batch(client, job_id, poll_interval=10, max_wait=3600):
    """Wait for a batch job to complete using polling."""
    elapsed = 0
    terminal_states = {"SUCCESS", "FAILED", "TIMEOUT_EXCEEDED", "CANCELLED"}

    while elapsed < max_wait:
        job = client.batch.jobs.get(job_id=job_id)
        print(f"[{elapsed}s] Status: {job.status}")

        if job.status in terminal_states:
            return job

        time.sleep(poll_interval)
        elapsed += poll_interval

    raise TimeoutError(f"Job {job_id} still running after {max_wait}s")

# Usage
completed_job = wait_for_batch(client, job.id)
print(f"Final result: {completed_job.status}")

A ten-second interval works for a batch lasting a few minutes; for an overnight run processing several hundred thousand lines, increase it to a minute or more. Greater polling precision adds no value there and only accumulates unnecessary API calls.

Download Results

Once the status reaches SUCCESS, the job exposes an output_file that you download and parse line by line:

import json

if completed_job.status == "SUCCESS":
    # Get the output file ID
    output_file_id = completed_job.output_file

    # Download the content
    result_data = client.files.download(file_id=output_file_id)

    # Parse JSONL results
    results = {}
    for line in result_data.read().decode("utf-8").strip().split("\n"):
        entry = json.loads(line)
        custom_id = entry["custom_id"]
        response = entry["response"]
        results[custom_id] = response

    # Show an example
    for cid, resp in list(results.items())[:3]:
        content = resp["body"]["choices"][0]["message"]["content"]
        print(f"\n--- {cid} ---")
        print(content[:200])

The results file is also in JSONL format, and each line carries the original custom_id alongside the model’s full response. This is where the naming effort from the previous lesson pays off: indexing by custom_id in a dictionary is enough to reconcile responses with your source documents, regardless of the order in which they return. Displaying the first three entries truncated to 200 characters is a useful sanity check before you start writing to your database.

List and Manage Your Jobs

You can filter the job inventory by status, letting you see at any moment what is actually running on your account:

# List all currently running jobs
running_jobs = client.batch.jobs.list(status="RUNNING")
for j in running_jobs.data:
    print(f"Job {j.id} - Model: {j.model} - Created: {j.created_at}")

And when a job drags on or you spot a prompt error after submission, cancellation avoids paying for useless processing:

# Cancel a job that is taking too long
client.batch.jobs.cancel(job_id="job-abc123")

Cancellation is asynchronous: the status first moves to CANCELLATION_REQUESTED, then to CANCELLED once the stop is effective. Do not consider the resource freed immediately upon the call’s return.

End-to-End Workflow

Assembled, these four steps form a short script you can place directly in a scheduler:

from mistralai.client import Mistral
import json
import os
import time

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# 1. Upload the file
with open("batch_requests.jsonl", "rb") as f:
    batch_file = client.files.upload(
        file={"file_name": "batch_requests.jsonl", "content": f},
        purpose="batch"
    )

# 2. Create the job
job = client.batch.jobs.create(
    input_files=[batch_file.id],
    model="mistral-small-latest",
    endpoint="/v1/chat/completions"
)

# 3. Wait for completion
completed = wait_for_batch(client, job.id)

# 4. Download and parse results
if completed.status == "SUCCESS":
    data = client.files.download(file_id=completed.output_file)
    for line in data.read().decode("utf-8").strip().split("\n"):
        entry = json.loads(line)
        print(f"{entry['custom_id']}: OK")
else:
    print(f"Batch failed: {completed.status}")

Reuse this skeleton on your own corpus with just ten lines: you will validate the entire chain in a few minutes before committing to the full volume.

Key Takeaways

• Create a batch job with client.batch.jobs.create() by specifying the file, model, and endpoint
• Seven possible statuses, including three terminal ones: SUCCESS, FAILED, CANCELLED
• Polling with an interval is the standard pattern for tracking progress
• Results are in JSONL format, linked to requests via custom_id
• You can list, inspect, and cancel your jobs at any time