Mastra for Real-Time Data Streaming: Building Event-Driven Agents with WebSocket Integrations

The landscape of AI application development is shifting rapidly from batch processing to real-time responsiveness. In modern enterprise and consumer applications, AI agents cannot afford to wait for a complete session to end before reacting. They need to process streaming data, maintain state across continuous interactions, and deliver sub-second feedback. This is where Mastra, a TypeScript-first AI agent framework, shines. By natively supporting asynchronous workflows and integrations with real-time transport layers like WebSockets, Mastra allows developers to build event-driven agents that feel truly alive.

Why Event-Driven Architecture Matters for AI Agents

Traditional request-response models are ill-suited for conversational AI or data-heavy monitoring systems. An event-driven approach allows an agent to react to discrete occurrences—such as a new user message, a sensor reading, or a financial tick—without blocking the main thread. When combined with WebSockets, which provide full-duplex communication channels, the latency between the event occurrence and the agent's action is minimized. This is critical for applications like live trading bots, real-time customer support, or collaborative coding assistants where every millisecond counts.

Setting Up the Mastra Agent

Mastra simplifies the creation of AI agents by allowing you to define tools, prompts, and logic in a declarative manner. To start, you need to install the core package and set up a basic agent structure. The agent will be responsible for processing incoming WebSocket messages and generating contextual responses.

import { createAgent } from 'mastra';
import { openai } from '@mastra/core/models';

export const realtimeAgent = createAgent({
  name: 'Realtime Support Agent',
  description: 'An agent that processes real-time user queries',
  model: openai('gpt-4o'),
  instructions: `
    You are a helpful assistant. 
    Keep your responses concise as they are being streamed.
    Acknowledge immediate user intents before providing detailed answers.
  `,
  tools: {
    lookupOrder: async ({ orderId }) => {
      // Simulate a fast database lookup
      return { status: 'processing', estimatedTime: '2 days' };
    }
  }
});

Integrating WebSockets with Mastra

The power of this setup lies in how we connect the agent to the client. We use a WebSocket server to maintain a persistent connection. When a message arrives, we trigger the Mastra agent to stream its response back directly to the socket. This avoids the overhead of closing and reopening HTTP connections for each interaction.

Below is a practical example using Node.js and the ws library to handle the WebSocket layer. Note how we utilize Mastra's streaming capabilities to push tokens to the client as they are generated.

import WebSocket, { WebSocketServer } from 'ws';
import { server } from 'http';
import { realtimeAgent } from './agent'; // Import from above

const server = http.createServer();
const wss = new WebSocketServer({ server });

wss.on('connection', (ws) => {
  console.log('Client connected');

  ws.on('message', async (message) => {
    try {
      const userPrompt = message.toString();
      
      // Use Mastra's stream method for real-time output
      const stream = await realtimeAgent.stream(userPrompt);

      // Forward chunks as they arrive
      for await (const chunk of stream) {
        if (chunk.type === 'text-delta') {
          ws.send(JSON.stringify({ 
            type: 'token', 
            content: chunk.text 
          }));
        }
      }
      
      ws.send(JSON.stringify({ type: 'done' }));
    } catch (error) {
      console.error('Agent error:', error);
      ws.send(JSON.stringify({ type: 'error', message: 'Something went wrong' }));
    }
  });

  ws.on('close', () => {
    console.log('Client disconnected');
  });
});

server.listen(8080, () => {
  console.log('WebSocket server running on ws://localhost:8080');
});

Handling State and Concurrency

In real-time scenarios, multiple messages might arrive in quick succession. Mastra agents are stateless by default per invocation, but they can maintain context through memory modules or external state stores. For high-concurrency WebSocket servers, it is crucial to ensure that each WebSocket connection maps to a specific agent context or session ID. This prevents data leakage between users and ensures that the "conversation history" is correctly scoped to the specific client connection. You can extend the example above by passing a sessionId from the WebSocket metadata into the Mastra agent's options to retrieve relevant past interactions.

Conclusion

Building real-time AI agents requires more than just connecting an LLM to a database; it demands a robust event-driven architecture that can handle persistent connections and streaming data. Mastra provides the ideal abstraction layer to manage agent logic, while WebSockets provide the transport mechanism for low-latency communication. By combining these technologies, developers can create responsive, intelligent applications that react to the world in real-time, offering a superior user experience in dynamic environments.