Complete Mistral AI API Tutorial: Building AI Applications with Mistral's LLM Models

Mistral AI is one of the leading large language model (LLM) providers from France, offering high-performance models with exceptional efficiency. With an easy-to-use API, Mistral provides various models ranging from lightweight options like Mistral Small to powerful ones like Mistral Large and Codestral for coding tasks.

In this tutorial, we will learn how to use the Mistral AI API comprehensively with Python, from initial setup, chat completion, function calling, embeddings, to advanced usage including streaming, JSON mode, and agents.

Why Mistral AI?

Mistral AI has several advantages that make it worth considering:

• High efficiency: Mistral models are designed to deliver optimal performance with smaller parameter sizes compared to competitors
• Open-weight models: Some models like Mistral 7B and Mixtral are available as open-weight
• Competitive pricing: Lower cost per token compared to many other providers
• Fast inference: Quick response times thanks to efficient model architecture
• Multilingual support: Good performance across various languages
• Function calling: Native support for function calling and tool use

Installation and Setup

Installing the Python SDK

pip install mistralai

Getting an API Key

Configuring Environment Variables

export MISTRALAPIKEY="your-api-key-here"

Or save it in a .env file:

MISTRALAPIKEY=your-api-key-here

Verifying Installation

import os
from mistralai import Mistral

client = Mistral(apikey=os.environ["MISTRALAPIKEY"])


Test connection by listing models
models = client.models.list()
for model in models.data:
    print(f"- {model.id}")

import os
from mistralai import Mistral

client = Mistral(apikey=os.environ["MISTRALAPIKEY"])

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {"role": "user", "content": "Explain what machine learning is in 3 sentences."}
    ]
)

print(response.choices[0].message.content)

Using System Messages

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {
            "role": "system",
            "content": "You are a data science expert who explains technical concepts using simple language and everyday examples."
        },
        {
            "role": "user",
            "content": "What is the difference between supervised and unsupervised learning?"
        }
    ]
)

print(response.choices[0].message.content)

Multi-turn Conversation

messages = [
    {"role": "system", "content": "You are a patient and helpful Python tutor."},
    {"role": "user", "content": "How do I create a list comprehension in Python?"},
]

response = client.chat.complete(
    model="mistral-large-latest",
    messages=messages
)

assistantreply = response.choices[0].message.content

print("Assistant:", assistantreply)

messages.append({"role": "assistant", "content": assistantreply})

messages.append({"role": "user", "content": "Give me an example with a filter condition."})

response = client.chat.complete(
    model="mistral-large-latest",
    messages=messages
)

print("Assistant:", response.choices[0].message.content)

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {"role": "user", "content": "Write a short poem about artificial intelligence."}
    ],
    temperature=0.7,
    maxtokens=500,
    topp=0.95,

)

print(response.choices[0].message.content)

Token usage information
usage = response.usage
print(f"\nPrompt tokens: {usage.prompttokens}")
print(f"Completion tokens: {usage.completiontokens}")

print(f"Total tokens: {usage.totaltokens}")

Streaming Responses

For longer responses, streaming allows displaying output progressively:

stream = client.chat.stream(
    model="mistral-large-latest",
    messages=[
        {"role": "user", "content": "Explain the Transformer architecture in detail."}
    ]
)

for chunk in stream:
    content = chunk.data.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

print()

Streaming with Response Collection

fullresponse = ""

stream = client.chat.stream(
    model="mistral-large-latest",
    messages=[
        {"role": "user", "content": "Create a brief tutorial about pandas DataFrames."}
    ]
)

for chunk in stream:
    content = chunk.data.choices[0].delta.content
    if content:
        fullresponse += content
        print(content, end="", flush=True)

print(f"\n\nTotal characters: {len(fullresponse)}")

import json

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {
            "role": "user",
            "content": """Analyze the sentiment of the following review and provide output in JSON format 
            with fields: text, sentiment (positive/negative/neutral), confidence (0-1), keywords.

            Review: "This product is amazing! The quality is outstanding and shipping was fast. 
            Highly recommended for everyone."
            """
        }
    ],
    responseformat={"type": "jsonobject"}

)

result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2))

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {
            "role": "user",
            "content": """Extract information from the following text into JSON format:

            "TechForward Inc., located at 456 Innovation Drive, San Francisco, 
            was founded in 2018 by Sarah Chen. The company operates in the 
            artificial intelligence sector with 250 employees."

            Schema: {companyname, address, foundedyear, founder, industry, employeecount}"""
        }
    ],
    responseformat={"type": "jsonobject"}
)

data = json.loads(response.choices[0].message.content)
print(json.dumps(data, indent=2))

Function Calling (Tool Use)

One of Mistral AI's powerful features is function calling capability:

Defining Tools

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "getweather",

            "description": "Get weather information for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g., London, Tokyo"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculatebmi",
            "description": "Calculate Body Mass Index (BMI)",
            "parameters": {
                "type": "object",
                "properties": {
                    "weightkg": {

                        "type": "number",
                        "description": "Body weight in kilograms"
                    },
                    "heightcm": {
                        "type": "number",
                        "description": "Height in centimeters"
                    }
                },
                "required": ["weightkg", "heightcm"]
            }
        }
    }
]

Executing Function Calling

def getweather(city, unit="celsius"):
    weatherdata = {
        "London": {"temp": 15, "condition": "Partly cloudy", "humidity": 72},
        "Tokyo": {"temp": 28, "condition": "Sunny", "humidity": 60},
        "New York": {"temp": 22, "condition": "Clear", "humidity": 55},
    }
    data = weatherdata.get(city, {"temp": 20, "condition": "Unknown", "humidity": 65})

    if unit == "fahrenheit":
        data["temp"] = data["temp"]  9/5 + 32

    return json.dumps(data)

def calculatebmi(weightkg, heightcm):
    heightm = heightcm / 100
    bmi = weightkg / (heightm  2)

    if bmi < 18.5:
        category = "Underweight"
    elif bmi < 25:
        category = "Normal"
    elif bmi < 30:
        category = "Overweight"
    else:
        category = "Obese"
    return json.dumps({"bmi": round(bmi, 1), "category": category})

availablefunctions = {

    "getweather": getweather,

    "calculatebmi": calculatebmi

}

messages = [
    {"role": "user", "content": "What's the weather like in Tokyo today?"}
]

response = client.chat.complete(
    model="mistral-large-latest",
    messages=messages,
    tools=tools,
    toolchoice="auto"
)

message = response.choices[0].message

if message.toolcalls:

    messages.append(message)

    for toolcall in message.toolcalls:

        functionname = toolcall.function.name

        functionargs = json.loads(toolcall.function.arguments)


        print(f"Calling function: {functionname}({functionargs})")


        functionresult = availablefunctionsfunctionname

        messages.append({
            "role": "tool",
            "name": functionname,
            "content": functionresult,

            "toolcallid": toolcall.id
        })

    finalresponse = client.chat.complete(

        model="mistral-large-latest",
        messages=messages,
        tools=tools
    )

    print(finalresponse.choices[0].message.content)

response = client.embeddings.create(
    model="mistral-embed",
    inputs=["Machine learning is a branch of artificial intelligence."]
)

embedding = response.data[0].embedding
print(f"Embedding dimensions: {len(embedding)}")
print(f"Sample values: {embedding[:5]}")

import numpy as np

documents = [
    "Python is a popular programming language for data science.",
    "TensorFlow is a deep learning framework from Google.",
    "Docker simplifies application deployment using containers.",
    "Pandas is used for tabular data manipulation and analysis.",
    "Kubernetes is a container orchestration platform.",
    "Scikit-learn provides classic machine learning algorithms.",
    "React is a JavaScript library for building user interfaces.",
    "FastAPI is a modern Python framework for building APIs.",
]

def getembeddings(texts):

    response = client.embeddings.create(
        model="mistral-embed",
        inputs=texts
    )
    return [item.embedding for item in response.data]

docembeddings = getembeddings(documents)


def cosinesimilarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a)  np.linalg.norm(b))

def search(query, topk=3):

    queryembedding = getembeddings([query])[0]


    similarities = []
    for i, docemb in enumerate(docembeddings):

        sim = cosinesimilarity(queryembedding, docemb)
        similarities.append((i, sim))

    similarities.sort(key=lambda x: x[1], reverse=True)

    print(f"Query: {query}\n")
    for idx, sim in similarities[:topk]:

        print(f"  [{sim:.4f}] {documents[idx]}")
    print()

search("library for data analysis")
search("how to deploy applications")
search("machine learning framework")

import asyncio
from mistralai import Mistral

async def processbatch(prompts):
    client = Mistral(apikey=os.environ["MISTRALAPIKEY"])


    async def processsingle(prompt):
        response = await client.chat.completeasync(

            model="mistral-small-latest",
            messages=[{"role": "user", "content": prompt}],
            maxtokens=200
        )
        return response.choices[0].message.content

    tasks = [processsingle(prompt) for prompt in prompts]

    results = await asyncio.gather(tasks)

    return results

prompts = [
    "What is a neural network?",
    "Explain the concept of gradient descent.",
    "What is the difference between CNN and RNN?",
    "What is transfer learning?",
    "Explain the attention mechanism.",
]

results = asyncio.run(processbatch(prompts))

for prompt, result in zip(prompts, results):
    print(f"Q: {prompt}")
    print(f"A: {result[:100]}...")
    print()

import time
from mistralai import Mistral
from mistralai.models import HTTPValidationError

def chatwithretry(client, messages, model="mistral-large-latest", maxretries=3):

    for attempt in range(maxretries):
        try:
            response = client.chat.complete(
                model=model,
                messages=messages
            )
            return response.choices[0].message.content
        except HTTPValidationError as e:
            print(f"Validation error: {e}")
            raise
        except Exception as e:
            if attempt < maxretries - 1:

                waittime = 2  attempt

                print(f"Error: {e}. Retrying in {waittime}s...")

                time.sleep(waittime)
            else:
                raise

result = chatwithretry(
    client,
    [{"role": "user", "content": "Hello, how are you?"}]
)
print(result)

knowledgebase = {
    "producta": "Product A is a project management software with Kanban board, time tracking, and reporting features. Pricing starts at $50/month for teams up to 10 people.",
    "productb": "Product B is a data analytics platform supporting real-time visualization, custom dashboards, and integration with 50+ data sources. Pricing starts at $150/month.",

    "refundpolicy": "Refunds can be requested within 30 days of purchase. The refund process takes 5-7 business days. Refunds do not apply to annual plans past 14 days.",
    "operatinghours": "Customer support is available Monday-Friday, 9:00 AM - 6:00 PM EST. For high-urgency issues, a 24/7 hotline is available at 1-800-555-0123."

}

def retrievecontext(query, knowledgebase):

    queryembedding = getembeddings([query])[0]


    kbtexts = list(knowledgebase.values())

    kbkeys = list(knowledgebase.keys())

    kbembeddings = getembeddings(kbtexts)

    similarities = []
    for i, emb in enumerate(kbembeddings):

        sim = cosinesimilarity(queryembedding, emb)

        similarities.append((i, sim))

    similarities.sort(key=lambda x: x[1], reverse=True)
    topresults = similarities[:2]

    context = "\n\n".join([kbtexts[idx] for idx,  in topresults])

    return context

def ragchat(query):
    context = retrievecontext(query, knowledgebase)

    messages = [
        {
            "role": "system",
            "content": f"""You are a customer support assistant. Answer questions based on the provided context.
            If the information is not available in the context, say that you don't have that information.

            Context:
            {context}"""
        },
        {"role": "user", "content": query}
    ]

    response = client.chat.complete(
        model="mistral-large-latest",
        messages=messages,
        temperature=0.3
    )

    return response.choices[0].message.content

print(ragchat("How much does the data analytics product cost?"))

print()
print(ragchat("How do I request a refund?"))

import base64

def encodeimage(imagepath):
    with open(imagepath, "rb") as f:

        return base64.b64encode(f.read()).decode("utf-8")

imagebase64 = encodeimage("chart.png")


response = client.chat.complete(
    model="pixtral-large-latest",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "imageurl",
                    "imageurl": {

                        "url": f"data:image/png;base64,{imagebase64}"
                    }
                },
                {
                    "type": "text",
                    "text": "Analyze this chart. What are the main trends and what insights can be drawn?"
                }
            ]
        }
    ]
)

print(response.choices[0].message.content)

response = client.chat.complete(
    model="codestral-latest",
    messages=[
        {
            "role": "system",
            "content": "You are an expert Python developer. Write clean, efficient, and well-documented code."
        },
        {
            "role": "user",
            "content": """Create a Python class for managing PostgreSQL database connections 
            with connection pooling, retry logic, and context manager support."""
        }
    ],
    temperature=0.2
)

print(response.choices[0].message.content)

def smartmodelselection(taskcomplexity):
    if taskcomplexity == "simple":
        return "mistral-small-latest"
    elif taskcomplexity == "medium":

        return "mistral-medium-latest"
    else:
        return "mistral-large-latest"

def estimatecost(prompttokens, completiontokens, model):
    pricing = {
        "mistral-small-latest": {"input": 0.001, "output": 0.003},
        "mistral-large-latest": {"input": 0.003, "output": 0.009},
    }
    rates = pricing.get(model, pricing["mistral-small-latest"])
    cost = (prompttokens / 1000  rates["input"]) + (completiontokens / 1000  rates["output"])

    return cost

systemprompt = """You are a professional data analyst assistant.

RULES:
Always provide answers based on data and facts
If uncertain, communicate your confidence level
Use structured formats for complex answers
Provide actionable recommendations

OUTPUT FORMAT:
Use bullet points for lists
Use tables for comparisons
Include a summary at the end"""

response = client.chat.complete(
    model="mistral-large-latest",
    messages=[
        {"role": "system", "content": systemprompt},
        {"role": "user", "content": "Compare PostgreSQL vs MongoDB for an e-commerce application."}
    ],
    temperature=0.3
)

import time
from collections import deque

class RateLimiter:
    def init(self, maxrequestsperminute=60):

        self.maxrequests = maxrequestsperminute

        self.requests = deque()

    def waitifneeded(self):

        now = time.time()
        while self.requests and self.requests[0] < now - 60:
            self.requests.popleft()

        if len(self.requests) >= self.maxrequests:
            sleeptime = 60 - (now - self.requests[0])

            if sleeptime > 0:
                time.sleep(sleeptime)


        self.requests.append(time.time())

ratelimiter = RateLimiter(maxrequestsperminute=50)


def ratelimitedchat(messages):

    ratelimiter.waitifneeded()
    return client.chat.complete(
        model="mistral-small-latest",
        messages=messages
    )

import hashlib
import json

class ResponseCache:
    def init(self):
        self.cache = {}

    def makekey(self, model, messages, kwargs):

        content = json.dumps({"model": model, "messages": messages, kwargs}, sortkeys=True)

        return hashlib.md5(content.encode()).hexdigest()

    def getorcreate(self, client, model, messages, kwargs):

        key = self.makekey(model, messages, kwargs)


        if key in self.cache:
            print("Cache hit!")
            return self.cache[key]

        response = client.chat.complete(model=model, messages=messages, *kwargs)
        result = response.choices[0].message.content
        self.cache[key] = result
        return result

cache = ResponseCache()

result1 = cache.getorcreate(client, "mistral-small-latest", [{"role": "user", "content": "What is an API?"}])

result2 = cache.getorcreate(client, "mistral-small-latest", [{"role": "user", "content": "What is an API?"}])

import os
import json
from mistralai import Mistral

class CustomerServiceBot:
    def init(self):
        self.client = Mistral(apikey=os.environ["MISTRALAPIKEY"])
        self.conversationhistory = []

        self.systemprompt = """You are a customer service bot for TechMart online store.

        Company information:
Operating hours: Monday-Saturday, 9:00 AM - 8:00 PM EST
Email: support@techmart.com
Phone: 1-800-555-TECH
        

        Policies:
Product warranty: 1 year for electronics, 6 months for accessories
Returns: 7 days after receiving the item
Free shipping for orders over $50
        

        Guidelines:
Respond warmly and professionally
If you cannot help, direct to human customer service
Use clear and concise language"""
        

        self.conversationhistory.append({

            "role": "system",
            "content": self.systemprompt
        })

    def chat(self, usermessage):

        self.conversationhistory.append({
            "role": "user",
            "content": usermessage

        })

        response = self.client.chat.complete(
            model="mistral-small-latest",
            messages=self.conversationhistory,
            temperature=0.3,
            maxtokens=500

        )

        assistantmessage = response.choices[0].message.content

        self.conversationhistory.append({

            "role": "assistant",
            "content": assistantmessage
        })

        return assistantmessage


    def reset(self):
        self.conversationhistory = [{"role": "system", "content": self.system_prompt}]

bot = CustomerServiceBot()

questions = [
    "Hi, I'd like to ask about your return policy.",
    "What if the item arrived damaged?",
    "How long does the refund process take?",
]

for q in questions:
    print(f"User: {q}")
    answer = bot.chat(q)
    print(f"Bot: {answer}\n")