Express.js stream

Hello! How can i pipe with response object from Express.js when using .stream() function?

5 Replies

Hi @Radu Gabor ! Let me get that for you you can do something like this:

import * as handlers from "@mastra/server/handlers";
import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import express, { Request, Response } from "express";
import { toAISdkFormat } from "@mastra/ai-sdk";
import { pipeUIMessageStreamToResponse } from "ai";
import "dotenv/config";

const myAgent = new Agent({
  name: "My Agent",
  instructions: "you are a helpful agent that answers questions",
  model: "openrouter/openai/gpt-4o-mini",
});

const mastra = new Mastra({
  agents: {
    myAgent,
  },
});

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

const port = process.env.PORT ?? 3000;

app.post("/agent/stream", async (req: Request, res: Response) => {
  // get prompt from req body
  const { prompt } = req.body;

  // Set SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("Access-Control-Allow-Origin", "*");
  res.setHeader("Access-Control-Allow-Headers", "Cache-Control");

  try {
    const result = await myAgent.stream(prompt);

    pipeUIMessageStreamToResponse({
      response: res,
      stream: toAISdkFormat(result, { from: "agent" }),
    });
  } catch (error) {
    console.error("Stream error:", error);
    if (!res.headersSent) {
      res.writeHead(500);
      res.end(`data: {"error": "Stream failed"}\n\n`);
    }
  }
});

app.listen(port, () => {
  console.log(`Server listening on port ${port}`);
});

import * as handlers from "@mastra/server/handlers";
import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import express, { Request, Response } from "express";
import { toAISdkFormat } from "@mastra/ai-sdk";
import { pipeUIMessageStreamToResponse } from "ai";
import "dotenv/config";

const myAgent = new Agent({
  name: "My Agent",
  instructions: "you are a helpful agent that answers questions",
  model: "openrouter/openai/gpt-4o-mini",
});

const mastra = new Mastra({
  agents: {
    myAgent,
  },
});

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

const port = process.env.PORT ?? 3000;

app.post("/agent/stream", async (req: Request, res: Response) => {
  // get prompt from req body
  const { prompt } = req.body;

  // Set SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("Access-Control-Allow-Origin", "*");
  res.setHeader("Access-Control-Allow-Headers", "Cache-Control");

  try {
    const result = await myAgent.stream(prompt);

    pipeUIMessageStreamToResponse({
      response: res,
      stream: toAISdkFormat(result, { from: "agent" }),
    });
  } catch (error) {
    console.error("Stream error:", error);
    if (!res.headersSent) {
      res.writeHead(500);
      res.end(`data: {"error": "Stream failed"}\n\n`);
    }
  }
});

app.listen(port, () => {
  console.log(`Server listening on port ${port}`);
});

Radu GaborOP•5d ago

Thank you @Romain! @Romain another question if i may, is there a way to tranform the stream to a AI SDK v4 format?

_roamin_•5d ago

Unfortunately, we don't support the v4 format, but you should be able to create a stream transform to turn the v5 stream into a v4 stream, you could look at this comment on GH for inspiration https://github.com/vercel/ai/issues/7993#issuecomment-3180974654

GitHub

Recommendation for migration to v5 with mixed versions between serv...

Description We are in the process of migration to v5 but will possibly have clients still on v4 for up to 24 hours after the latest server is deployed. Since there are breaking changes to the data ...

Radu GaborOP•5d ago

Is there a difference between putting "format: 'aisdk'" in stream function and wrapping with toAISdkFormat?

_roamin_•3d ago

"format: aisdk" was kind of our first shot at converting streams, but we realized that creating utils outside of our agentic loop code was more appropriate. The "format" option might be deprecated eventually.

Gaming

Programming

Express.js stream

Did you find this page helpful?