The MistralAzure Client in Python — Mistral Deployment: Cloud and Self-Hosting

Coding with MistralAzure

Your Azure endpoint is deployed and responding; now you need to write the code that will use it reliably. This lesson details the Python MistralAzure client: initialization, the patterns you’ll use daily—simple calls, streaming, asynchronous execution, and function calling—and the few specifics that distinguish Azure from the direct API.

Installation and Configuration

Updated November 2, 2026 — Mistral SDK 2 and later. Since version 2.0 of the Python SDK (March 2026), the Azure client imports from mistralai.azure.client, and its version 1 parameters azure_endpoint and azure_api_key have become server_url and api_key. The code in this lesson follows the current version (3.7.0, November 1, 2026). Sources: v2.0.0 release notes · Mistral on Azure.

The Azure client is included in the mistralai package; no additional dependencies are required.

pip install mistralai

Initialization requires two parameters:

import os
from mistralai.azure.client import MistralAzure

client = MistralAzure(
    server_url=os.environ["AZUREAI_ENDPOINT"],
    api_key=os.environ["AZUREAI_API_KEY"]
)

server_url — Your Azure endpoint URL
api_key — The endpoint’s authentication key

A third, optional parameter, api_version, defaults to 2024-05-01-preview. The SDK adds this to every request automatically as a URL parameter.

Chat Completions

From here, the experience mirrors the standard client. Use a system message to frame the model’s role, a user message for the question, and read the response from choices[0].message.content.

response = client.chat.complete(
    model="azureai",
    messages=[
        {"role": "system", "content": "You are a technical assistant specializing in Python."},
        {"role": "user", "content": "How do I implement a cache decorator?"}
    ]
)

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

Generation parameters behave exactly as they do with the direct Mistral API. If a response is too unpredictable or verbose, correct it here, not in the prompt.

response = client.chat.complete(
    model="azureai",
    messages=[
        {"role": "user", "content": "Generate a haiku about cloud computing."}
    ],
    temperature=0.7,
    max_tokens=150,
    top_p=0.95
)

temperature — Controls creativity (0.0 = deterministic, 1.0 = creative)
max_tokens — Limits response length
top_p — Nucleus sampling

Streaming

For responses spanning several hundred words, a standard call leaves the user staring at a frozen screen for several seconds. Streaming eliminates this perceived wait by displaying text as it generates. This is the default mode whenever a human interface is involved.

stream = client.chat.stream(
    model="azureai",
    messages=[
        {"role": "user", "content": "Explain how a load balancer works."}
    ]
)

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

Note the if content check: some chunks arrive empty and would print None to the screen without this precaution.

Asynchronous Calls

In a web server—FastAPI, Django async—a blocking call ties up a worker for the entire generation duration, often several seconds per request. The asynchronous client yields control during the wait, allowing a single process to handle dozens of concurrent requests.

import asyncio
from mistralai.azure.client import MistralAzure

client = MistralAzure(
    server_url=os.environ["AZUREAI_ENDPOINT"],
    api_key=os.environ["AZUREAI_API_KEY"]
)

async def generate_response(prompt: str) -> str:
    response = await client.chat.complete_async(
        model="azureai",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Execution
result = asyncio.run(generate_response("What is Kubernetes?"))
print(result)

Function Calling

Function calling behaves identically to the direct API. You describe available tools, the model decides whether to call one, and your code actually executes the function—the model only formulates the intent and arguments.

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "Name of the city"}
                },
                "required": ["city"]
            }
        }
    }
]

response = client.chat.complete(
    model="azureai",
    messages=[{"role": "user", "content": "What is the weather in Paris?"}],
    tools=tools,
    tool_choice="auto"
)

# Check if the model wants to call a function
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"Function called: {tool_call.function.name}")
    print(f"Arguments: {args}")

Error Handling

In production, three families of errors recur, and they demand different responses. A 429 signals a rate limit and requires a retry with exponential backoff. A 401 indicates an invalid or expired key: retrying is useless, so fail immediately with a clear message. A 404 almost always points to a misconfigured AZUREAI_ENDPOINT variable. Distinguishing these cases saves you from spending an hour debugging a typo in a URL.

import time
from mistralai.azure.client import MistralAzure

def robust_call(client, messages, max_retries=3):
    """Call with exponential retry and error handling."""
    for attempt in range(max_retries):
        try:
            return client.chat.complete(
                model="azureai",
                messages=messages
            )
        except Exception as e:
            error_str = str(e)
            if "429" in error_str:
                wait = 2 ** attempt
                print(f"Rate limit — retrying in {wait}s")
                time.sleep(wait)
            elif "401" in error_str:
                raise ValueError("Invalid or expired API key") from e
            elif "404" in error_str:
                raise ValueError("Endpoint not found — check AZUREAI_ENDPOINT") from e
            else:
                if attempt == max_retries - 1:
                    raise
                time.sleep(1)
    raise RuntimeError("Failed after all retries")

Measuring Consumption

Every response includes a usage object detailing consumed tokens. Read this systematically: it is the only reliable measure for estimating costs and spotting a system prompt that has become bloated from repeated edits.

response = client.chat.complete(
    model="azureai",
    messages=[{"role": "user", "content": "Hello"}]
)

usage = response.usage
print(f"Prompt tokens  : {usage.prompt_tokens}")
print(f"Response tokens: {usage.completion_tokens}")
print(f"Total tokens   : {usage.total_tokens}")

Key Takeaways

The MistralAzure client works like the standard client, using model="azureai"
Streaming (chat.stream) is recommended for user interfaces
Asynchronous calls (complete_async) are essential for web servers
Function calling and JSON mode work identically to the direct API
Always implement exponential retry to handle 429 errors