Mistral AI Events & Async Patterns

Overview

Implement async patterns and event handling for Mistral AI integrations. Note: Mistral AI does not have native webhooks, so this skill covers async patterns and event-driven architectures.

Prerequisites

• Mistral AI SDK installed

• Queue system (Redis, SQS, etc.) for async processing

• Event emitter or pub/sub for notifications

• Background job processor (BullMQ, etc.)

Instructions

Step 1: Event-Driven Chat Architecture

import { EventEmitter } from 'events'; import Mistral from '@mistralai/mistralai';

// Custom event types interface MistralEvents { 'chat:start': { requestId: string; model: string; timestamp: Date }; 'chat:chunk': { requestId: string; content: string; index: number }; 'chat:complete': { requestId: string; fullResponse: string; usage: any }; 'chat:error': { requestId: string; error: Error }; }

class MistralEventEmitter extends EventEmitter { private client: Mistral;

constructor() { super(); this.client = new Mistral({ apiKey: process.env.MISTRAL_API_KEY! }); }

async chat(requestId: string, messages: any[], model = 'mistral-small-latest') { this.emit('chat:start', { requestId, model, timestamp: new Date() });

try {
  const stream = await this.client.chat.stream({ model, messages });

  let fullResponse = '';
  let index = 0;

  for await (const event of stream) {
    const content = event.data?.choices?.[0]?.delta?.content;
    if (content) {
      fullResponse += content;
      this.emit('chat:chunk', { requestId, content, index: index++ });
    }
  }

  const usage = { totalTokens: fullResponse.length / 4 }; // Estimate
  this.emit('chat:complete', { requestId, fullResponse, usage });

  return fullResponse;
} catch (error) {
  this.emit('chat:error', { requestId, error: error as Error });
  throw error;
}

} }

// Usage const mistral = new MistralEventEmitter();

mistral.on('chat:start', ({ requestId, model }) => { console.log([${requestId}] Starting chat with ${model}); });

mistral.on('chat:chunk', ({ requestId, content }) => { process.stdout.write(content); });

mistral.on('chat:complete', ({ requestId, usage }) => { console.log(\n[${requestId}] Complete. Tokens: ${usage.totalTokens}); });

mistral.on('chat:error', ({ requestId, error }) => { console.error([${requestId}] Error:, error.message); });

await mistral.chat('req-123', [{ role: 'user', content: 'Hello!' }]);

Step 2: Background Job Processing with BullMQ

import { Queue, Worker, Job } from 'bullmq'; import Mistral from '@mistralai/mistralai'; import Redis from 'ioredis';

const connection = new Redis(process.env.REDIS_URL);

// Define job types interface ChatJob { id: string; messages: Array<{ role: string; content: string }>; model: string; callback?: string; // Webhook URL to call on completion }

// Create queue const chatQueue = new Queue<ChatJob>('mistral-chat', { connection });

// Create worker const chatWorker = new Worker<ChatJob>( 'mistral-chat', async (job: Job<ChatJob>) => { const client = new Mistral({ apiKey: process.env.MISTRAL_API_KEY! });

const response = await client.chat.complete({
  model: job.data.model,
  messages: job.data.messages,
});

const result = {
  jobId: job.id,
  requestId: job.data.id,
  content: response.choices?.[0]?.message?.content,
  usage: response.usage,
  completedAt: new Date().toISOString(),
};

// Call webhook if provided
if (job.data.callback) {
  await fetch(job.data.callback, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(result),
  });
}

return result;

}, { connection, concurrency: 5, limiter: { max: 10, duration: 1000, // Max 10 jobs per second # 1000: 1 second in ms }, } );

// Event handlers chatWorker.on('completed', (job, result) => { console.log(Job ${job.id} completed:, result); });

chatWorker.on('failed', (job, err) => { console.error(Job ${job?.id} failed:, err.message); });

// API endpoint to queue jobs export async function POST(request: Request) { const body = await request.json();

const job = await chatQueue.add('chat', { id: crypto.randomUUID(), messages: body.messages, model: body.model || 'mistral-small-latest', callback: body.callback, }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 }, # 1 second in ms });

return Response.json({ jobId: job.id, status: 'queued', statusUrl: /api/jobs/${job.id}, }); }

Step 3: Webhook Notification System

import crypto from 'crypto';

interface WebhookPayload { event: string; timestamp: string; data: any; }

class WebhookNotifier { private secret: string;

constructor(secret: string) { this.secret = secret; }

private sign(payload: string): string { return crypto .createHmac('sha256', this.secret) .update(payload) .digest('hex'); }

async notify(url: string, event: string, data: any): Promise<boolean> { const payload: WebhookPayload = { event, timestamp: new Date().toISOString(), data, };

const body = JSON.stringify(payload);
const signature = this.sign(body);

try {
  const response = await fetch(url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Webhook-Signature': signature,
      'X-Webhook-Timestamp': payload.timestamp,
    },
    body,
  });

  return response.ok;
} catch (error) {
  console.error('Webhook failed:', error);
  return false;
}

} }

// Usage const notifier = new WebhookNotifier(process.env.WEBHOOK_SECRET!);

// After Mistral completion await notifier.notify( 'https://yourapp.com/webhooks/mistral', 'chat.completed', { requestId: 'req-123', content: response.choices?.[0]?.message?.content, usage: response.usage, } );

Step 4: Server-Sent Events (SSE) for Real-Time Updates

// api/chat/stream/route.ts import Mistral from '@mistralai/mistralai';

export async function POST(request: Request) { const { messages } = await request.json();

const client = new Mistral({ apiKey: process.env.MISTRAL_API_KEY! });

const stream = await client.chat.stream({ model: 'mistral-small-latest', messages, });

const encoder = new TextEncoder();

const readable = new ReadableStream({ async start(controller) { // Send initial event controller.enqueue(encoder.encode(event: start\ndata: {"status":"started"}\n\n));

  try {
    let tokenCount = 0;
    for await (const event of stream) {
      const content = event.data?.choices?.[0]?.delta?.content;
      if (content) {
        tokenCount++;
        controller.enqueue(
          encoder.encode(`event: chunk\ndata: ${JSON.stringify({ content, tokenCount })}\n\n`)
        );
      }
    }

    // Send completion event
    controller.enqueue(
      encoder.encode(`event: complete\ndata: ${JSON.stringify({ totalTokens: tokenCount })}\n\n`)
    );
  } catch (error: any) {
    controller.enqueue(
      encoder.encode(`event: error\ndata: ${JSON.stringify({ error: error.message })}\n\n`)
    );
  }

  controller.close();
},

});

return new Response(readable, { headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', }, }); }

Step 5: Client-Side SSE Consumption

// Client-side JavaScript function streamChat(messages: any[]): EventSource { const eventSource = new EventSource('/api/chat/stream', { // Note: EventSource doesn't support POST directly // Use fetch with ReadableStream instead, or use a library });

eventSource.addEventListener('start', (e) => { console.log('Stream started'); });

eventSource.addEventListener('chunk', (e) => { const { content } = JSON.parse(e.data); process.stdout.write(content); });

eventSource.addEventListener('complete', (e) => { const { totalTokens } = JSON.parse(e.data); console.log(\nComplete. Tokens: ${totalTokens}); eventSource.close(); });

eventSource.addEventListener('error', (e) => { console.error('Stream error:', e); eventSource.close(); });

return eventSource; }

Output

• Event-driven Mistral AI integration

• Background job processing

• Webhook notification system

• Real-time streaming with SSE

Error Handling

Issue Cause Solution

Job stuck in queue Worker crashed Implement job timeout and retry

Webhook failed Network/auth issue Retry with exponential backoff

SSE disconnected Client timeout Implement reconnection logic

Event backpressure Too many events Implement buffering

Examples

Python Async Pattern

import asyncio from mistralai import Mistral

async def process_batch(prompts: list[str]): client = Mistral(api_key=os.environ.get("MISTRAL_API_KEY"))

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

results = await asyncio.gather(*[process_one(p) for p in prompts])
return results

Resources

• BullMQ Documentation

• Server-Sent Events MDN

• Mistral AI Streaming

Next Steps

For performance optimization, see mistral-performance-tuning .