Building Autonomous AI Agents with Tool Calling, Streaming SSE & MCP in Node.js & Vue 3

// Definition conforming to OpenAI / Anthropic / MCP Tool Schema
export const queryDatabaseTool = {
  name: "query_sales_analytics",
  description: "Queries the aggregated database for sales revenue, order volumes, and customer cohort metrics within a specific date range.",
  parameters: {
    type: "object",
    properties: {
      startDate: {
        type: "string",
        description: "ISO 8601 start date (e.g. 2026-01-01)"
      },
      endDate: {
        type: "string",
        description: "ISO 8601 end date (e.g. 2026-03-31)"
      },
      metric: {
        type: "string",
        enum: ["revenue", "order_count", "average_order_value"],
        description: "The metric calculation requested by the user"
      }
    },
    required: ["startDate", "endDate", "metric"],
    additionalProperties: false
  },
  // Sandboxed Tool Implementation
  async execute({ startDate, endDate, metric }) {
    // Validate inputs & execute parameterized query safely
    const result = await db.query(
      `SELECT calculate_metric($1, $2, $3) AS data`,
      [startDate, endDate, metric]
    );
    return JSON.stringify(result.rows);
  }
};

import express from 'express';
import { GoogleGenerativeAI } from '@google/generative-ai';
import { queryDatabaseTool } from './tools/analyticsTool.js';

const app = express();
app.use(express.json());

const toolRegistry = {
  [queryDatabaseTool.name]: queryDatabaseTool.execute
};

app.post('/api/agent/stream', async (req, res) => {
  const { prompt, conversationHistory = [] } = req.body;

  // Initialize Server-Sent Events headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache, no-transform');
  res.setHeader('Connection', 'keep-alive');

  const sendEvent = (event, data) => {
    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
  };

  try {
    const messages = [...conversationHistory, { role: 'user', content: prompt }];
    let iterations = 0;
    const MAX_STEPS = 6; // Guard against infinite reasoning loops

    while (iterations < MAX_STEPS) {
      iterations++;
      sendEvent('status', { step: iterations, message: 'Reasoning...' });

      // Call LLM with tool declarations
      const response = await aiClient.chat.completions.create({
        model: 'gpt-4o', // or gemini-2.0-flash / claude-3-5-sonnet
        messages,
        tools: [{ type: 'function', function: queryDatabaseTool }],
        stream: true
      });

      let fullContent = '';
      let toolCallChunks = [];

      for await (const chunk of response) {
        const delta = chunk.choices[0]?.delta;
        
        if (delta?.content) {
          fullContent += delta.content;
          sendEvent('token', { text: delta.content });
        }

        if (delta?.tool_calls) {
          toolCallChunks.push(delta.tool_calls);
        }
      }

      // If no tool call was requested, finish response
      if (!toolCallChunks.length) {
        sendEvent('done', { totalSteps: iterations });
        break;
      }

      // Execute Tool Calls
      const toolCall = assembleToolCall(toolCallChunks);
      sendEvent('tool_start', { name: toolCall.name, args: toolCall.args });

      const executor = toolRegistry[toolCall.name];
      const toolOutput = await executor(JSON.parse(toolCall.args));

      sendEvent('tool_result', { name: toolCall.name, result: toolOutput });

      // Append assistant's tool call & tool result to memory for next turn
      messages.push({ role: 'assistant', tool_calls: [toolCall] });
      messages.push({ role: 'tool', tool_call_id: toolCall.id, content: toolOutput });
    }

    res.end();
  } catch (error) {
    sendEvent('error', { message: error.message });
    res.end();
  }
});

import { ref } from 'vue';

export function useAgentStream() {
  const messages = ref([]);
  const isStreaming = ref(false);
  const activeTool = ref(null);

  async function executeAgent(userPrompt) {
    isStreaming.value = true;
    messages.value.push({ role: 'user', text: userPrompt });

    const assistantMsg = ref({ role: 'assistant', text: '', toolsExecuted: [] });
    messages.value.push(assistantMsg.value);

    const response = await fetch('/api/agent/stream', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ prompt: userPrompt })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder('utf-8');
    let buffer = '';

    while (true) {
      const { value, done } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n\n');
      buffer = lines.pop(); // Keep partial stream chunk

      for (const block of lines) {
        if (!block.trim()) continue;
        const [eventLine, dataLine] = block.split('\n');
        const event = eventLine.replace('event: ', '').trim();
        const data = JSON.parse(dataLine.replace('data: ', '').trim());

        if (event === 'token') {
          assistantMsg.value.text += data.text;
        } else if (event === 'tool_start') {
          activeTool.value = data.name;
        } else if (event === 'tool_result') {
          assistantMsg.value.toolsExecuted.push({ name: data.name, output: data.result });
          activeTool.value = null;
        }
      }
    }

    isStreaming.value = false;
  }

  return { messages, isStreaming, activeTool, executeAgent };
}